O que são filtros?

• Filtro de Invocação de Função - este filtro é executado sempre que a KernelFunction é invocado. Permite:

  • Acesso à informação sobre a função executada e os seus argumentos
  • Gestão de exceções durante a execução da função
  • Sobreposição do resultado da função, seja antes (por exemplo, para cenários de cache) ou depois da execução (por exemplo, para cenários de IA responsável)
  • Repetir a função no caso de uma falha (por exemplo, mudar para um modelo alternativo de IA)

• Filtro de Renderização por Prompt - este filtro é ativado antes da operação de renderização do prompt, habilitando:

  • Visualizar e modificar o prompt que será enviado à IA (por exemplo, para redação de RAG ou PII)
  • Impedir a submissão rápida à IA ao sobrescrever o resultado da função (por exemplo, para Cache Semântica)

• Filtro de Invocação de Função - este filtro é executado sempre que a KernelFunction é invocado. Permite:

  • Acesso à informação sobre a função executada e os seus argumentos
  • Gestão de exceções durante a execução da função
  • Sobreposição do resultado da função, seja antes (por exemplo, para cenários de cache) ou depois da execução (por exemplo, para cenários de IA responsável)
  • Tentar novamente a função em caso de falha (por exemplo, mudar para um modelo alternativo de IA)

• Filtro de Renderização por Prompt - este filtro é ativado antes da operação de renderização do prompt, habilitando:

  • Visualizar e modificar o prompt que será enviado à IA
  • Impedir a submissão rápida à IA ao sobrescrever o resultado da função (por exemplo, para Cache Semântica)

• Filtro de Invocação de Função - este filtro é executado sempre que a KernelFunction é invocado. Permite:

  • Acesso à informação sobre a função executada e os seus argumentos
  • Gestão de exceções durante a execução da função
  • Sobreposição do resultado da função, seja antes (por exemplo, para cenários de cache) ou depois da execução (por exemplo, para cenários de IA responsável)
  • Repetir a função em caso de falha

• Filtro de Renderização por Prompt - este filtro é ativado antes da operação de renderização do prompt, habilitando:

  • Visualizar e modificar o prompt que será enviado à IA
  • Impedir a submissão imediata à IA ao sobrescrever o resultado da função

Para usar um filtro, primeiro define-o e depois adiciona-o ao Kernel objeto, seja através de injeção de dependências ou da propriedade apropriada Kernel . Ao usar injeção de dependências, a ordem dos filtros não é garantida, pelo que, com múltiplos filtros, a ordem de execução pode ser imprevisível.

Para usar um filtro, pode definir uma função com os parâmetros necessários e registá-la no Kernel objeto usando o add_filter método (passando um FilterTypes valor ou o seu equivalente em cadeia), ou usar o @kernel.filter decorador para definir e registar o filtro numa só 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 injeção de dependências:

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 filtros 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)

Também pode usar o @kernel.filter decorador para registar 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 filtros 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 injeção de dependências:

IKernelBuilder builder = Kernel.CreateBuilder();

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

Kernel kernel = builder.Build();

Adicionar filtro usando a propriedade Kernel

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

Exemplos de código

• Exemplos de filtros 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)

Também pode usar o @kernel.filter decorador para registar 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 filtros 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 injeção de dependências:

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

Tal como nos outros tipos de filtro, também pode registar 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 no Kernel Semântico podem ser invocadas de duas formas: modo streaming e modo não-streaming. No modo streaming, uma função normalmente devolve IAsyncEnumerable<T>, enquanto no modo não-streaming, devolve FunctionResult. Esta distinção afeta como os resultados podem ser anulados no filtro: no modo streaming, o novo valor do resultado da função deve ser do tipo IAsyncEnumerable<T>, enquanto no modo não-streaming, pode simplesmente ser do tipo T. Para determinar qual o tipo de resultado que deve ser devolvido, a context.IsStreaming flag está disponível no modelo de contexto do 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";
    }
}

Utilização de filtros com IChatCompletionService

Nos casos em que IChatCompletionService é usado diretamente em vez de Kernel, os filtros só serão invocados quando um Kernel objeto é passado como parâmetro para os métodos de serviço de conclusão de chat, pois os filtros são anexados à Kernel 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 no Kernel Semântico podem ser chamadas de duas formas: em modo de streaming e em modo não-streaming. No modo streaming, uma função normalmente devolve um AsyncGenerator[T] objeto onde T é um tipo de conteúdo em streaming, enquanto no modo não-streaming, devolve FunctionResult. Esta distinção afeta como os resultados podem ser anulados no filtro: no modo streaming, o novo valor do resultado da função também deve ser do tipo AsyncGenerator[T]. Para determinar qual o tipo de resultado que deve ser devolvido, a context.is_streaming flag está disponível em todos os modelos de contexto do filtro.

Portanto, para construir um filtro registador simples para a invocação de uma função de transmissão, deveria usar 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 filtros de invocação de funções de streaming

Mais informações em breve.

Ao utilizar injeção de dependência, a ordem dos filtros não é garantida. Se a ordem dos filtros for importante, recomenda-se adicionar filtros diretamente ao Kernel objeto usando as propriedades apropriadas. Esta abordagem permite que filtros sejam adicionados, removidos ou reordenados em tempo de execução.

Os filtros são executados pela ordem em que são adicionados ao objeto Kernel — seja através add_filter ou pelo @kernel.filter decorador. Como a ordem de execução pode afetar o comportamento, é importante gerir cuidadosamente a ordem dos filtros.

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

• Deteçã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 repetição com um filtro