O que são filtros?

• Filtro de Invocação de Função – esse filtro é executado sempre que um KernelFunction é invocado. Permite:

  • Acesso a informações sobre a função que está sendo executada e seus argumentos
  • Tratamento de exceções durante a execução da função
  • Substituição do resultado da função, seja antes da execução (por exemplo, para cenários de cache) ou depois (como em cenários de IA responsável).
  • Tentar novamente a função em caso de falha (por exemplo, alternar para um modelo de IA alternativo)

• Filtro de Renderização de Prompt – esse filtro é disparado antes da operação de renderização de prompt, habilitando:

  • Exibindo e modificando o prompt que será enviado para a IA (por exemplo, para RAG ou a redação de PII)
  • Prevenindo o envio de prompts à IA ao substituir o resultado da função (por exemplo, para Cache Semântico)

• Filtro de Invocação de Função – esse filtro é executado sempre que um KernelFunction é invocado. Permite:

  • Acesso a informações sobre a função que está sendo executada e seus argumentos
  • Tratamento de exceções durante a execução da função
  • Substituição do resultado da função, quer antes (por exemplo, em cenários de cache), quer após a execução (por exemplo, em cenários de IA responsável)
  • Tentar novamente a função em caso de falha (por exemplo, alternar para um modelo de IA alternativo)

• Filtro de Renderização de Prompt – esse filtro é disparado antes da operação de renderização de prompt, habilitando:

  • Exibindo e modificando o prompt que será enviado para a IA
  • Prevenindo o envio de prompt à IA substituindo o resultado da função (por exemplo, para Cache Semântico)

• Filtro de Invocação de Função – esse filtro é executado sempre que um KernelFunction é invocado. Permite:

  • Acesso a informações sobre a função que está sendo executada e seus argumentos
  • Tratamento de exceções durante a execução da função
  • Substituição do resultado da função, antes (por exemplo, cenários de cache) ou após a execução (por exemplo, cenários de IA responsáveis)
  • Repetição da função em caso de falha

• Filtro de Renderização de Prompt – esse filtro é disparado antes da operação de renderização de prompt, habilitando:

  • Exibindo e modificando o prompt que será enviado para a IA
  • Prevenção do envio de prompt à IA substituindo o resultado padrão da função

Para usar um filtro, primeiro defina-o e depois adicione-o ao objeto Kernel por meio de injeção de dependência ou da propriedade apropriada Kernel. Ao usar a injeção de dependência, a ordem dos filtros não é garantida, portanto, com vários filtros, a ordem de execução pode ser imprevisível.

Para usar um filtro, você pode definir uma função com os parâmetros necessários e registrá-la no Kernel objeto usando o add_filter método (passando um FilterTypes valor ou seu equivalente de cadeia de caracteres) ou usar o @kernel.filter decorador para definir e registrar o filtro em uma etapa.

/// <summary>
/// Example of function invocation filter to perform logging before and after function invocation.
/// </summary>
public sealed class LoggingFilter(ILogger logger) : IFunctionInvocationFilter
{
    public async Task OnFunctionInvocationAsync(FunctionInvocationContext context, Func<FunctionInvocationContext, Task> next)
    {
        logger.LogInformation("FunctionInvoking - {PluginName}.{FunctionName}", context.Function.PluginName, context.Function.Name);

        await next(context);

        logger.LogInformation("FunctionInvoked - {PluginName}.{FunctionName}", context.Function.PluginName, context.Function.Name);
    }
}

Adicionar filtro usando a injeção de dependência:

IKernelBuilder builder = Kernel.CreateBuilder();

builder.Services.AddSingleton<IFunctionInvocationFilter, LoggingFilter>();

Kernel kernel = builder.Build();

Adicionar filtro usando a propriedade Kernel

kernel.FunctionInvocationFilters.Add(new LoggingFilter(logger));

Exemplos de código

• Exemplos de filtro de invocação de função

import logging
from typing import Awaitable, Callable
from semantic_kernel.filters import FilterTypes, FunctionInvocationContext

logger = logging.getLogger(__name__)

async def logger_filter(context: FunctionInvocationContext, next: Callable[[FunctionInvocationContext], Awaitable[None]]) -> None:
    logger.info(f"FunctionInvoking - {context.function.plugin_name}.{context.function.name}")

    await next(context)

    logger.info(f"FunctionInvoked - {context.function.plugin_name}.{context.function.name}")

# Add filter to the kernel
kernel.add_filter(FilterTypes.FUNCTION_INVOCATION, logger_filter)

Você também pode usar o @kernel.filter decorador para registrar um filtro diretamente:

@kernel.filter(FilterTypes.FUNCTION_INVOCATION)
async def logger_filter(context: FunctionInvocationContext, next: Callable[[FunctionInvocationContext], Awaitable[None]]) -> None:
    logger.info(f"FunctionInvoking - {context.function.plugin_name}.{context.function.name}")

    await next(context)

    logger.info(f"FunctionInvoked - {context.function.plugin_name}.{context.function.name}")

Exemplos de código

• Exemplos de filtro de invocação de função

Mais informações em breve.

/// <summary>
/// Example of prompt render filter which overrides rendered prompt before sending it to AI.
/// </summary>
public class SafePromptFilter : IPromptRenderFilter
{
    public async Task OnPromptRenderAsync(PromptRenderContext context, Func<PromptRenderContext, Task> next)
    {
        // Example: get function information
        var functionName = context.Function.Name;

        await next(context);

        // Example: override rendered prompt before sending it to AI
        context.RenderedPrompt = "Safe prompt";
    }
}

Adicionar filtro usando a injeção de dependência:

IKernelBuilder builder = Kernel.CreateBuilder();

builder.Services.AddSingleton<IPromptRenderFilter, SafePromptFilter>();

Kernel kernel = builder.Build();

Adicionar filtro com a propriedade Kernel.

kernel.PromptRenderFilters.Add(new SafePromptFilter());

Exemplos de código

• Exemplos de filtro de renderização de prompt

from typing import Awaitable, Callable
from semantic_kernel.filters import FilterTypes, PromptRenderContext

async def safe_prompt_filter(
    context: PromptRenderContext,
    next: Callable[[PromptRenderContext], Awaitable[None]],
) -> None:
    # Example: get function information
    function_name = context.function.name

    await next(context)

    # Example: override the rendered prompt before sending it to the AI
    context.rendered_prompt = f"Safe prompt: {context.rendered_prompt or ''}"

# Register the filter on the kernel
kernel.add_filter(FilterTypes.PROMPT_RENDERING, safe_prompt_filter)

Você também pode usar o @kernel.filter decorador para registrar um filtro diretamente:

@kernel.filter(FilterTypes.PROMPT_RENDERING)
async def prompt_rendering_filter(context: PromptRenderContext, next):
    await next(context)
    context.rendered_prompt = f"You pretend to be Mosscap, but you are Papssom who is the opposite of Moscapp in every way {context.rendered_prompt or ''}"

Exemplos de código

• Exemplos de filtro de renderização de prompt

Mais informações em breve.

/// <summary>
/// Example of auto function invocation filter which terminates function calling process as soon as we have the desired result.
/// </summary>
public sealed class EarlyTerminationFilter : IAutoFunctionInvocationFilter
{
    public async Task OnAutoFunctionInvocationAsync(AutoFunctionInvocationContext context, Func<AutoFunctionInvocationContext, Task> next)
    {
        // Call the function first.
        await next(context);

        // Get a function result from context.
        var result = context.Result.GetValue<string>();

        // If the result meets the condition, terminate the process.
        // Otherwise, the function calling process will continue.
        if (result == "desired result")
        {
            context.Terminate = true;
        }
    }
}

Adicionar filtro usando a injeção de dependência:

IKernelBuilder builder = Kernel.CreateBuilder();

builder.Services.AddSingleton<IAutoFunctionInvocationFilter, EarlyTerminationFilter>();

Kernel kernel = builder.Build();

Adicionar filtro usando a propriedade Kernel

kernel.AutoFunctionInvocationFilters.Add(new EarlyTerminationFilter());

Exemplos de código

• Exemplos de filtro de invocação de função automática

from semantic_kernel.filters import FilterTypes, AutoFunctionInvocationContext

@kernel.filter(FilterTypes.AUTO_FUNCTION_INVOCATION)
async def auto_function_invocation_filter(context: AutoFunctionInvocationContext, next):
    await next(context)
    if context.function_result == "desired result":
        context.terminate = True

Assim como acontece com os outros tipos de filtro, você também pode registrar o filtro usando kernel.add_filter:

kernel.add_filter(FilterTypes.AUTO_FUNCTION_INVOCATION, auto_function_invocation_filter)

Exemplos de código

• Exemplos de filtro de invocação de função automática

Mais informações em breve.

As funções em Kernel semântico podem ser invocadas de duas maneiras: streaming e não streaming. No modo de streaming, uma função normalmente retorna IAsyncEnumerable<T>, enquanto no modo não streaming, ela retorna FunctionResult. Essa distinção afeta como os resultados podem ser substituídos no filtro: no modo de streaming, o novo valor do resultado da função deve ser do tipo IAsyncEnumerable<T>, enquanto no modo não streaming, ele pode simplesmente ser do tipo T. Para determinar qual tipo de resultado precisa ser retornado, o context.IsStreaming sinalizador está disponível no modelo de contexto de filtro.

/// <summary>Filter that can be used for both streaming and non-streaming invocation modes at the same time.</summary>
public sealed class DualModeFilter : IFunctionInvocationFilter
{
    public async Task OnFunctionInvocationAsync(FunctionInvocationContext context, Func<FunctionInvocationContext, Task> next)
    {
        // Call next filter in pipeline or actual function.
        await next(context);

        // Check which function invocation mode is used.
        if (context.IsStreaming)
        {
            // Return IAsyncEnumerable<string> result in case of streaming mode.
            var enumerable = context.Result.GetValue<IAsyncEnumerable<string>>();
            context.Result = new FunctionResult(context.Result, OverrideStreamingDataAsync(enumerable!));
        }
        else
        {
            // Return just a string result in case of non-streaming mode.
            var data = context.Result.GetValue<string>();
            context.Result = new FunctionResult(context.Result, OverrideNonStreamingData(data!));
        }
    }

    private async IAsyncEnumerable<string> OverrideStreamingDataAsync(IAsyncEnumerable<string> data)
    {
        await foreach (var item in data)
        {
            yield return $"{item} - updated from filter";
        }
    }

    private string OverrideNonStreamingData(string data)
    {
        return $"{data} - updated from filter";
    }
}

Usando filtros com IChatCompletionService

Nos casos IChatCompletionService em que é usado diretamente em vez de Kernel, os filtros só serão invocados quando um Kernel objeto for passado como um parâmetro para os métodos de serviço de conclusão de chat, pois os Kernel filtros são anexados à instância.

Kernel kernel = Kernel.CreateBuilder()
    .AddOpenAIChatCompletion("gpt-4", "api-key")
    .Build();

kernel.FunctionInvocationFilters.Add(new MyFilter());

IChatCompletionService chatCompletionService = kernel.GetRequiredService<IChatCompletionService>();

// Passing a Kernel here is required to trigger filters.
ChatMessageContent result = await chatCompletionService.GetChatMessageContentAsync(chatHistory, executionSettings, kernel);

As funções em Kernel semântico podem ser invocadas de duas maneiras: streaming e não streaming. No modo de streaming, uma função normalmente retorna um objeto AsyncGenerator[T] onde T é um tipo de conteúdo de streaming, enquanto no modo não de streaming, retorna FunctionResult. Essa distinção afeta como os resultados podem ser substituídos no filtro: no modo de streaming, o novo valor do resultado da função também deve ser do tipo AsyncGenerator[T]. Para determinar qual tipo de resultado precisa ser retornado, o context.is_streaming sinalizador está disponível em todos os modelos de contexto de filtro.

Portanto, para criar um filtro de logger simples para uma invocação de função de streaming, você usaria algo assim:

@kernel.filter(FilterTypes.FUNCTION_INVOCATION)
async def streaming_exception_handling(
    context: FunctionInvocationContext,
    next: Callable[[FunctionInvocationContext], Awaitable[None]],
):
    await next(context)
    if not context.is_streaming:
        return

    async def override_stream(stream):
        try:
            async for partial in stream:
                yield partial
        except Exception as e:
            yield [
                StreamingChatMessageContent(role=AuthorRole.ASSISTANT, content=f"Exception caught: {e}", choice_index=0)
            ]

    stream = context.result.value
    context.result = FunctionResult(function=context.result.function, value=override_stream(stream))

Exemplos de código

• Exemplos de filtro de invocação de função de streaming

Mais informações em breve.

Ao usar a injeção de dependência, a ordem dos filtros não é garantida. Se a ordem dos filtros for importante, é recomendável adicionar filtros diretamente ao Kernel objeto usando as propriedades apropriadas. Essa abordagem permite que os filtros sejam adicionados, removidos ou reordenados em runtime.

Os filtros são executados na ordem em que são adicionados ao objeto Kernel, seja por meio add_filter ou pelo @kernel.filter decorador. Como a ordem de execução pode afetar o comportamento, é importante gerenciar a ordem de filtro com cuidado.

Considere o seguinte exemplo:

def func():
    print('function')


@kernel.filter(FilterTypes.FUNCTION_INVOCATION)
async def filter1(context: FunctionInvocationContext, next):
    print('before filter 1')
    await next(context)
    print('after filter 1')

@kernel.filter(FilterTypes.FUNCTION_INVOCATION)
async def filter2(context: FunctionInvocationContext, next):
    print('before filter 2')
    await next(context)
    print('after filter 2')

Ao executar a função, a saída será:

before filter 1
before filter 2
function
after filter 2
after filter 1

• Detecção e redação de PII com filtros
• Cache semântico com filtros
• Segurança de conteúdo com filtros
• Resumo de texto e verificação da qualidade da tradução com filtros

• Cache semântico com filtros
• Tente novamente com um modelo diferente
• Lógica de tentativa novamente com um filtro