Что такое фильтры?

• Фильтр вызова функции — этот фильтр выполняется при каждом KernelFunction вызове. Это обеспечивает:

  • Доступ к сведениям о выполняемой функции и его аргументах
  • Обработка исключений во время выполнения функции
  • Переопределение результата функции либо до (например, для сценария кэширования), либо после выполнения (например, для сценариев ответственного ИИ)
  • Повторная попытка функции в случае сбоя (например, переключение на альтернативную модель ИИ)

• Фильтр отрисовки промпта — этот фильтр активируется перед операцией отрисовки промпта, обеспечивая:

  • Просмотр и изменение запроса, которое будет отправлено в ИИ (например, для редакта RAG или PII)
  • Предотвращение отправки запроса в ИИ путем переопределения результата функции (например, для семантического кэширования)

• Фильтр вызова функции — этот фильтр выполняется при каждом KernelFunction вызове. Это обеспечивает:

  • Доступ к сведениям о выполняемой функции и его аргументах
  • Обработка исключений во время выполнения функции
  • Переопределение результата функции либо до (например, для сценария кэширования), либо после выполнения (например, для сценариев ответственного ИИ)
  • Повторная попытка функции в случае сбоя (например, переключение на альтернативную модель ИИ)

• Фильтр отрисовки промпта — этот фильтр активируется перед операцией отрисовки промпта, обеспечивая:

  • Просмотр и изменение запроса, которое будет отправлено в ИИ
  • Предотвращение отправки запроса в ИИ путем переопределения результата функции (например, для семантического кэширования)

• Фильтр вызова функции — этот фильтр выполняется при каждом KernelFunction вызове. Это обеспечивает:

  • Доступ к сведениям о выполняемой функции и его аргументах
  • Обработка исключений во время выполнения функции
  • Переопределение результата функции либо до (например, для сценария кэширования), либо после выполнения (например, для сценариев ответственного ИИ)
  • Повторная попытка функции в случае сбоя

• Фильтр отрисовки промпта — этот фильтр активируется перед операцией отрисовки промпта, обеспечивая:

  • Просмотр и изменение запроса, которое будет отправлено в ИИ
  • Предотвращение отправки запроса в ИИ через переопределение результата функции

Чтобы использовать фильтр, сначала определите его, а затем добавьте его в Kernel объект через внедрение зависимостей или соответствующее Kernel свойство. При использовании внедрения зависимостей порядок фильтров не гарантируется, поэтому при использовании нескольких фильтров порядок выполнения может быть непредсказуемым.

Чтобы использовать фильтр, можно определить функцию с необходимыми параметрами и зарегистрировать ее в Kernel объекте с помощью add_filter метода (передавая FilterTypes значение или его строковый эквивалент), или использовать @kernel.filter декоратор для определения и регистрации фильтра на одном шаге.

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

Добавьте фильтр с помощью внедрения зависимостей:

IKernelBuilder builder = Kernel.CreateBuilder();

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

Kernel kernel = builder.Build();

Добавление фильтра с помощью Kernel свойства:

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

Примеры кода

• Примеры фильтров вызовов функций

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)

Вы также можете использовать декоратор @kernel.filter для непосредственной регистрации фильтра.

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

Примеры кода

• Примеры фильтров вызовов функций

Дополнительные сведения в ближайшее время.

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

Добавьте фильтр с помощью внедрения зависимостей:

IKernelBuilder builder = Kernel.CreateBuilder();

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

Kernel kernel = builder.Build();

Добавление фильтра с помощью Kernel свойства:

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

Примеры кода

• Примеры фильтра отрисовки запроса

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)

Вы также можете использовать декоратор @kernel.filter для непосредственной регистрации фильтра.

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

Примеры кода

• Примеры фильтра отрисовки запроса

Дополнительные сведения в ближайшее время.

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

Добавьте фильтр с помощью внедрения зависимостей:

IKernelBuilder builder = Kernel.CreateBuilder();

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

Kernel kernel = builder.Build();

Добавление фильтра с помощью Kernel свойства:

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

Примеры кода

• Примеры фильтров автоматического вызова функции

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

Как и в случае с другими типами фильтров, вы также можете зарегистрировать фильтр с помощью kernel.add_filter:

kernel.add_filter(FilterTypes.AUTO_FUNCTION_INVOCATION, auto_function_invocation_filter)

Примеры кода

• Примеры фильтров автоматического вызова функции

Дополнительные сведения в ближайшее время.

Функции в Semantic Kernel можно вызывать двумя способами: стриминг и нестриминг. В режиме потоковой передачи функция обычно возвращается IAsyncEnumerable<T>, в то время как она возвращается FunctionResultв режиме, отличном от потоковой передачи. Это различие влияет на то, как результаты могут быть переопределены в фильтре: в режиме стриминга новое значение результата функции должно иметь тип IAsyncEnumerable<T>, тогда как в непотоковом режиме оно может быть просто типом T. Чтобы определить, какой тип результата необходимо вернуть, context.IsStreaming флаг доступен в модели контекста фильтра.

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

Использование фильтров с IChatCompletionService

В случаях, когда используется IChatCompletionService напрямую, вместо Kernel, фильтры будут вызываться только тогда, когда объект Kernel передается в качестве параметра методам службы завершения чата, так как фильтры присоединяются к экземпляру 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);

Функции в Semantic Kernel можно вызывать двумя способами: стриминг и нестриминг. В режиме потоковой передачи функция обычно возвращает AsyncGenerator[T] объект, где T является типом потокового контента, в то время как в режиме, отличном от потоковой передачи, он возвращает.FunctionResult Это различие влияет на то, как результаты можно переопределить в фильтре: в режиме потоковой передачи новое значение результата функции также должно быть типом AsyncGenerator[T]. Чтобы определить, какой тип результата необходимо вернуть, context.is_streaming флаг доступен во всех моделях контекста фильтра.

Таким образом, чтобы создать простой фильтр средства ведения журнала для вызова функции потоковой передачи, используйте примерно следующее:

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

Примеры кода

• Примеры фильтра вызова функции потоковой передачи

Дополнительные сведения в ближайшее время.

При использовании внедрения зависимостей порядок фильтров не гарантируется. Если порядок фильтров важен, рекомендуется добавить фильтры непосредственно в Kernel объект с помощью соответствующих свойств. Такой подход позволяет добавлять, удалять или переупорядочивать фильтры во время выполнения.

Фильтры выполняются в том порядке, в котором они добавляются в объект ядра, будь то через декоратор add_filter или @kernel.filter. Так как порядок выполнения может повлиять на поведение, важно тщательно управлять порядком фильтрации.

Рассмотрим следующий пример:

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

При выполнении функции выходные данные будут:

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

• ** Обнаружение и удаление PII с фильтрами
• Семантическое кэширование с помощью фильтров
• Безопасность содержимого с фильтрами
• Сводка текста и проверка качества перевода с помощью фильтров

• Семантическое кэширование с помощью фильтров
• Повторите попытку с другой моделью
• Логика повторных попыток с фильтром