Partilhar via

Middleware do agente

O middleware no Agent Framework fornece uma maneira poderosa de intercetar, modificar e aprimorar as interações do agente em vários estágios de execução. Você pode usar middleware para implementar preocupações transversais, como registro, validação de segurança, tratamento de erros e transformação de resultados sem modificar o agente principal ou a lógica da função.

O Agent Framework pode ser personalizado usando três tipos diferentes de middleware:

  1. Middleware de execução do agente: Permite a intercetação de todas as execuções do agente, para que a entrada e a saída possam ser inspecionadas e/ou modificadas conforme necessário.
  2. Middleware de chamada de função: Permite a intercetação de todas as chamadas de função executadas pelo agente, para que a entrada e a saída possam ser inspecionadas e modificadas conforme necessário.
  3. IChatClient middleware: Permite a interceção de chamadas para uma IChatClient implementação, onde um agente está a utilizar IChatClient para chamadas de inferência, por exemplo, quando utiliza ChatClientAgent.

Todos os tipos de middleware são implementados por meio de um retorno de chamada de função e, quando várias instâncias de middleware do mesmo tipo são registradas, elas formam uma cadeia, onde se espera que cada instância de middleware chame a próxima na cadeia, por meio de um fornecido nextFunc.

Os tipos de middleware de execução e chamada de função do agente podem ser registrados em um agente, usando o construtor de agentes com um objeto de agente existente.

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

Importante

Idealmente, ambos runFunc devem runStreamingFunc ser fornecidos; ao fornecer apenas o middleware não-streaming, o agente usará tanto para invocações de streaming como não-streaming, o que bloqueará o streaming para correr em modo não-streaming para satisfazer as expectativas do middleware.

Observação

Há uma sobrecarga Use(sharedFunc: ...) adicional que permite fornecer o mesmo middleware para não-streaming e streaming sem bloquear o streaming, no entanto, o middleware partilhado não poderá intercetar ou sobrescrever a saída, tornando esta a melhor opção apenas para cenários onde só precisa de inspecionar/modificar a entrada antes de chegar ao agente.

IChatClient O middleware pode ser registrado em um IChatClient antes de ser usado com um ChatClientAgent, usando o padrão Chat Client Builder.

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 o middleware também pode ser registrado usando um método de fábrica ao construir um agente por meio de um dos métodos auxiliares em clientes 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());

Middleware de execução do agente

Aqui está um exemplo de middleware de execução do agente, que pode inspecionar e/ou modificar a entrada e a saída da execução do agente.

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

Agent Run Streaming Middleware

Aqui está um exemplo de middleware de streaming run por agentes, que pode inspecionar e/ou modificar a entrada e saída da execução de streaming agent.

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

Middleware de chamada de função

Observação

Atualmente, o middleware de chamada de função só é suportado com um AIAgent que usa Microsoft.Extensions.AI.FunctionInvokingChatClient, por exemplo. ChatClientAgent

Aqui está um exemplo de middleware de chamada de função, que pode inspecionar e/ou modificar a função que está sendo chamada, e o resultado da chamada de função.

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

É possível encerrar o loop de chamada de função com middleware de chamada de função definindo o fornecido FunctionInvocationContext.Terminate como true. Isso impedirá que o loop de chamada de função emita uma solicitação para o serviço de inferência contendo os resultados da chamada de função após a chamada de função. Se houver mais de uma função disponível para invocação durante essa iteração, ela também pode impedir que quaisquer funções restantes sejam executadas.

Advertência

Encerrar o loop de chamada de função pode resultar em seu thread ser deixado em um estado inconsistente, por exemplo, contendo conteúdo de chamada de função sem conteúdo de resultado de função. Isso pode resultar na inutilização do thread para outras execuções.

middleware IChatClient

Aqui está um exemplo de middleware do cliente de chat, que pode inspecionar e/ou modificar a entrada e saída da solicitação para o serviço de inferência que o cliente de chat fornece.

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

Observação

Para obter mais informações sobre IChatClient middleware, consulte Custom IChatClient middleware na documentação do Microsoft.Extensions.AI.

Function-Based Middleware

O middleware baseado em funções é a maneira mais simples de implementar middleware usando funções assíncronas. Essa abordagem é ideal para operações sem monitoração de estado e fornece uma solução leve para cenários comuns de middleware.

Middleware do agente

O middleware do agente interceta e modifica a execução da execução do agente. Utiliza o AgentRunContext que contém:

  • agent: O agente que está sendo invocado
  • messages: Lista de mensagens de chat na conversa
  • is_streaming: Booleano indicando se a resposta está transmitindo
  • metadata: Dicionário para armazenar dados adicionais entre middleware
  • result: A resposta do agente (pode ser modificada)
  • terminate: Sinalizar para parar o processamento adicional
  • kwargs: Argumentos de palavra-chave adicionais passados para o método de execução do agente

O next chamável continua a cadeia de middleware ou executa o agente se for o último middleware.

Aqui está um exemplo de registro simples com lógica antes e depois next de chamável:

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

Middleware de função

O middleware de função interceta chamadas de função dentro dos agentes. Utiliza o FunctionInvocationContext que contém:

  • function: A função que está sendo invocada
  • arguments: Os argumentos validados para a função
  • metadata: Dicionário para armazenar dados adicionais entre middleware
  • result: O valor de retorno da função (pode ser modificado)
  • terminate: Sinalizar para parar o processamento adicional
  • kwargs: Argumentos de palavra-chave adicionais passados para o método de chat que invocou esta função

O next chamável continua para o próximo middleware ou executa a função real.

Aqui está um exemplo de registro simples com lógica antes e depois next de chamável:

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

Middleware de bate-papo

O middleware de chat interceta solicitações de bate-papo enviadas para modelos de IA. Utiliza o ChatContext que contém:

  • chat_client: O cliente de chat que está sendo invocado
  • messages: Lista de mensagens que estão sendo enviadas para o serviço de IA
  • chat_options: As opções para o pedido de chat
  • is_streaming: Booleano indicando se esta é uma invocação de streaming
  • metadata: Dicionário para armazenar dados adicionais entre middleware
  • result: A resposta do chat da IA (pode ser modificada)
  • terminate: Sinalizar para parar o processamento adicional
  • kwargs: Argumentos de palavra-chave adicionais passados para o cliente de chat

O next chamável continua para o próximo middleware ou envia a solicitação para o serviço de IA.

Aqui está um exemplo de registro simples com lógica antes e depois next de chamável:

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

Decoradores de middleware de função

Os decoradores fornecem declaração de tipo de middleware explícita sem exigir anotações de tipo. Eles são úteis quando:

  • Você não usa anotações de tipo
  • Você precisa de uma declaração explícita de tipo de middleware
  • Você deseja evitar incompatibilidades de tipo
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 Middleware

O middleware baseado em classe é útil para operações com monitoração de estado ou lógica complexa que se beneficia de padrões de design orientados a objetos.

Classe de middleware do agente

O middleware do agente baseado em classe usa um process método que tem a mesma assinatura e o mesmo comportamento do middleware baseado em função. O process método recebe os mesmos context parâmetros e next é invocado exatamente da mesma maneira.

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

Classe de middleware de função

O middleware de função baseado em classe também usa um process método com a mesma assinatura e comportamento do middleware baseado em função. O método recebe o mesmo context e next parâmetros.

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

Classe de middleware de bate-papo

O middleware de chat baseado em classe segue o mesmo padrão com um process método que tem assinatura e comportamento idênticos ao middleware de chat baseado em função.

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

Registo de Middleware

O middleware pode ser registrado em dois níveis com escopos e comportamentos diferentes.

Agent-Level vs Run-Level Middleware

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

Principais diferenças:

  • Nível do agente: persistente em todas as execuções, configurado uma vez ao criar o agente
  • Nível de execução: Aplicado apenas a execuções específicas, permite personalização por solicitação
  • Ordem de execução: middleware do agente (externo) → Executar middleware (interno) → Execução do agente

Rescisão de middleware

O middleware pode encerrar a execução antecipadamente usando context.terminateo . Isso é útil para verificações de segurança, limitação de taxa ou falhas de validação.

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)

O que significa rescisão:

  • Definição de context.terminate = True sinais de que o processamento deve parar
  • Você pode fornecer um resultado personalizado antes de encerrar para dar feedback aos usuários
  • A execução do agente é completamente ignorada quando o middleware é encerrado

Substituição do resultado do middleware

O middleware pode substituir os resultados em cenários de não streaming e streaming, permitindo que você modifique ou substitua completamente as respostas do agente.

O tipo de context.result resultado depende se a invocação do agente é streaming ou não:

  • Não streaming: context.result contém uma AgentRunResponse resposta com a resposta completa
  • Streaming: context.result contém um gerador assíncrono AgentRunResponseUpdate que produz pedaços

Você pode usar context.is_streaming para diferenciar entre esses cenários e lidar com substituições de resultados adequadamente.

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

Essa abordagem de middleware permite que você implemente transformação de resposta sofisticada, filtragem de conteúdo, aprimoramento de resultados e personalização de streaming, mantendo a lógica do agente limpa e focada.

Próximos passos

Respostas de Antecedentes dos Agentes

  • Last updated on