Compartir a través de

Middleware del agente

El middleware de Agent Framework proporciona una manera eficaz de interceptar, modificar y mejorar las interacciones del agente en varias fases de ejecución. Puede usar middleware para implementar problemas transversales como el registro, la validación de seguridad, el control de errores y la transformación de resultados sin modificar el agente principal ni la lógica de la función.

Agent Framework se puede personalizar mediante tres tipos diferentes de middleware:

  1. Middleware de ejecución del agente: permite interceptar todas las ejecuciones del agente, de modo que la entrada y la salida se puedan inspeccionar o modificar según sea necesario.
  2. Middleware de llamada a funciones: permite la interceptación de todas las llamadas de función ejecutadas por el agente, de modo que la entrada y la salida se puedan inspeccionar y modificar según sea necesario.
  3. IChatClient middleware: permite la interceptación de llamadas a una IChatClient implementación, donde un agente usa IChatClient para las llamadas de inferencia, por ejemplo, al usar ChatClientAgent.

Todos los tipos de middleware se implementan a través de una devolución de llamada de función y cuando se registran varias instancias de middleware del mismo tipo, forman una cadena, donde se espera que cada instancia de middleware llame a la siguiente de la cadena, a través de un proporcionado nextFunc.

Los tipos de middleware de ejecución y de llamada de agente se pueden registrar en un agente mediante el generador de agentes con un objeto de agente existente.

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

Importante

Lo ideal es que tanto como runFuncrunStreamingFunc se deba proporcionar, al proporcionar solo el middleware que no sea de streaming, el agente lo usará tanto para invocaciones de streaming como para no streaming, y esto bloqueará el streaming para que se ejecute en modo que no sea de streaming para que sea suficiente con las expectativas de middleware.

Nota:

Hay una sobrecarga Use(sharedFunc: ...) adicional que le permite proporcionar el mismo middleware para no streaming y streaming sin bloquear el streaming; sin embargo, el middleware compartido no podrá interceptar ni invalidar la salida, hacer que esta sea la mejor opción solo para escenarios en los que solo necesita inspeccionar o modificar la entrada antes de llegar al agente.

IChatClient el middleware se puede registrar en un IChatClient antes de que se use con , ChatClientAgentmediante el patrón del generador de clientes de chat.

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 el middleware también se puede registrar mediante un método de fábrica al construir un agente a través de uno de los métodos auxiliares en los clientes del 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 ejecución del agente

Este es un ejemplo de middleware de ejecución del agente, que puede inspeccionar o modificar la entrada y salida de la ejecución del 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;
}

Middleware de ejecución de streaming del agente

Este es un ejemplo de middleware de streaming de ejecución de agente, que puede inspeccionar o modificar la entrada y salida de la ejecución de streaming del agente.

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 llamada a funciones

Nota:

Actualmente, la llamada a funciones de middleware solo se admite con un AIAgent que usa Microsoft.Extensions.AI.FunctionInvokingChatClient, por ejemplo, ChatClientAgent.

Este es un ejemplo de la función que llama al middleware, que puede inspeccionar o modificar la función a la que se llama y el resultado de la llamada a la función.

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

Es posible finalizar el bucle de llamada de función con middleware de llamada de función estableciendo el proporcionado FunctionInvocationContext.Terminate en true. Esto impedirá que el bucle de llamada de función emita una solicitud al servicio de inferencia que contiene los resultados de la llamada de función después de la invocación de función. Si había más de una función disponible para la invocación durante esta iteración, también puede impedir que se ejecuten las funciones restantes.

Advertencia

Finalizar el bucle de llamada de función puede provocar que el subproceso se quede en un estado incoherente, por ejemplo, que contenga contenido de llamada de función sin contenido de resultado de función. Esto puede dar lugar a que el subproceso no se pueda usar para futuras ejecuciones.

Middleware IChatClient

Este es un ejemplo de middleware de cliente de chat, que puede inspeccionar o modificar la entrada y salida de la solicitud al servicio de inferencia que proporciona el cliente de chat.

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

Nota:

Para obtener más información sobre IChatClient el middleware, consulte Middleware IChatClient personalizado en la documentación de Microsoft.Extensions.AI.

Middleware de Function-Based

El middleware basado en funciones es la manera más sencilla de implementar middleware mediante funciones asincrónicas. Este enfoque es ideal para las operaciones sin estado y proporciona una solución ligera para escenarios comunes de middleware.

Middleware del agente

El middleware del agente intercepta y modifica la ejecución de la ejecución del agente. Usa el AgentRunContext objeto que contiene:

  • agent: el agente al que se invoca.
  • messages: lista de mensajes de chat en la conversación
  • is_streaming: booleano que indica si la respuesta es streaming.
  • metadata: diccionario para almacenar datos adicionales entre middleware
  • result: la respuesta del agente (se puede modificar)
  • terminate: marca para detener el procesamiento adicional
  • kwargs: argumentos de palabra clave adicionales pasados al método run del agente

El next invocable continúa la cadena de middleware o ejecuta el agente si es el último middleware.

Este es un ejemplo de registro sencillo con lógica antes y después next de llamar a :

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 función

El middleware de función intercepta las llamadas de función dentro de los agentes. Usa el FunctionInvocationContext objeto que contiene:

  • function: la función que se invoca.
  • arguments: argumentos validados para la función
  • metadata: diccionario para almacenar datos adicionales entre middleware
  • result: el valor devuelto de la función (se puede modificar)
  • terminate: marca para detener el procesamiento adicional
  • kwargs: argumentos de palabra clave adicionales pasados al método de chat que invocó esta función

El next invocable continúa con el siguiente middleware o ejecuta la función real.

Este es un ejemplo de registro sencillo con lógica antes y después next de llamar a :

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

Chat Middleware

El middleware de chat intercepta las solicitudes de chat enviadas a los modelos de IA. Usa el ChatContext objeto que contiene:

  • chat_client: el cliente de chat que se invoca.
  • messages: lista de mensajes que se envían al servicio de IA
  • chat_options: las opciones de la solicitud de chat
  • is_streaming: booleano que indica si se trata de una invocación de streaming.
  • metadata: diccionario para almacenar datos adicionales entre middleware
  • result: la respuesta del chat de la inteligencia artificial (se puede modificar)
  • terminate: marca para detener el procesamiento adicional
  • kwargs: argumentos de palabra clave adicionales pasados al cliente de chat

El next invocable continúa con el siguiente middleware o envía la solicitud al servicio de IA.

Este es un ejemplo de registro sencillo con lógica antes y después next de llamar a :

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 función

Los decoradores proporcionan una declaración de tipo de middleware explícita sin necesidad de anotaciones de tipo. Son útiles cuando:

  • No se usan anotaciones de tipo
  • Necesita una declaración de tipo de middleware explícita.
  • Quiere evitar errores de coincidencia de tipos.
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")

Middleware de Class-Based

El middleware basado en clases es útil para operaciones con estado o lógica compleja que se beneficia de patrones de diseño orientados a objetos.

Clase middleware del agente

El middleware del agente basado en clases usa un process método que tiene la misma firma y comportamiento que el middleware basado en funciones. El process método recibe los mismos context parámetros y next y se invoca exactamente de la misma manera.

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

Function Middleware (clase)

El middleware de función basado en clases también usa un process método con la misma firma y comportamiento que el middleware basado en funciones. El método recibe los mismos context parámetros y 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")

Clase de middleware de chat

El middleware de chat basado en clases sigue el mismo patrón con un process método que tiene una firma y un comportamiento idénticos al middleware de chat basado en funciones.

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

Registro de middleware

El middleware se puede registrar en dos niveles con distintos ámbitos y comportamientos.

Middleware de Agent-Level frente a 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?")

Diferencias clave:

  • Nivel de agente: persistente en todas las ejecuciones, configuradas una vez al crear el agente
  • Nivel de ejecución: solo se aplica a ejecuciones específicas, permite la personalización por solicitud.
  • Orden de ejecución: middleware del agente (más externo) → Ejecución del middleware (más interno) → ejecución del agente

Finalización del middleware

El middleware puede finalizar la ejecución al principio mediante context.terminate. Esto resulta útil para las comprobaciones de seguridad, la limitación de velocidad o los errores de validación.

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)

Qué terminación significa:

  • Establecer context.terminate = True señales de que el procesamiento debe detenerse
  • Puede proporcionar un resultado personalizado antes de terminar para proporcionar comentarios a los usuarios.
  • La ejecución del agente se omite completamente cuando finaliza el middleware.

Invalidación de resultados de middleware

El middleware puede invalidar los resultados en escenarios que no son de streaming y streaming, lo que permite modificar o reemplazar completamente las respuestas del agente.

El tipo de resultado de context.result depende de si la invocación del agente es streaming o no de streaming:

  • No streaming: context.result contiene un AgentRunResponse objeto con la respuesta completa.
  • Streaming: context.result contiene un generador asincrónico que produce AgentRunResponseUpdate fragmentos.

Puede usar context.is_streaming para diferenciar entre estos escenarios y controlar las invalidaciones de resultados correctamente.

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

Este enfoque de middleware permite implementar una sofisticada transformación de respuesta, filtrado de contenido, mejora de resultados y personalización de streaming, al tiempo que mantiene la lógica del agente limpia y centrada.

Pasos siguientes

Respuestas en segundo plano del agente

  • Last updated on