Semantic Kernel의 필터(Filter)란

Semantic Kernel의 필터(Filter)란

필터는 함수가 언제, 어떻게 실행되는지에 대한 제어와 가시성을 제공해 보안을 강화하는 장치예요. 이를 통해 책임 있는 AI(Responsible AI) 원칙을 여러분의 작업에 적용하고, 솔루션이 엔터프라이즈 준비(enterprise ready) 상태임을 확신할 수 있어요.

예를 들어 필터는 승인(approval) 흐름이 시작되기 전에 권한을 검증하는 데 활용됩니다. 필터가 승인 제출을 하려는 사람의 권한을 확인하는 거죠. 이렇게 하면 선택된 일부 사람만 프로세스를 시작할 수 있게 됩니다.

출처: 공식 문서 — Semantic Kernel Filters

필터에는 세 가지 유형이 있어요.

  • 함수 호출 필터(Function Invocation Filter) — KernelFunction이 호출될 때마다 실행됩니다. 다음과 같은 것을 할 수 있어요.

    • 실행 중인 함수와 그 인자에 대한 정보에 접근
    • 함수 실행 중 예외 처리
    • 함수 결과 오버라이드 — 실행 전(예: 캐싱 시나리오) 또는 실행 후(예: 책임 있는 AI 시나리오)
    • 실패 시 함수 재시도(예: 대체 AI 모델로 전환)

  • 프롬프트 렌더 필터(Prompt Render Filter) — 프롬프트 렌더링 작업 전에 트리거됩니다. 다음과 같은 것을 할 수 있어요.

  • 자동 함수 호출 필터(Auto Function Invocation Filter) — 함수 호출 필터와 비슷하지만 자동 함수 호출(automatic function calling) 범위 안에서 동작하며, 채팅 기록, 실행될 모든 함수 목록, 반복 카운터 같은 추가 컨텍스트를 제공합니다. 또한 자동 함수 호출 과정을 종료할 수도 있어요(예: 계획된 세 함수 중 두 번째에서 원하는 결과를 얻은 경우).

각 필터는 함수 실행이나 프롬프트 렌더링에 관한 모든 관련 정보를 담은 context 객체를 포함합니다. 또한 각 필터는 파이프라인의 다음 필터나 함수 자체를 실행할 next 델리게이트/콜백을 갖고 있어, 함수 실행에 대한 제어를 제공합니다(예: 악성 프롬프트나 인자 처리). 같은 유형의 필터를 여러 개 등록할 수 있고, 각자 고유한 책임을 가집니다.

필터에서 next 델리게이트를 호출하는 것은 다음 등록 필터나 원본 작업(함수 호출이든 프롬프트 렌더링이든)으로 진행하는 데 필수적이에요. next를 호출하지 않으면 작업이 실행되지 않습니다.

C#에서 필터 사용하기. 필터를 쓰려면 먼저 정의한 뒤, 의존성 주입이나 적절한 Kernel 속성을 통해 Kernel 객체에 추가하세요. 의존성 주입을 쓰면 필터 순서가 보장되지 않으므로, 여러 필터가 있으면 실행 순서가 예측 불가능할 수 있어요.

Python에서 필터 사용하기. 필터를 쓰려면 필요한 파라미터를 가진 함수를 정의하고 kernel 객체에 add_filter 메서드로 등록(FilterTypes 값이나 그 문자열 동등값 전달)하거나, @kernel.filter 데코레이터로 정의·등록을 한 번에 할 수 있어요.

함수 호출 필터

이 필터는 프롬프트에서 만든 함수든 메서드에서 만든 함수든, Semantic Kernel 함수가 호출될 때마다 트리거됩니다.

C# 로깅 필터 예시.

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

Python 로깅 필터 예시.


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

프롬프트 렌더 필터

이 필터는 프롬프트에서 만든 함수가 호출되는 등 프롬프트 렌더링 작업 중에만 호출됩니다. 메서드에서 만든 Semantic Kernel 함수에 대해서는 트리거되지 않아요.

C# 예시. 아래는 AI로 보내기 전에 렌더링된 프롬프트를 오버라이드하는 프롬프트 렌더 필터입니다.

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

Python 예시.

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

자동 함수 호출 필터

이 필터는 자동 함수 호출 과정 중에만 호출됩니다. 그 과정 밖에서 함수가 호출되면 트리거되지 않아요.

C# 예시. 아래는 원하는 결과를 얻는 즉시 함수 호출 과정을 종료하는 자동 함수 호출 필터입니다.

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

Python 예시.


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>(Python에서는 AsyncGenerator[T])를 반환하고, 비스트리밍 모드에서는 FunctionResult를 반환합니다. 이 구분은 필터에서 결과를 어떻게 오버라이드할 수 있는지에 영향을 줍니다. 스트리밍 모드에서는 새 함수 결과 값이 반드시 IAsyncEnumerable<T> 타입이어야 하고, 비스트리밍 모드에서는 그냥 T 타입일 수 있어요. 어떤 결과 타입을 반환해야 하는지 결정하려면 필터 컨텍스트 모델의 context.IsStreaming(Python에서는 context.is_streaming) 플래그를 쓰면 됩니다.

C# 예시. 아래는 스트리밍·비스트리밍 호출 모드를 동시에 처리하는 필터입니다.

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

C#: IChatCompletionService와 함께 필터 사용하기. Kernel 대신 IChatCompletionService를 직접 쓸 때는, 필터가 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);

Python: 스트리밍 함수 호출용 로거 필터. 스트리밍 함수 호출을 위한 단순 로거 필터를 만들려면 이런 형태를 쓰면 됩니다.

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

필터 순서(Ordering)

C#. 의존성 주입을 쓰면 필터 순서가 보장되지 않아요. 필터 순서가 중요하다면 적절한 속성을 사용해 Kernel 객체에 직접 필터를 추가하는 걸 권장합니다. 이렇게 하면 런타임에 필터를 추가·제거·재배열할 수 있어요.

Python. 필터는 add_filter든 @kernel.filter 데코레이터든, Kernel 객체에 추가된 순서대로 실행됩니다. 실행 순서가 동작에 영향을 줄 수 있으므로 필터 순서를 신중히 관리하는 게 중요합니다.

다음 예시를 봐 보세요.

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

즉, Python 필터는 등록 순서대로 전방향(before)으로 실행되고, next를 거쳐 되돌아오면서 역순으로 후방향(after) 코드가 실행돼요. 마치 파이프라인이 쌓이고 해제되는 것과 같습니다.

더 알아보기 (Learn more)