에이전트 미들웨어

에이전트 프레임워크는 세 가지 유형의 미들웨어를 사용하여 사용자 지정할 수 있습니다.

1. 에이전트 실행 미들웨어: 필요에 따라 입력 및 출력을 검사 및/또는 수정할 수 있도록 모든 에이전트 실행을 차단할 수 있습니다.
2. 함수 호출 미들웨어: 필요에 따라 입력 및 출력을 검사하고 수정할 수 있도록 에이전트에서 실행하는 모든 함수 호출의 가로채기를 허용합니다.
3. IChatClient 미들웨어: 에이전트가 유추 호출 IChatClient 에 사용하는 IChatClient 구현(예: 사용 시)에 대한 호출을 차단할 수 있습니다 ChatClientAgent.

모든 유형의 미들웨어는 함수 콜백을 통해 구현되며, 동일한 형식의 여러 미들웨어 인스턴스가 등록되면 체인을 형성합니다. 여기서 각 미들웨어 인스턴스는 제공된 nextFunc것을 통해 체인의 다음 인스턴스를 호출해야 합니다.

에이전트 실행 및 함수 호출 미들웨어 형식은 기존 에이전트 개체와 함께 에이전트 작성기를 사용하여 에이전트에 등록할 수 있습니다.

var middlewareEnabledAgent = originalAgent
    .AsBuilder()
        .Use(runFunc: CustomAgentRunMiddleware, runStreamingFunc: CustomAgentRunStreamingMiddleware)
        .Use(CustomFunctionCallingMiddleware)
    .Build();

IChatClient미들웨어는 채팅 클라이언트 작성기 패턴을 사용하여 미들웨어를 사용하기 IChatClient전에 등록 ChatClientAgent 할 수 있습니다.

var chatClient = new AzureOpenAIClient(new Uri("https://<myresource>.openai.azure.com"), new AzureCliCredential())
    .GetChatClient(deploymentName)
    .AsIChatClient();

var middlewareEnabledChatClient = chatClient
    .AsBuilder()
        .Use(getResponseFunc: CustomChatClientMiddleware, getStreamingResponseFunc: null)
    .Build();

var agent = new ChatClientAgent(middlewareEnabledChatClient, instructions: "You are a helpful assistant.");

IChatClient SDK 클라이언트에서 도우미 메서드 중 하나를 통해 에이전트를 생성할 때 팩터리 메서드를 사용하여 미들웨어를 등록할 수도 있습니다.

var agent = new AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential())
    .GetChatClient(deploymentName)
    .CreateAIAgent("You are a helpful assistant.", clientFactory: (chatClient) => chatClient
        .AsBuilder()
            .Use(getResponseFunc: CustomChatClientMiddleware, getStreamingResponseFunc: null)
        .Build());

에이전트 실행 미들웨어

다음은 에이전트 실행에서 입력 및 출력을 검사 및/또는 수정할 수 있는 에이전트 실행 미들웨어의 예입니다.

async Task<AgentRunResponse> CustomAgentRunMiddleware(
    IEnumerable<ChatMessage> messages,
    AgentThread? thread,
    AgentRunOptions? options,
    AIAgent innerAgent,
    CancellationToken cancellationToken)
{
    Console.WriteLine(messages.Count());
    var response = await innerAgent.RunAsync(messages, thread, options, cancellationToken).ConfigureAwait(false);
    Console.WriteLine(response.Messages.Count);
    return response;
}

에이전트 스트리밍 미들웨어 실행

다음은 에이전트 스트리밍 실행에서 입력 및 출력을 검사 및/또는 수정할 수 있는 에이전트 실행 스트리밍 미들웨어의 예입니다.

async IAsyncEnumerable<AgentRunResponseUpdate> CustomAgentRunStreamingMiddleware(
    IEnumerable<ChatMessage> messages,
    AgentThread? thread,
    AgentRunOptions? options,
    AIAgent innerAgent,
    [EnumeratorCancellation] CancellationToken cancellationToken)
{
    Console.WriteLine(messages.Count());
    List<AgentRunResponseUpdate> updates = [];
    await foreach (var update in innerAgent.RunStreamingAsync(messages, thread, options, cancellationToken))
    {
        updates.Add(update);
        yield return update;
    }

    Console.WriteLine(updates.ToAgentRunResponse().Messages.Count);
}

함수 호출 미들웨어

다음은 호출되는 함수와 함수 호출의 결과를 검사 및/또는 수정할 수 있는 함수 호출 미들웨어의 예입니다.

async ValueTask<object?> CustomFunctionCallingMiddleware(
    AIAgent agent,
    FunctionInvocationContext context,
    Func<FunctionInvocationContext, CancellationToken, ValueTask<object?>> next,
    CancellationToken cancellationToken)
{
    Console.WriteLine($"Function Name: {context!.Function.Name}");
    var result = await next(context, cancellationToken);
    Console.WriteLine($"Function Call Result: {result}");

    return result;
}

제공된 FunctionInvocationContext.Terminate 값을 true로 설정하여 함수 호출 미들웨어를 사용하여 함수 호출 루프를 종료할 수 있습니다. 이렇게 하면 함수 호출 루프가 함수 호출 후 함수 호출 결과를 포함하는 유추 서비스에 요청을 실행하지 못하게 됩니다. 이 반복 중에 호출에 사용할 수 있는 함수가 두 개 이상 있는 경우 나머지 함수가 실행되지 않을 수도 있습니다.

IChatClient 미들웨어

다음은 채팅 클라이언트가 제공하는 유추 서비스에 대한 요청에 대한 입력 및 출력을 검사 및/또는 수정할 수 있는 채팅 클라이언트 미들웨어의 예입니다.

async Task<ChatResponse> CustomChatClientMiddleware(
    IEnumerable<ChatMessage> messages,
    ChatOptions? options,
    IChatClient innerChatClient,
    CancellationToken cancellationToken)
{
    Console.WriteLine(messages.Count());
    var response = await innerChatClient.GetResponseAsync(messages, options, cancellationToken);
    Console.WriteLine(response.Messages.Count);

    return response;
}

Function-Based 미들웨어

함수 기반 미들웨어는 비동기 함수를 사용하여 미들웨어를 구현하는 가장 간단한 방법입니다. 이 방법은 상태 비지정 작업에 적합하며 일반적인 미들웨어 시나리오를 위한 간단한 솔루션을 제공합니다.

에이전트 미들웨어

에이전트 미들웨어는 에이전트 실행 실행을 가로채고 수정합니다. 다음을 포함하는 항목을 AgentRunContext 사용합니다.

• agent: 호출되는 에이전트
• messages: 대화의 채팅 메시지 목록
• is_streaming: 응답이 스트리밍 중인지 여부를 나타내는 부울
• metadata: 미들웨어 간에 추가 데이터를 저장하기 위한 사전
• result: 에이전트의 응답(수정 가능)
• terminate: 추가 처리를 중지하는 플래그
• kwargs: 에이전트 실행 메서드에 전달된 추가 키워드 인수

호출 가능은 next 미들웨어 체인을 계속하거나 마지막 미들웨어인 경우 에이전트를 실행합니다.

다음은 호출 가능 전후 next 논리를 사용하는 간단한 로깅 예제입니다.

async def logging_agent_middleware(
    context: AgentRunContext,
    next: Callable[[AgentRunContext], Awaitable[None]],
) -> None:
    """Agent middleware that logs execution timing."""
    # Pre-processing: Log before agent execution
    print("[Agent] Starting execution")

    # Continue to next middleware or agent execution
    await next(context)

    # Post-processing: Log after agent execution
    print("[Agent] Execution completed")

함수 미들웨어

함수 미들웨어는 에이전트 내의 함수 호출을 차단합니다. 다음을 포함하는 항목을 FunctionInvocationContext 사용합니다.

• function: 호출되는 함수
• arguments: 함수의 유효성이 검사된 인수입니다.
• metadata: 미들웨어 간에 추가 데이터를 저장하기 위한 사전
• result: 함수의 반환 값(수정 가능)
• terminate: 추가 처리를 중지하는 플래그
• kwargs: 이 함수를 호출한 채팅 메서드에 전달된 추가 키워드 인수

호출 가능 항목은 next 다음 미들웨어로 계속 진행되거나 실제 함수를 실행합니다.

다음은 호출 가능 전후 next 논리를 사용하는 간단한 로깅 예제입니다.

async def logging_function_middleware(
    context: FunctionInvocationContext,
    next: Callable[[FunctionInvocationContext], Awaitable[None]],
) -> None:
    """Function middleware that logs function execution."""
    # Pre-processing: Log before function execution
    print(f"[Function] Calling {context.function.name}")

    # Continue to next middleware or function execution
    await next(context)

    # Post-processing: Log after function execution
    print(f"[Function] {context.function.name} completed")

채팅 미들웨어

채팅 미들웨어는 AI 모델로 전송된 채팅 요청을 차단합니다. 다음을 포함하는 항목을 ChatContext 사용합니다.

• chat_client: 호출되는 채팅 클라이언트
• messages: AI 서비스로 전송되는 메시지 목록
• chat_options: 채팅 요청에 대한 옵션
• is_streaming: 스트리밍 호출인지 여부를 나타내는 부울
• metadata: 미들웨어 간에 추가 데이터를 저장하기 위한 사전
• result: AI의 채팅 응답(수정 가능)
• terminate: 추가 처리를 중지하는 플래그
• kwargs: 채팅 클라이언트에 전달된 추가 키워드 인수

호출 가능은 next 다음 미들웨어로 계속 진행되거나 AI 서비스에 요청을 보냅니다.

다음은 호출 가능 전후 next 논리를 사용하는 간단한 로깅 예제입니다.

async def logging_chat_middleware(
    context: ChatContext,
    next: Callable[[ChatContext], Awaitable[None]],
) -> None:
    """Chat middleware that logs AI interactions."""
    # Pre-processing: Log before AI call
    print(f"[Chat] Sending {len(context.messages)} messages to AI")

    # Continue to next middleware or AI service
    await next(context)

    # Post-processing: Log after AI response
    print("[Chat] AI response received")

함수 미들웨어 데코레이터

데코레이터는 형식 주석 없이 명시적 미들웨어 형식 선언을 제공합니다. 다음과 같은 경우에 유용합니다.

• 형식 주석을 사용하지 않음
• 명시적 미들웨어 형식 선언이 필요합니다.
• 형식 불일치를 방지하려고 합니다.

from agent_framework import agent_middleware, function_middleware, chat_middleware

@agent_middleware  # Explicitly marks as agent middleware
async def simple_agent_middleware(context, next):
    """Agent middleware with decorator - types are inferred."""
    print("Before agent execution")
    await next(context)
    print("After agent execution")

@function_middleware  # Explicitly marks as function middleware
async def simple_function_middleware(context, next):
    """Function middleware with decorator - types are inferred."""
    print(f"Calling function: {context.function.name}")
    await next(context)
    print("Function call completed")

@chat_middleware  # Explicitly marks as chat middleware
async def simple_chat_middleware(context, next):
    """Chat middleware with decorator - types are inferred."""
    print(f"Processing {len(context.messages)} chat messages")
    await next(context)
    print("Chat processing completed")

Class-Based 미들웨어

클래스 기반 미들웨어는 개체 지향 디자인 패턴의 이점을 제공하는 상태 저장 작업 또는 복잡한 논리에 유용합니다.

에이전트 미들웨어 클래스

클래스 기반 에이전트 미들웨어는 함수 기반 미들웨어와 서명 및 동작이 동일한 메서드를 사용합니다 process . 메서드는 process 동일한 context 매개 변수와 next 매개 변수를 수신하고 정확히 동일한 방식으로 호출됩니다.

from agent_framework import AgentMiddleware, AgentRunContext

class LoggingAgentMiddleware(AgentMiddleware):
    """Agent middleware that logs execution."""

    async def process(
        self,
        context: AgentRunContext,
        next: Callable[[AgentRunContext], Awaitable[None]],
    ) -> None:
        # Pre-processing: Log before agent execution
        print("[Agent Class] Starting execution")

        # Continue to next middleware or agent execution
        await next(context)

        # Post-processing: Log after agent execution
        print("[Agent Class] Execution completed")

함수 미들웨어 클래스

클래스 기반 함수 미들웨어는 함수 기반 미들웨어와 동일한 서명 및 동작을 가진 메서드도 사용합니다 process . 메서드는 동일한 context 매개 변수를 받습니다 next .

from agent_framework import FunctionMiddleware, FunctionInvocationContext

class LoggingFunctionMiddleware(FunctionMiddleware):
    """Function middleware that logs function execution."""

    async def process(
        self,
        context: FunctionInvocationContext,
        next: Callable[[FunctionInvocationContext], Awaitable[None]],
    ) -> None:
        # Pre-processing: Log before function execution
        print(f"[Function Class] Calling {context.function.name}")

        # Continue to next middleware or function execution
        await next(context)

        # Post-processing: Log after function execution
        print(f"[Function Class] {context.function.name} completed")

채팅 미들웨어 클래스

클래스 기반 채팅 미들웨어는 함수 기반 채팅 미들웨어에 process 대한 서명 및 동작이 동일한 메서드를 사용하여 동일한 패턴을 따릅니다.

from agent_framework import ChatMiddleware, ChatContext

class LoggingChatMiddleware(ChatMiddleware):
    """Chat middleware that logs AI interactions."""

    async def process(
        self,
        context: ChatContext,
        next: Callable[[ChatContext], Awaitable[None]],
    ) -> None:
        # Pre-processing: Log before AI call
        print(f"[Chat Class] Sending {len(context.messages)} messages to AI")

        # Continue to next middleware or AI service
        await next(context)

        # Post-processing: Log after AI response
        print("[Chat Class] AI response received")

미들웨어 등록

미들웨어는 서로 다른 범위와 동작을 사용하여 두 수준에서 등록할 수 있습니다.

Agent-Level 및 Run-Level 미들웨어

from agent_framework.azure import AzureAIAgentClient
from azure.identity.aio import AzureCliCredential

# Agent-level middleware: Applied to ALL runs of the agent
async with AzureAIAgentClient(async_credential=credential).create_agent(
    name="WeatherAgent",
    instructions="You are a helpful weather assistant.",
    tools=get_weather,
    middleware=[
        SecurityAgentMiddleware(),  # Applies to all runs
        TimingFunctionMiddleware(),  # Applies to all runs
    ],
) as agent:

    # This run uses agent-level middleware only
    result1 = await agent.run("What's the weather in Seattle?")

    # This run uses agent-level + run-level middleware
    result2 = await agent.run(
        "What's the weather in Portland?",
        middleware=[  # Run-level middleware (this run only)
            logging_chat_middleware,
        ]
    )

    # This run uses agent-level middleware only (no run-level)
    result3 = await agent.run("What's the weather in Vancouver?")

주요 차이점:

• 에이전트 수준: 에이전트를 만들 때 한 번 구성된 모든 실행에서 영구적
• 실행 수준: 특정 실행에만 적용되며 요청당 사용자 지정 허용
• 실행 순서: 에이전트 미들웨어(가장 바깥쪽) → 미들웨어 실행(가장 안쪽) → 에이전트 실행

미들웨어 종료

미들웨어는 .를 사용하여 context.terminate실행을 조기에 종료할 수 있습니다. 보안 검사, 속도 제한 또는 유효성 검사 실패에 유용합니다.

async def blocking_middleware(
    context: AgentRunContext,
    next: Callable[[AgentRunContext], Awaitable[None]],
) -> None:
    """Middleware that blocks execution based on conditions."""
    # Check for blocked content
    last_message = context.messages[-1] if context.messages else None
    if last_message and last_message.text:
        if "blocked" in last_message.text.lower():
            print("Request blocked by middleware")
            context.terminate = True
            return

    # If no issues, continue normally
    await next(context)

종료의 의미:

• 처리가 중지되어야 하는 신호 설정 context.terminate = True
• 종료하기 전에 사용자 지정 결과를 제공하여 사용자에게 피드백을 제공할 수 있습니다.
• 미들웨어가 종료되면 에이전트 실행이 완전히 건너뜁니다.

미들웨어 결과 재정의

미들웨어는 비 스트리밍 및 스트리밍 시나리오에서 결과를 재정의하여 에이전트 응답을 수정하거나 완전히 바꿀 수 있습니다.

결과 형식 context.result 은 에이전트 호출이 스트리밍인지 아니면 비 스트리밍인지에 따라 달라집니다.

• 비 스트리밍: context.result 전체 응답이 포함된 AgentRunResponse 항목 포함
• 스트리밍: context.result 청크를 생성하는 AgentRunResponseUpdate 비동기 생성기를 포함합니다.

이러한 시나리오를 구분하고 결과 재정의를 적절하게 처리하는 데 사용할 context.is_streaming 수 있습니다.

async def weather_override_middleware(
    context: AgentRunContext,
    next: Callable[[AgentRunContext], Awaitable[None]]
) -> None:
    """Middleware that overrides weather results for both streaming and non-streaming."""

    # Execute the original agent logic
    await next(context)

    # Override results if present
    if context.result is not None:
        custom_message_parts = [
            "Weather Override: ",
            "Perfect weather everywhere today! ",
            "22°C with gentle breezes. ",
            "Great day for outdoor activities!"
        ]

        if context.is_streaming:
            # Streaming override
            async def override_stream() -> AsyncIterable[AgentRunResponseUpdate]:
                for chunk in custom_message_parts:
                    yield AgentRunResponseUpdate(contents=[TextContent(text=chunk)])

            context.result = override_stream()
        else:
            # Non-streaming override
            custom_message = "".join(custom_message_parts)
            context.result = AgentRunResponse(
                messages=[ChatMessage(role=Role.ASSISTANT, text=custom_message)]
            )

이 미들웨어 접근 방식을 사용하면 에이전트 논리를 정리하고 집중하면서 정교한 응답 변환, 콘텐츠 필터링, 결과 향상 및 스트리밍 사용자 지정을 구현할 수 있습니다.