¿Qué son los filtros?

Los filtros mejoran la seguridad al proporcionar control y visibilidad sobre cómo y cuándo se ejecutan las funciones. Esto es necesario para infundir principios de inteligencia artificial responsables en su trabajo para que se sienta seguro de que la solución está lista para la empresa.

Por ejemplo, los filtros se aprovechan para validar los permisos antes de que comience un flujo de aprobación. El filtro se ejecuta para comprobar los permisos de la persona que busca enviar una aprobación. Esto significa que solo un grupo selecto de personas podrá iniciar el proceso.

Aquí se proporciona un buen ejemplo de filtros en nuestro artículo del blog detallado sobre el Kernel semántico, que trata acerca de Filtros.   Filtros de kernel semántico

Hay tres tipos de filtros:

  • Filtro de invocación de función: este filtro se ejecuta cada vez que se invoca un KernelFunction . Permite:

    • Acceso a información sobre la función que se ejecuta y sus argumentos
    • Control de excepciones durante la ejecución de la función
    • Anulación del resultado de la función, ya sea antes (por ejemplo, para escenarios de almacenamiento en caché) o después de la ejecución (por ejemplo, para escenarios de inteligencia artificial responsable)
    • Reintento de la función en caso de error (por ejemplo, cambiar a un modelo de IA alternativo)

  • Filtro de representación de prompt: este filtro se desencadena antes de la operación de representación del prompt, lo que habilita:

    • Visualización y modificación de la solicitud que se enviará a la IA (por ejemplo, para RAG o la redacción de PII)
    • Impedir el envío de mensajes a la inteligencia artificial invalidando el resultado de la función (por ejemplo, para el almacenamiento en caché semántico)

  • Filtro de invocación de función: este filtro se ejecuta cada vez que se invoca un KernelFunction . Permite:

    • Acceso a información sobre la función que se ejecuta y sus argumentos
    • Control de excepciones durante la ejecución de la función
    • Invalidación del resultado de la función, ya sea antes (por ejemplo, para escenarios de almacenamiento en caché) o después de la ejecución (por ejemplo, para escenarios de inteligencia artificial responsables)
    • Reintento de la función en caso de error (por ejemplo, cambiar a un modelo de IA alternativo)

  • Filtro de renderizado de mensaje de aviso: este filtro se desencadena antes de la operación de renderizado del mensaje de aviso, lo que habilita:

    • Visualización y modificación del mensaje que se enviará a la inteligencia artificial
    • Impedir el envío de mensajes a la inteligencia artificial invalidando el resultado de la función (por ejemplo, para el almacenamiento en caché semántico)

  • Filtro de invocación de función: este filtro se ejecuta cada vez que se invoca un KernelFunction . Permite:

    • Acceso a información sobre la función que se ejecuta y sus argumentos
    • Control de excepciones durante la ejecución de la función
    • Invalidación del resultado de la función, ya sea antes (por ejemplo, para el escenario de almacenamiento en caché) o después de la ejecución (por ejemplo, para escenarios de inteligencia artificial responsables)
    • Reintento de la función en caso de error

  • Filtro de renderizado del indicador: este filtro se activa antes de la operación de renderizado del indicador, lo que habilita:

    • Visualización y modificación del mensaje que se enviará a la inteligencia artificial
    • Impedir el envío de mensajes a la inteligencia artificial invalidando el resultado de la función
  • Filtro de invocación de función automática: similar al filtro de invocación de función, este filtro funciona dentro del ámbito de automatic function calling, proporcionando contexto adicional, incluido el historial de chat, una lista de todas las funciones que se van a ejecutar y contadores de iteración. También permite la finalización del proceso de llamada de función automática (por ejemplo, si se obtiene un resultado deseado a partir del segundo de tres funciones planeadas).

Cada filtro incluye un context objeto que contiene toda la información relevante sobre la ejecución de la función o la representación del indicador. Además, cada filtro tiene un next delegado/devolución de llamada para ejecutar el siguiente filtro en la canalización o la función en sí, ofreciendo control sobre la ejecución de funciones (por ejemplo, en casos de solicitudes o argumentos maliciosos). Se pueden registrar varios filtros del mismo tipo, cada uno con su propia responsabilidad.

En un filtro, llamar al next delegado es esencial para continuar con el siguiente filtro registrado o la operación original (ya sea la invocación de función o la presentación del aviso). Sin llamar a next, la operación no se ejecutará.

Para usar un filtro, defina primero y, a continuación, agréguelo al Kernel objeto mediante la inserción de dependencias o la propiedad adecuada Kernel . Al usar la inserción de dependencias, no se garantiza el orden de los filtros, por lo que, con varios filtros, el orden de ejecución puede ser impredecible.

Para usar un filtro, puede definir una función con los parámetros necesarios y registrarla en el objeto mediante el Kerneladd_filter método (pasando un FilterTypes valor o su equivalente de cadena) o usar el @kernel.filter decorador para definir y registrar el filtro en un paso.

Filtro de invocación de función

Este filtro se desencadena cada vez que se invoca una función de Kernel semántico, independientemente de si es una función creada a partir de una instrucción o un método.

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

Agregue el filtro mediante la inserción de dependencias:

IKernelBuilder builder = Kernel.CreateBuilder();

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

Kernel kernel = builder.Build();

Agregue un filtro usando la propiedad Kernel.

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

Ejemplos de código


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)

También puede usar el @kernel.filter decorador para registrar un filtro directamente:


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

Ejemplos de código

Más información próximamente.

Filtro Rápido de Renderizado

Este filtro solo se invoca durante una operación de procesamiento de indicación, como cuando se llama a una función creada a partir de una indicación. No se desencadenará para funciones de Kernel semántico creadas desde métodos.

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

Agregue el filtro mediante la inserción de dependencias:

IKernelBuilder builder = Kernel.CreateBuilder();

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

Kernel kernel = builder.Build();

Agregue filtro usando la propiedad Kernel.

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

Ejemplos de código

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)

También puede usar el @kernel.filter decorador para registrar un filtro directamente:

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

Ejemplos de código

Más información próximamente.

Filtro de invocación de función automática

Este filtro solo se invoca durante un proceso de llamada de función automática. No se desencadenará cuando se invoque una función fuera de este proceso.

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

Agregue el filtro mediante la inserción de dependencias:

IKernelBuilder builder = Kernel.CreateBuilder();

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

Kernel kernel = builder.Build();

Agregue filtro mediante la propiedad Kernel:

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

Ejemplos de código


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

Al igual que con los otros tipos de filtro, también puede registrar el filtro mediante kernel.add_filter:

kernel.add_filter(FilterTypes.AUTO_FUNCTION_INVOCATION, auto_function_invocation_filter)

Ejemplos de código

Más información próximamente.

Invocación de streaming y no streaming

Las funciones de Kernel semántico se pueden invocar de dos maneras: streaming y no streaming. En el modo de streaming, una función normalmente devuelve IAsyncEnumerable<T>, mientras que en modo que no es de streaming, devuelve FunctionResult. Esta distinción afecta a cómo se pueden invalidar los resultados en el filtro: en modo de streaming, el nuevo valor de resultado de la función debe ser de tipo IAsyncEnumerable<T>, mientras que en modo que no es de streaming, simplemente puede ser de tipo T. Para determinar qué tipo de resultado se debe devolver, la context.IsStreaming marca está disponible en el 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";
    }
}

Uso de filtros con IChatCompletionService

En los casos en los que se use directamente IChatCompletionService en lugar de Kernel, los filtros solo se invocarán cuando se pase un objeto Kernel como parámetro a los métodos del servicio de finalización de chat, ya que los filtros se adjuntan a la instancia 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);

Las funciones de Kernel semántico se pueden invocar de dos maneras: streaming y no streaming. En el modo de streaming, una función normalmente devuelve un AsyncGenerator[T] objeto donde T es un tipo de contenido de streaming, mientras que en modo que no es de streaming, devuelve FunctionResult. Esta distinción afecta a cómo se pueden invalidar los resultados en el filtro: en modo de streaming, el nuevo valor de resultado de la función también debe ser de tipo AsyncGenerator[T]. Para determinar qué tipo de resultado se debe devolver, la context.is_streaming marca está disponible en todos los modelos de contexto de filtro.

Por lo tanto, para crear un filtro de registrador simple para una invocación de función de streaming, usaría algo parecido a esto:

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

Ejemplos de código

Más información próximamente.

Pedido

Al usar la inserción de dependencias, no se garantiza el orden de los filtros. Si el orden de los filtros es importante, se recomienda agregar filtros directamente al Kernel objeto mediante las propiedades adecuadas. Este enfoque permite agregar, quitar o reordenar filtros en tiempo de ejecución.

Los filtros se ejecutan en el orden en que se agregan al objeto Kernel, ya sea a través de add_filter o el decorador @kernel.filter. Dado que el orden de ejecución puede afectar al comportamiento, es importante administrar cuidadosamente el orden de filtro.

Tenga en cuenta el siguiente ejemplo:

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

Al ejecutar la función, la salida será:

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

Más ejemplos

  • Last updated on