Che cosa sono i filtri?

• Filtro chiamata funzione : questo filtro viene eseguito ogni volta che viene richiamato un oggetto KernelFunction . Consente:

  • Accesso alle informazioni sulla funzione in esecuzione e sui relativi argomenti
  • Gestione delle eccezioni durante l'esecuzione della funzione
  • Sovrascrittura del risultato della funzione, prima (ad esempio per scenari di memorizzazione nella cache) o dopo l'esecuzione (ad esempio per scenari di AI responsabile)
  • Ripetizione di tentativi della funzione in caso di errore (ad esempio, passaggio a un modello di intelligenza artificiale alternativo)

• Filtro di rendering del prompt - questo filtro viene attivato prima dell'operazione di rendering del prompt, abilitando:

  • Visualizzazione e modifica del prompt che verrà inviato all'intelligenza artificiale (ad esempio, per la redazione RAG o PII)
  • Impedire l'invio di richieste all'intelligenza artificiale eseguendo l'override del risultato della funzione (ad esempio, per la memorizzazione nella cache semantica)

• Filtro chiamata funzione : questo filtro viene eseguito ogni volta che viene richiamato un oggetto KernelFunction . Consente:

  • Accesso alle informazioni sulla funzione in esecuzione e sui relativi argomenti
  • Gestione delle eccezioni durante l'esecuzione della funzione
  • Sovrascrittura del risultato della funzione, prima (ad esempio per scenari di memorizzazione nella cache) o dopo l'esecuzione (ad esempio per scenari di AI responsabile)
  • Ripetizione di tentativi della funzione in caso di errore (ad esempio, passaggio a un modello di intelligenza artificiale alternativo)

• Filtro di rendering del prompt - questo filtro viene attivato prima dell'operazione di rendering del prompt, abilitando:

  • Visualizzazione e modifica del prompt che verrà inviato all'intelligenza artificiale
  • Impedire l'invio di richieste all'intelligenza artificiale eseguendo l'override del risultato della funzione (ad esempio, per la memorizzazione nella cache semantica)

• Filtro chiamata funzione : questo filtro viene eseguito ogni volta che viene richiamato un oggetto KernelFunction . Consente:

  • Accesso alle informazioni sulla funzione in esecuzione e sui relativi argomenti
  • Gestione delle eccezioni durante l'esecuzione della funzione
  • Sovrascrittura del risultato della funzione, prima (ad esempio per scenari di memorizzazione nella cache) o dopo l'esecuzione (ad esempio per scenari di AI responsabile)
  • Ripetizione dei tentativi della funzione in caso di errore

• Filtro di rendering del prompt - questo filtro viene attivato prima dell'operazione di rendering del prompt, abilitando:

  • Visualizzazione e modifica del prompt che verrà inviato all'intelligenza artificiale
  • Impedire l'invio di prompt all'intelligenza artificiale sovrascrivendo il risultato della funzione

Per utilizzare un filtro, prima definirlo, quindi aggiungerlo all'oggetto Kernel attraverso l'iniezione delle dipendenze o la proprietà appropriata Kernel. Quando si usa l'inserimento delle dipendenze, l'ordine dei filtri non è garantito, quindi con più filtri l'ordine di esecuzione può essere imprevedibile.

Per usare un filtro, è possibile definire una funzione con i parametri necessari e registrarla nell'oggetto Kernel usando il add_filter metodo (passando un FilterTypes valore o il relativo equivalente stringa) oppure usare l'elemento @kernel.filter Decorator per definire e registrare il filtro in un unico passaggio.

/// <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);
    }
}

Aggiungere un filtro usando l'iniezione di dipendenze:

IKernelBuilder builder = Kernel.CreateBuilder();

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

Kernel kernel = builder.Build();

Aggiungere un filtro mediante la proprietà Kernel

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

Esempi di codice

• Esempi di filtri di chiamata di funzione

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)

È anche possibile usare il decorator @kernel.filter per registrare direttamente un filtro.

@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}")

Esempi di codice

• Esempi di filtri di chiamata di funzione

Altre informazioni saranno presto disponibili.

/// <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";
    }
}

Aggiungere un filtro usando l'iniezione di dipendenze:

IKernelBuilder builder = Kernel.CreateBuilder();

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

Kernel kernel = builder.Build();

Aggiungere un filtro mediante la proprietà Kernel

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

Esempi di codice

• Esempi di filtro per il rendering del 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)

È anche possibile usare il decorator @kernel.filter per registrare direttamente un filtro.

@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 ''}"

Esempi di codice

• Esempi di filtro per il rendering del prompt

Altre informazioni saranno presto disponibili.

/// <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;
        }
    }
}

Aggiungere un filtro usando l'iniezione di dipendenze:

IKernelBuilder builder = Kernel.CreateBuilder();

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

Kernel kernel = builder.Build();

Aggiungere un filtro mediante la proprietà Kernel

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

Esempi di codice

• Esempi di filtri per l'invocazione automatica delle funzioni

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

Come per gli altri tipi di filtro, è anche possibile registrare il filtro usando kernel.add_filter:

kernel.add_filter(FilterTypes.AUTO_FUNCTION_INVOCATION, auto_function_invocation_filter)

Esempi di codice

• Esempi di filtri per l'invocazione automatica delle funzioni

Altre informazioni saranno presto disponibili.

Le funzioni in Kernel semantico possono essere richiamate in due modi: streaming e non streaming. In modalità di streaming, una funzione restituisce IAsyncEnumerable<T>in genere , mentre in modalità non di streaming restituisce FunctionResult. Questa distinzione influisce sul modo in cui è possibile eseguire l'override dei risultati nel filtro: in modalità di streaming, il nuovo valore del risultato della funzione deve essere di tipo , mentre in modalità non di streaming può essere semplicemente di tipo IAsyncEnumerable<T>T. Per determinare quale tipo di risultato deve essere restituito, il context.IsStreaming flag è disponibile nel modello di contesto del 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";
    }
}

Uso dei filtri con IChatCompletionService

Nei casi in cui IChatCompletionService viene usato direttamente anziché Kernel, i filtri verranno richiamati solo quando un Kernel oggetto viene passato come parametro ai metodi del servizio di completamento della chat, in quanto i filtri vengono collegati all'istanza Kernel .

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

Le funzioni in Kernel semantico possono essere richiamate in due modi: streaming e non streaming. In modalità di streaming, una funzione restituisce in genere un AsyncGenerator[T] oggetto in cui T è un tipo di contenuto di streaming, mentre in modalità non di streaming restituisce FunctionResult. Questa distinzione influisce sul modo in cui è possibile eseguire l'override dei risultati nel filtro: in modalità di streaming, anche il nuovo valore del risultato della funzione deve essere di tipo AsyncGenerator[T]. Per determinare quale tipo di risultato deve essere restituito, il context.is_streaming flag è disponibile in tutti i modelli di contesto del filtro.

Per creare quindi un semplice filtro logger per una chiamata di funzione di streaming, è necessario usare un codice simile al seguente:

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

Esempi di codice

• Esempi di filtri di chiamata della funzione di streaming

Altre informazioni saranno presto disponibili.

Quando si usa l'inserimento delle dipendenze, l'ordine dei filtri non è garantito. Se l'ordine dei filtri è importante, è consigliabile aggiungere filtri direttamente all'oggetto Kernel usando le proprietà appropriate. Questo approccio consente di aggiungere, rimuovere o riordinare i filtri in fase di esecuzione.

I filtri vengono eseguiti nell'ordine in cui vengono aggiunti all'oggetto Kernel, sia tramite add_filter che tramite @kernel.filter decorator. Poiché l'ordine di esecuzione può influire sul comportamento, è importante gestire attentamente l'ordine di filtro.

Considerare l'esempio riportato di seguito.

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

Quando si esegue la funzione, l'output sarà:

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

• Rilevamento e reindirizzamento delle informazioni personali con filtri
• Memorizzazione nella cache semantica con filtri
• Sicurezza dei contenuti con filtri
• Riassunto del testo e controllo qualità traduzione con filtri

• Memorizzazione nella cache semantica con filtri
• Riprovare con un modello diverso
• Logica di ripetizione dei tentativi con un filtro