Condividi tramite

Middleware agente

Il middleware in Agent Framework consente di intercettare, modificare e migliorare le interazioni degli agenti in varie fasi di esecuzione. È possibile usare il middleware per implementare problemi trasversali, ad esempio la registrazione, la convalida della sicurezza, la gestione degli errori e la trasformazione dei risultati senza modificare l'agente principale o la logica della funzione.

Agent Framework può essere personalizzato usando tre diversi tipi di middleware:

  1. Middleware esecuzione agente: consente l'intercettazione di tutte le esecuzioni dell'agente, in modo che l'input e l'output possano essere controllati e/o modificati in base alle esigenze.
  2. Middleware per le chiamate di funzione: consente l'intercettazione di tutte le chiamate di funzione eseguite dall'agente, in modo che l'input e l'output possano essere controllati e modificati in base alle esigenze.
  3. IChatClient middleware: consente l'intercettazione di chiamate a un'implementazione IChatClient , in cui un agente usa IChatClient per le chiamate di inferenza, ad esempio quando si usa ChatClientAgent.

Tutti i tipi di middleware vengono implementati tramite un callback di funzione e, quando vengono registrate più istanze middleware dello stesso tipo, formano una catena, in cui ogni istanza del middleware deve chiamare la successiva nella catena, tramite un specificato nextFunc.

L'esecuzione e la funzione dell'agente che chiamano i tipi middleware possono essere registrati in un agente usando il generatore di agenti con un oggetto agente esistente.

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

Importante

Idealmente sia runFunc che runStreamingFunc devono essere forniti, quando si fornisce solo il middleware non di streaming, l'agente lo userà sia per le chiamate di streaming che per le chiamate non di streaming e questo bloccherà l'esecuzione del flusso in modalità non di streaming per sufficienti le aspettative del middleware.

Annotazioni

Esiste un overload Use(sharedFunc: ...) aggiuntivo che consente di fornire lo stesso middleware per lo stesso flusso e non di streaming senza bloccare lo streaming, ma il middleware condiviso non sarà in grado di intercettare o eseguire l'override dell'output, rendere questa l'opzione migliore solo per gli scenari in cui è necessario esaminare o modificare l'input prima che raggiunga l'agente.

IChatClient il middleware può essere registrato in un oggetto IChatClient prima che venga usato con un ChatClientAgent, usando il modello di generatore di client di 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 il middleware può anche essere registrato usando un metodo factory quando si costruisce un agente tramite uno dei metodi helper nei client 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 esecuzione agente

Di seguito è riportato un esempio di middleware di esecuzione dell'agente, in grado di esaminare e/o modificare l'input e l'output dell'esecuzione dell'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 di esecuzione dell'agente

Di seguito è riportato un esempio di middleware di streaming eseguito dall'agente, in grado di esaminare e/o modificare l'input e l'output dell'esecuzione del flusso dell'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 per le chiamate di funzioni

Annotazioni

Il middleware per la chiamata di funzioni è attualmente supportato solo con un AIAgent oggetto che usa Microsoft.Extensions.AI.FunctionInvokingChatClient, ad esempio ChatClientAgent.

Di seguito è riportato un esempio di middleware che chiama una funzione, che può esaminare e/o modificare la funzione chiamata e il risultato dalla chiamata di funzione.

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

È possibile terminare il ciclo di chiamate di funzione con una funzione che chiama middleware impostando l'oggetto fornito FunctionInvocationContext.Terminate su true. Ciò impedirà al ciclo di chiamata della funzione di inviare una richiesta al servizio di inferenza contenente i risultati della chiamata di funzione dopo la chiamata di funzione. Se sono disponibili più funzioni per la chiamata durante questa iterazione, può anche impedire l'esecuzione di altre funzioni.

Avvertimento

L'interruzione del ciclo di chiamate di funzione può comportare che il thread venga lasciato in uno stato incoerente, ad esempio contenente il contenuto della chiamata di funzione senza contenuto dei risultati della funzione. Ciò può comportare l'inutilizzabilità del thread per ulteriori esecuzioni.

Middleware IChatClient

Di seguito è riportato un esempio di middleware client di chat, che può esaminare e/o modificare l'input e l'output per la richiesta al servizio di inferenza fornito dal client di 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;
}

Annotazioni

Per altre informazioni sul IChatClient middleware, vedere Middleware IChatClient personalizzato nella documentazione di Microsoft.Extensions.AI.

middleware Function-Based

Il middleware basato su funzioni è il modo più semplice per implementare il middleware usando funzioni asincrone. Questo approccio è ideale per le operazioni senza stato e offre una soluzione leggera per scenari middleware comuni.

Middleware agente

Il middleware agente intercetta e modifica l'esecuzione dell'agente. Usa l'oggetto AgentRunContext che contiene:

  • agent: l'agente richiamato
  • messages: elenco di messaggi di chat nella conversazione
  • is_streaming: valore booleano che indica se la risposta è in streaming
  • metadata: dizionario per l'archiviazione di dati aggiuntivi tra middleware
  • result: la risposta dell'agente (può essere modificata)
  • terminate: flag per arrestare un'ulteriore elaborazione
  • kwargs: argomenti di parole chiave aggiuntivi passati al metodo run dell'agente

Il next callable continua la catena del middleware o esegue l'agente se è l'ultimo middleware.

Ecco un semplice esempio di registrazione con logica prima e dopo next la chiamata:

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 per le funzioni

Il middleware della funzione intercetta le chiamate di funzione all'interno degli agenti. Usa l'oggetto FunctionInvocationContext che contiene:

  • function: funzione richiamata
  • arguments: argomenti convalidati per la funzione
  • metadata: dizionario per l'archiviazione di dati aggiuntivi tra middleware
  • result: valore restituito della funzione (modificabile)
  • terminate: flag per arrestare un'ulteriore elaborazione
  • kwargs: argomenti di parole chiave aggiuntivi passati al metodo di chat che ha richiamato questa funzione

Il next callable continua al middleware successivo o esegue la funzione effettiva.

Ecco un semplice esempio di registrazione con logica prima e dopo next la chiamata:

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

Il middleware di chat intercetta le richieste di chat inviate ai modelli di intelligenza artificiale. Usa l'oggetto ChatContext che contiene:

  • chat_client: il client di chat richiamato
  • messages: elenco di messaggi inviati al servizio di intelligenza artificiale
  • chat_options: opzioni per la richiesta di chat
  • is_streaming: valore booleano che indica se si tratta di una chiamata di streaming
  • metadata: dizionario per l'archiviazione di dati aggiuntivi tra middleware
  • result: la risposta di chat dall'intelligenza artificiale (può essere modificata)
  • terminate: flag per arrestare un'ulteriore elaborazione
  • kwargs: argomenti di parole chiave aggiuntivi passati al client di chat

Il next callable continua con il middleware successivo o invia la richiesta al servizio di intelligenza artificiale.

Ecco un semplice esempio di registrazione con logica prima e dopo next la chiamata:

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

Decorator middleware per le funzioni

Gli elementi Decorator forniscono una dichiarazione di tipo middleware esplicita senza richiedere annotazioni di tipo. Sono utili quando:

  • Non si usano annotazioni di tipo
  • È necessaria una dichiarazione esplicita del tipo middleware
  • Si vuole impedire la mancata corrispondenza del 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")

middleware Class-Based

Il middleware basato su classi è utile per operazioni con stato o logica complessa che trae vantaggio dai modelli di progettazione orientati agli oggetti.

Classe middleware agente

Il middleware dell'agente basato su classi usa un process metodo con la stessa firma e comportamento del middleware basato su funzioni. Il process metodo riceve gli stessi context parametri e next viene richiamato esattamente nello stesso modo.

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 middleware per le funzioni

Il middleware delle funzioni basato su classi usa anche un process metodo con la stessa firma e comportamento del middleware basato su funzioni. Il metodo riceve gli stessi context parametri e 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")

Classe middleware di chat

Il middleware di chat basato su classi segue lo stesso modello con un process metodo con firma e comportamento identici al middleware di chat basato su funzioni.

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

Registrazione del middleware

Il middleware può essere registrato a due livelli con ambiti e comportamenti diversi.

Agent-Level e 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?")

Differenze principali:

  • Livello agente: persistente in tutte le esecuzioni, configurate una sola volta durante la creazione dell'agente
  • Livello di esecuzione: applicato solo a esecuzioni specifiche, consente la personalizzazione per richiesta
  • Ordine di esecuzione: middleware agente (più esterno) → esegui middleware (più interno) →'esecuzione dell'agente

Terminazione middleware

Il middleware può terminare l'esecuzione in anticipo usando context.terminate. Ciò è utile per i controlli di sicurezza, la limitazione della frequenza o gli errori di convalida.

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)

Che cosa significa la terminazione:

  • Impostazione context.terminate = True dei segnali che l'elaborazione deve arrestare
  • È possibile fornire un risultato personalizzato prima di terminare per inviare commenti e suggerimenti agli utenti
  • L'esecuzione dell'agente viene ignorata completamente quando termina il middleware

Override dei risultati del middleware

Il middleware può eseguire l'override dei risultati in scenari non di streaming e di streaming, consentendo di modificare o sostituire completamente le risposte dell'agente.

Il tipo di risultato in context.result dipende dal fatto che la chiamata dell'agente sia in streaming o non in streaming:

  • Non in streaming: context.result contiene un oggetto AgentRunResponse con la risposta completa
  • Streaming: context.result contiene un generatore asincrono AgentRunResponseUpdate che produce blocchi

È possibile usare context.is_streaming per distinguere questi scenari e gestire gli override dei risultati in modo appropriato.

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

Questo approccio middleware consente di implementare sofisticate trasformazioni di risposta, filtro del contenuto, miglioramento dei risultati e personalizzazione dello streaming mantenendo la logica dell'agente pulita e mirata.

Passaggi successivi

Risposte in background dell'agente

  • Last updated on