Qu’est-ce que les filtres ?

• Filtre d’appel de fonction : ce filtre est exécuté chaque fois qu’un KernelFunction appel est appelé. Il autorise :

  • Accès aux informations sur la fonction en cours d’exécution et ses arguments
  • Gestion des exceptions pendant l’exécution de la fonction
  • Substitution du résultat de la fonction, soit avant (par exemple pour des scénarios de mise en cache) soit après l'exécution (par exemple pour des scénarios d'IA responsable)
  • Nouvelle tentative de la fonction en cas d’échec (par exemple, basculement vers un autre modèle IA)

• Filtre de rendu d’invite : ce filtre est déclenché avant l’opération de rendu d’invite, en activant :

  • Affichage et modification de l'invite qui sera envoyée à l’IA (par exemple, pour RAG ou pour la rédaction d'informations personnelles)
  • Empêcher l’envoi d’invites à l’IA en remplaçant le résultat de la fonction (par exemple, pour la mise en cache sémantique)

• Filtre d’appel de fonction : ce filtre est exécuté chaque fois qu’un KernelFunction appel est appelé. Il autorise :

  • Accès aux informations sur la fonction en cours d’exécution et ses arguments
  • Gestion des exceptions pendant l’exécution de la fonction
  • Substitution du résultat de la fonction, soit avant (par exemple, pour les scénarios de mise en cache) soit après l'exécution (par exemple, pour les scénarios d'IA responsable)
  • Nouvelle tentative de la fonction en cas d’échec (par exemple, basculement vers un autre modèle IA)

• Filtre de rendu d’invite : ce filtre est déclenché avant l’opération de rendu d’invite, en activant :

  • Affichage et modification de l’invite qui sera envoyée à l’IA
  • Empêcher l’envoi d’invites à l’IA en remplaçant le résultat de la fonction (par exemple, pour la mise en cache sémantique)

• Filtre d’appel de fonction : ce filtre est exécuté chaque fois qu’un KernelFunction appel est appelé. Il autorise :

  • Accès aux informations sur la fonction en cours d’exécution et ses arguments
  • Gestion des exceptions pendant l’exécution de la fonction
  • Substitution du résultat de la fonction, avant l'exécution (par exemple pour des scénarios de mise en cache) ou après l'exécution (par exemple pour les scénarios d'IA responsables)
  • Nouvelle tentative de la fonction en cas d’échec

• Filtre de rendu d’invite : ce filtre est déclenché avant l’opération de rendu d’invite, en activant :

  • Affichage et modification de l’invite qui sera envoyée à l’IA
  • Empêcher l’envoi d’invites à l’IA en remplaçant le résultat de la fonction

Pour utiliser un filtre, commencez par le définir, puis ajoutez-le à l’objet Kernel via l’injection de dépendances ou la propriété appropriée Kernel . Lorsque vous utilisez l’injection de dépendances, l’ordre des filtres n’est pas garanti. Par conséquent, avec plusieurs filtres, l’ordre d’exécution peut être imprévisible.

Pour utiliser un filtre, vous pouvez définir une fonction avec les paramètres requis et l’inscrire sur l’objet à l’aide de la Kerneladd_filter méthode (en passant une FilterTypes valeur ou son équivalent de chaîne) ou utiliser le @kernel.filter décorateur pour définir et inscrire le filtre en une seule étape.

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

Ajoutez un filtre à l’aide de l’injection de dépendances :

IKernelBuilder builder = Kernel.CreateBuilder();

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

Kernel kernel = builder.Build();

Ajoutez un filtre en utilisant la propriété Kernel.

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

Exemples de code

• Exemples de filtre d’appel de fonction

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)

Vous pouvez également utiliser le @kernel.filter décorateur pour inscrire un filtre directement :

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

Exemples de code

• Exemples de filtre d’appel de fonction

Plus d’informations prochainement.

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

Ajoutez un filtre à l’aide de l’injection de dépendances :

IKernelBuilder builder = Kernel.CreateBuilder();

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

Kernel kernel = builder.Build();

Ajoutez un filtre en utilisant la propriété Kernel :

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

Exemples de code

• Exemples de filtres de rendu de commande

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)

Vous pouvez également utiliser le @kernel.filter décorateur pour inscrire un filtre directement :

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

Exemples de code

• Exemples de filtre de rendu d'invite de commande

Plus d’informations prochainement.

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

Ajoutez un filtre à l’aide de l’injection de dépendances :

IKernelBuilder builder = Kernel.CreateBuilder();

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

Kernel kernel = builder.Build();

Ajoutez un filtre en utilisant la propriété Kernel.

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

Exemples de code

• Exemples de filtre d'invocation automatique de fonctions

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

Comme pour les autres types de filtres, vous pouvez également inscrire le filtre à l’aide de kernel.add_filter:

kernel.add_filter(FilterTypes.AUTO_FUNCTION_INVOCATION, auto_function_invocation_filter)

Exemples de code

• Exemples de filtre d'invocation de fonction automatique

Plus d’informations prochainement.

Les fonctions de Noyau sémantique peuvent être appelées de deux façons : la diffusion en continu et la non-diffusion en continu. En mode streaming, une fonction retourne IAsyncEnumerable<T>généralement , tandis qu’en mode non streaming, elle retourne FunctionResult. Cette distinction affecte la façon dont les résultats peuvent être substitués dans le filtre : en mode streaming, la nouvelle valeur de résultat de fonction doit être de type IAsyncEnumerable<T>, tandis que, en mode non streaming, elle peut simplement être de type T. Pour déterminer le type de résultat à renvoyer, l’indicateur context.IsStreaming est disponible dans le modèle de contexte de filtre.

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

Utilisation de filtres avec IChatCompletionService

Dans les cas où IChatCompletionService est utilisé directement au lieu de Kernel, les filtres ne seront appelés que lorsqu’un Kernel objet est passé en tant que paramètre aux méthodes du Kernel service d’achèvement de conversation, car les filtres sont attachés à l’instance 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);

Les fonctions de Noyau sémantique peuvent être appelées de deux façons : la diffusion en continu et la non-diffusion en continu. En mode streaming, une fonction retourne généralement un AsyncGenerator[T] objet où T est un type de contenu de diffusion en continu, tandis qu’en mode non streaming, il retourne FunctionResult. Cette distinction affecte la façon dont les résultats peuvent être substitués dans le filtre : en mode streaming, la nouvelle valeur de résultat de fonction doit également être de type AsyncGenerator[T]. Pour déterminer le type de résultat à retourner, l’indicateur context.is_streaming est disponible dans tous les modèles de contexte de filtre.

Ainsi, pour créer un filtre d’enregistreur d’événements simple pour un appel de fonction de diffusion en continu, vous devez utiliser quelque chose comme suit :

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

Exemples de code

• Exemples de filtre d’appel de fonction de streaming

Plus d’informations prochainement.

Lorsque vous utilisez l’injection de dépendances, l’ordre des filtres n’est pas garanti. Si l’ordre des filtres est important, il est recommandé d’ajouter des filtres directement à l’objet Kernel à l’aide de propriétés appropriées. Cette approche permet d’ajouter, de supprimer ou de réorganiser les filtres au moment de l’exécution.

Les filtres sont exécutés dans l’ordre dans lequel ils sont ajoutés à l’objet Kernel , qu’ils soient par le biais add_filter ou par le @kernel.filter décorateur. Étant donné que l’ordre d’exécution peut affecter le comportement, il est important de gérer soigneusement l’ordre de filtre.

Prenez en compte l’exemple suivants :

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

Lors de l’exécution de la fonction, la sortie est :

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

• Détection et rédaction des données personnelles avec des filtres
• Mise en cache sémantique avec des filtres
• Sécurité du contenu avec filtres
• Synthèse de texte et vérification de la qualité de la traduction avec des filtres

• Mise en cache sémantique avec des filtres
• Réessayer avec un autre modèle
• Logique de nouvelle tentative avec un filtre