Partager via

Intergiciel de l’agent

L’intergiciel (middleware) dans l’infrastructure agent offre un moyen puissant d’intercepter, de modifier et d’améliorer les interactions de l’agent à différentes étapes de l’exécution. Vous pouvez utiliser l’intergiciel pour implémenter des préoccupations croisées telles que la journalisation, la validation de la sécurité, la gestion des erreurs et la transformation des résultats sans modifier votre logique principale d’agent ou de fonction.

Agent Framework peut être personnalisé à l’aide de trois types d’intergiciels différents :

  1. Intergiciel d’exécution de l’agent : autorise l’interception de toutes les exécutions de l’agent, afin que l’entrée et la sortie puissent être inspectées et/ou modifiées selon les besoins.
  2. Intergiciel d’appel de fonction : autorise l’interception de tous les appels de fonction exécutés par l’agent, afin que l’entrée et la sortie puissent être inspectées et modifiées selon les besoins.
  3. IChatClient middleware : autorise l’interception des appels à une IChatClient implémentation, où un agent utilise IChatClient pour les appels d’inférence, par exemple lors de l’utilisation ChatClientAgent.

Tous les types d’intergiciels sont implémentés via un rappel de fonction, et lorsque plusieurs instances d’intergiciels du même type sont inscrites, elles forment une chaîne, où chaque instance d’intergiciel est censée appeler la suivante dans la chaîne, via une chaîne fournie nextFunc.

Les types d’intergiciels d’exécution et de fonction de l’agent peuvent être inscrits sur un agent à l’aide du générateur d’agents avec un objet agent existant.

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

Important

Dans l’idéal, les deux runFunc et runStreamingFunc doivent être fournis, quand ils fournissent uniquement le middleware non-streaming, l’agent l’utilisera à la fois pour les appels de diffusion en continu et de non-diffusion en continu, et cela bloquera l’exécution en mode hors streaming pour suffire aux attentes des middlewares.

Note

Il existe une surcharge Use(sharedFunc: ...) supplémentaire qui vous permet de fournir le même intergiciel pour le non-streaming et la diffusion en continu sans bloquer la diffusion en continu. Toutefois, le middleware partagé ne pourra pas intercepter ou remplacer la sortie, rendez cette meilleure option uniquement pour les scénarios où vous devez uniquement inspecter/modifier l’entrée avant d’atteindre l’agent.

IChatClient l’intergiciel peut être inscrit sur un IChatClient intergiciel avant d’être utilisé avec un ChatClientAgentmodèle de générateur client de conversation.

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 l’intergiciel peut également être inscrit à l’aide d’une méthode de fabrique lors de la construction d’un agent via l’une des méthodes d’assistance sur les clients du 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 d’exécution de l’agent

Voici un exemple d’intergiciel d’exécution d’agent, qui peut inspecter et/ou modifier l’entrée et la sortie de l’exécution de l’agent.

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

Intergiciel d’exécution de streaming de l’agent

Voici un exemple d’intergiciel d’exécution d’agent, qui peut inspecter et/ou modifier l’entrée et la sortie de l’exécution de streaming de l’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);
}

Intergiciel appelant la fonction

Note

Le middleware d’appel de fonction n’est actuellement pris en charge qu’avec un AIAgent intergiciel qui utilise Microsoft.Extensions.AI.FunctionInvokingChatClient, par exemple ChatClientAgent.

Voici un exemple d’intergiciel appelant une fonction, qui peut inspecter et/ou modifier la fonction appelée et le résultat de l’appel de fonction.

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

Il est possible d’arrêter la boucle d’appel de fonction avec l’intergiciel d’appel de fonction en définissant la valeur true fournie FunctionInvocationContext.Terminate . Cela empêche la boucle d’appel de fonction d’émettre une requête au service d’inférence contenant les résultats de l’appel de fonction après l’appel de fonction. S’il existe plusieurs fonctions disponibles pour l’appel pendant cette itération, il peut également empêcher l’exécution des fonctions restantes.

Avertissement

La fin de la boucle d’appel de fonction peut entraîner la gauche de votre thread dans un état incohérent, par exemple, contenant du contenu d’appel de fonction sans contenu de résultat de fonction. Cela peut entraîner l’inutilisable du thread pour d’autres exécutions.

Intergiciel IChatClient

Voici un exemple d’intergiciel client de conversation, qui peut inspecter et/ou modifier l’entrée et la sortie de la demande au service d’inférence fourni par le client de conversation.

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

Note

Pour plus d’informations sur IChatClient le middleware, consultez l’intergiciel IChatClient personnalisé dans la documentation Microsoft.Extensions.AI.

intergiciel Function-Based

L’intergiciel basé sur les fonctions est le moyen le plus simple d’implémenter un intergiciel à l’aide de fonctions asynchrones. Cette approche est idéale pour les opérations sans état et fournit une solution légère pour les scénarios d’intergiciel courants.

Intergiciel de l’agent

L’intergiciel agent intercepte et modifie l’exécution de l’exécution de l’agent. Il utilise les AgentRunContext éléments suivants :

  • agent: l’agent appelé
  • messages: liste des messages de conversation dans la conversation
  • is_streaming: booléen indiquant si la réponse est en streaming
  • metadata: Dictionnaire pour le stockage de données supplémentaires entre intergiciels
  • result: la réponse de l’agent (peut être modifiée)
  • terminate: indicateur pour arrêter le traitement ultérieur
  • kwargs: arguments de mot clé supplémentaires passés à la méthode d’exécution de l’agent

L’intergiciel next continue la chaîne d’intergiciels ou exécute l’agent s’il s’agit du dernier middleware.

Voici un exemple de journalisation simple avec une logique avant et après next l’appel :

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

Intergiciel de fonction

L’intergiciel de fonction intercepte les appels de fonction au sein des agents. Il utilise les FunctionInvocationContext éléments suivants :

  • function: fonction appelée
  • arguments: arguments validés pour la fonction
  • metadata: Dictionnaire pour le stockage de données supplémentaires entre intergiciels
  • result: valeur de retour de la fonction (peut être modifiée)
  • terminate: indicateur pour arrêter le traitement ultérieur
  • kwargs: arguments de mot clé supplémentaires passés à la méthode de conversation qui a appelé cette fonction

L’appelant next passe au middleware suivant ou exécute la fonction réelle.

Voici un exemple de journalisation simple avec une logique avant et après next l’appel :

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

Intergiciel de conversation

L’intergiciel de conversation intercepte les demandes de conversation envoyées aux modèles IA. Il utilise les ChatContext éléments suivants :

  • chat_client: le client de conversation appelé
  • messages: liste des messages envoyés au service IA
  • chat_options: options de la demande de conversation
  • is_streaming: booléen indiquant s’il s’agit d’un appel de diffusion en continu
  • metadata: Dictionnaire pour le stockage de données supplémentaires entre intergiciels
  • result: Réponse de conversation de l’IA (peut être modifiée)
  • terminate: indicateur pour arrêter le traitement ultérieur
  • kwargs: arguments de mot clé supplémentaires passés au client de conversation

L’appelant next continue à l’intergiciel suivant ou envoie la requête au service IA.

Voici un exemple de journalisation simple avec une logique avant et après next l’appel :

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

Décorateurs d’intergiciels de fonction

Les décorateurs fournissent une déclaration de type middleware explicite sans nécessiter d’annotations de type. Ils sont utiles quand :

  • Vous n’utilisez pas d’annotations de type
  • Vous avez besoin d’une déclaration de type middleware explicite
  • Vous souhaitez empêcher les incompatibilités de type
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")

intergiciel Class-Based

L’intergiciel basé sur des classes est utile pour les opérations avec état ou une logique complexe qui tire parti des modèles de conception orienté objet.

Classe Middleware agent

L’intergiciel d’agent basé sur une classe utilise une process méthode qui a la même signature et le même comportement que l’intergiciel basé sur les fonctions. La process méthode reçoit les mêmes paramètres et context est appelée de la même next façon.

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 de fonction

L’intergiciel de fonction basé sur des classes utilise également une process méthode avec la même signature et le même comportement que l’intergiciel basé sur les fonctions. La méthode reçoit les mêmes context paramètres.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")

Chat Middleware, classe

Le middleware de conversation basé sur une classe suit le même modèle avec une méthode qui a une signature et un process comportement identiques à l’intergiciel de conversation basé sur les fonctions.

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

Inscription d’intergiciel

Les intergiciels peuvent être inscrits à deux niveaux avec des étendues et des comportements différents.

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

Principales différences :

  • Niveau de l’agent : persistant sur toutes les exécutions, configuré une fois lors de la création de l’agent
  • Niveau d’exécution : appliqué uniquement aux exécutions spécifiques, autorise la personnalisation par requête
  • Ordre d’exécution : Intergiciel agent (externe) → Exécuter l’intergiciel (le plus interne) → Exécution de l’agent

Arrêt du middleware

L’intergiciel peut arrêter l’exécution tôt à l’aide context.terminatede . Cela est utile pour les vérifications de sécurité, la limitation du débit ou les échecs de validation.

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)

Ce que signifie l’arrêt :

  • Définition des context.terminate = True signaux que le traitement doit arrêter
  • Vous pouvez fournir un résultat personnalisé avant de terminer pour donner des commentaires aux utilisateurs
  • L’exécution de l’agent est complètement ignorée lorsque l’intergiciel se termine

Remplacement des résultats du middleware

Le middleware peut remplacer les résultats dans les scénarios de diffusion en continu et de diffusion en continu, ce qui vous permet de modifier ou de remplacer complètement les réponses de l’agent.

Le type context.result de résultat varie selon que l’appel de l’agent est en streaming ou non en streaming :

  • Non-streaming : context.result contient une AgentRunResponse réponse complète
  • Streaming : context.result contient un générateur asynchrone qui génère AgentRunResponseUpdate des blocs

Vous pouvez utiliser context.is_streaming pour différencier ces scénarios et gérer les remplacements de résultats de manière appropriée.

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

Cette approche d’intergiciel vous permet d’implémenter une transformation de réponse sophistiquée, un filtrage de contenu, une amélioration des résultats et une personnalisation du streaming tout en conservant la logique de votre agent propre et prioritaire.

Étapes suivantes

Réponses en arrière-plan de l’agent

  • Last updated on