Semantic Kernel 필터

Semantic Kernel 필터 (가드레일)

필터(Filters) 는 함수가 언제, 어떻게 실행되는지에 대한 제어와 가시성을 제공해서 보안을 강화해 주는 Semantic Kernel의 기능이에요. 책임 있는 AI(responsible AI) 원칙을 우리 작업에 자연스럽게 녹여, 만든 솔루션이 엔터프라이즈에 준비됐다는 확신을 갖게 해 주는 역할이죠.

예를 들어 승인(approval) 흐름이 시작되기 전에 권한을 검증할 때 필터를 활용해요. 승인을 제출하려는 사람의 권한을 필터가 먼저 확인해서, 선택된 사람만 프로세스를 시작할 수 있게 하는 거예요. 필터에 대한 자세한 내용은 Microsoft의 Semantic Kernel 필터 블로그 포스트를 추천해요.

출처: 공식문서

필터의 세 가지 유형

Function Invocation Filter — KernelFunction이 호출될 때마다 실행되는 필터예요. 다음을 할 수 있습니다.

  • 실행 중인 함수와 그 인자에 대한 정보에 접근
  • 함수 실행 중 발생하는 예외 처리
  • 함수 결과를 실행 전(예: 캐싱 시나리오) 또는 실행 후(예: responsible AI 시나리오)에 재정의
  • 실패 시 함수 재시도 (예: 대체 AI 모델로 전환)

Prompt Render Filter — 프롬프트 렌더링 연산 전에 트리거되는 필터예요. 다음을 할 수 있습니다.

Auto Function Invocation Filter — function invocation 필터와 비슷하지만, 자동 함수 호출(automatic function calling) 범위 안에서 동작해요. 채팅 히스토리, 실행될 함수 전체 목록, 반복 카운터 같은 추가 컨텍스트를 제공하고, (예를 들어 계획된 세 함수 중 두 번째에서 원하는 결과를 얻으면) 자동 함수 호출 프로세스를 종료시킬 수도 있어요.

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

필터에서 next를 호출하는 건 필수예요. 그래야 다음 필터나 원래 연산(함수 호출이든 프롬프트 렌더링이든)으로 진행되거든요. next를 호출하지 않으면 그 연산은 실행되지 않아요.

  • C# 사용법: 필터를 정의한 뒤 의존성 주입 또는 적절한 Kernel 프로퍼티로 Kernel 객체에 추가해요. 의존성 주입을 쓰면 필터 순서가 보장되지 않으니, 필터가 여럿이면 실행 순서가 예측 불가할 수 있어요.
  • Python 사용법: 필요한 파라미터의 함수를 정의해 Kernel 객체에 add_filter 메서드(또는 FilterTypes 값·문자열)로 등록하거나, @kernel.filter 데코레이터로 정의·등록을 한 번에 끝낼 수 있어요.

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

Prompt Render Filter

이 필터는 프롬프트 렌더링 연산 중에만 호출돼요. 예를 들어 프롬프트로 만든 함수가 호출될 때인데, 메서드로 만든 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.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)

Auto Function Invocation Filter

이 필터는 자동 함수 호출 프로세스 중에만 호출돼요. 그 프로세스 밖에서 함수를 호출하면 트리거되지 않아요.

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

builder.Services.AddSingleton<IAutoFunctionInvocationFilter, 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

스트리밍 vs 비스트리밍 호출

Semantic Kernel의 함수는 스트리밍과 비스트리밍 두 방식으로 호출돼요. 스트리밍 모드에서 함수는 보통 IAsyncEnumerable<T>를 반환하고, 비스트리밍 모드에서는 FunctionResult를 반환하죠. 이 차이는 필터에서 결과를 재정의하는 방식에 영향을 줘요. 스트리밍 모드에선 새 함수 결과 값이 IAsyncEnumerable<T> 타입이어야 하고, 비스트리밍 모드에서는 그냥 T 타입이면 돼요. 어떤 결과 타입을 돌려줘야 하는지 판단하려면 필터 컨텍스트 모델의 context.IsStreaming 플래그를 쓰면 됩니다.

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

Python — 스트리밍 함수 호출용 로거 필터는 이렇게 만들 수 있어요. 스트리밍 모드에선 새 함수 결과 값이 AsyncGenerator[T] 타입이어야 해요.

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

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

필터 순서

  • C#: 의존성 주입을 쓰면 필터 순서가 보장되지 않아요. 순서가 중요하면 적절한 프로퍼티로 Kernel 객체에 직접 필터를 추가하는 걸 권장해요. 이렇게 하면 런타임에 필터를 추가·제거·재정렬할 수 있어요.
  • Python: 필터는 add_filter든 @kernel.filter 데코레이터든 Kernel 객체에 추가된 순서대로 실행돼요. 실행 순서가 동작에 영향을 줄 수 있으니 필터 순서 관리가 중요하죠.

Python에서 두 필터를 정의한 예시를 보면, 실행 결과로 순서를 확인할 수 있어요.

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

함수를 실행하면 출력은 이렇게 됩니다. 필터 1 → 필터 2 순서로 먼저 실행되고(before), 함수 본문이 실행된 뒤엔 역순으로 감싸진 것처럼(after 2 → after 1) 나와요.

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

더 많은 예시

더 알아보기 (Learn more)