Freigeben über

Middleware für Agenten

Middleware im Agent Framework bietet eine leistungsstarke Möglichkeit, Agentinteraktionen in verschiedenen Phasen der Ausführung abzufangen, zu ändern und zu verbessern. Sie können Middleware verwenden, um übergreifende Bedenken wie Protokollierung, Sicherheitsüberprüfung, Fehlerbehandlung und Ergebnistransformation zu implementieren, ohne Ihre Kern-Agent- oder Funktionslogik zu ändern.

Das Agent Framework kann mit drei verschiedenen Middlewaretypen angepasst werden:

  1. Agent Run Middleware: Ermöglicht das Abfangen aller Agentausführungen, sodass Eingabe und Ausgabe bei Bedarf überprüft und/oder geändert werden können.
  2. Middleware für Funktionsaufrufe: Ermöglicht das Abfangen aller Funktionsaufrufe, die vom Agent ausgeführt werden, sodass Eingabe und Ausgabe bei Bedarf überprüft und geändert werden können.
  3. IChatClient Middleware: Ermöglicht das Abfangen von Aufrufen an eine IChatClient Implementierung, bei der ein Agent IChatClient für Ableitungsaufrufe verwendet wird, z. B. bei verwendung ChatClientAgent.

Alle Middlewaretypen werden über einen Funktionsrückruf implementiert, und wenn mehrere Middlewareinstanzen desselben Typs registriert werden, bilden sie eine Kette, in der jede Middleware-Instanz die nächste in der Kette über eine bereitgestellte nextFuncInstanz aufrufen soll.

Agent run and function calling middleware types can be registered on an agent, by using the agent builder with an existing agent object.

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

Von Bedeutung

Idealerweise sollten sowohl als auch runFuncrunStreamingFunc bereitgestellt werden, wenn nur die Nicht-Streaming-Middleware bereitgestellt wird, verwendet der Agent es sowohl für Streaming- als auch nicht-Streaming-Aufrufe, und dadurch wird das Streaming blockiert, um im Nicht-Streaming-Modus ausgeführt zu werden, um die Erwartungen der Middleware zu erfüllen.

Hinweis

Es gibt eine zusätzliche Überladung Use(sharedFunc: ...) , mit der Sie die gleiche Middleware für Nicht-Streaming und Streaming bereitstellen können, ohne das Streaming zu blockieren. Die freigegebene Middleware kann die Ausgabe jedoch nicht abfangen oder außer Kraft setzen. Stellen Sie dies nur für Szenarien, in denen Sie die Eingabe nur prüfen/ändern müssen, bevor sie den Agent erreicht.

IChatClientMiddleware kann vor der Verwendung mit einem IChatClientClient-Generator-Muster für Chats registriert ChatClientAgent werden.

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 Middleware kann auch mithilfe einer Factorymethode registriert werden, wenn ein Agent über eine der Hilfsmethoden auf SDK-Clients erstellt wird.

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

Agent Run Middleware

Nachfolgend finden Sie ein Beispiel für die Middleware des Agents, die die Eingabe und/oder Ausgabe des Agents überprüfen und/oder ändern kann.

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

Hier sehen Sie ein Beispiel für die Agentausführung von Streaming-Middleware, die die Eingabe und/oder Ausgabe der Agent-Streamingausführung prüfen und/oder ändern kann.

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 für Funktionsaufrufe

Hinweis

Die Middleware für Funktionsaufrufe wird derzeit nur mit einer AIAgent Funktion unterstützt, die Microsoft.Extensions.AI.FunctionInvokingChatClientz. B. verwendet wird. ChatClientAgent

Nachfolgend finden Sie ein Beispiel für die Middleware zum Aufrufen von Funktionen, die die aufgerufene Funktion prüfen und/oder ändern können, und das Ergebnis des Funktionsaufrufs.

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 ist möglich, die Funktionsaufrufschleife mit Funktionsaufrufs-Middleware zu beenden, indem sie die angegebene FunctionInvocationContext.Terminate Aufschrift auf "true" festlegen. Dadurch wird verhindert, dass die Funktionsaufrufschleife nach dem Aufruf der Funktion eine Anforderung an den Rückschlussdienst ausgibt, der die Ergebnisse des Funktionsaufrufs enthält. Wenn während dieser Iteration mehrere Funktionen zum Aufrufen verfügbar waren, kann es auch verhindern, dass verbleibende Funktionen ausgeführt werden.

Warnung

Das Beenden der Funktionsaufrufschleife kann dazu führen, dass der Thread in einem inkonsistenten Zustand verbleibt, z. B. mit Funktionsaufrufinhalten ohne Funktionsergebnisinhalt. Dies kann dazu führen, dass der Thread für weitere Ausführungen nicht mehr verwendet werden kann.

IChatClient-Middleware

Hier ist ein Beispiel für Chatclient-Middleware, die die Eingabe und/oder Ausgabe für die Anforderung an den vom Chatclient bereitgestellten Ableitungsdienst überprüfen und/oder ändern kann.

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

Hinweis

Weitere Informationen zu IChatClient Middleware finden Sie in der Microsoft.Extensions.AI-Dokumentation unter Custom IChatClient middleware .

Function-Based Middleware

Funktionsbasierte Middleware ist die einfachste Möglichkeit, Middleware mithilfe asynchroner Funktionen zu implementieren. Dieser Ansatz eignet sich ideal für zustandslose Vorgänge und bietet eine einfache Lösung für gängige Middleware-Szenarien.

Middleware für Agenten

Agent Middleware fängt ab und ändert die Ausführung des Agents. Es verwendet folgendes AgentRunContext :

  • agent: Der aufgerufene Agent
  • messages: Liste der Chatnachrichten in der Unterhaltung
  • is_streaming: Boolescher Wert, der angibt, ob die Antwort streamingt
  • metadata: Wörterbuch zum Speichern zusätzlicher Daten zwischen Middleware
  • result: Die Antwort des Agents (kann geändert werden)
  • terminate: Kennzeichnung, um die weitere Verarbeitung zu beenden
  • kwargs: Zusätzliche Schlüsselwortargumente, die an die Agent-Run-Methode übergeben werden

Der next Aufrufbare setzt die Middlewarekette fort oder führt den Agent aus, wenn es sich um die letzte Middleware handelt.

Nachfolgend finden Sie ein einfaches Protokollierungsbeispiel mit Logik vor und nach next dem Aufruf:

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

Funktions-Middleware

Funktions-Middleware fängt Funktionsaufrufe innerhalb von Agents ab. Es verwendet folgendes FunctionInvocationContext :

  • function: Die aufgerufene Funktion
  • arguments: Die überprüften Argumente für die Funktion
  • metadata: Wörterbuch zum Speichern zusätzlicher Daten zwischen Middleware
  • result: Der Rückgabewert der Funktion (kann geändert werden)
  • terminate: Kennzeichnung, um die weitere Verarbeitung zu beenden
  • kwargs: Zusätzliche Schlüsselwortargumente, die an die Chatmethode übergeben werden, die diese Funktion aufgerufen hat

Der next Aufrufbare setzt die nächste Middleware fort oder führt die eigentliche Funktion aus.

Nachfolgend finden Sie ein einfaches Protokollierungsbeispiel mit Logik vor und nach next dem Aufruf:

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

Chat-Middleware fängt Chatanfragen ab, die an KI-Modelle gesendet werden. Es verwendet folgendes ChatContext :

  • chat_client: Der aufgerufene Chatclient
  • messages: Liste der Nachrichten, die an den KI-Dienst gesendet werden
  • chat_options: Die Optionen für die Chatanfrage
  • is_streaming: Boolescher Wert, der angibt, ob es sich um einen Streamingaufruf handelt
  • metadata: Wörterbuch zum Speichern zusätzlicher Daten zwischen Middleware
  • result: Die Chatantwort von der KI (kann geändert werden)
  • terminate: Kennzeichnung, um die weitere Verarbeitung zu beenden
  • kwargs: Zusätzliche Schlüsselwortargumente, die an den Chatclient übergeben werden

Der next Aufrufbare wird weiterhin an die nächste Middleware gesendet oder die Anforderung an den KI-Dienst gesendet.

Nachfolgend finden Sie ein einfaches Protokollierungsbeispiel mit Logik vor und nach next dem Aufruf:

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

Middleware-Funktionsdekortoren

Dekorateure stellen explizite Middlewaretypdeklaration bereit, ohne Dass Typanmerkungen erforderlich sind. Sie sind hilfreich, wenn:

  • Sie verwenden keine Typanmerkungen.
  • Sie benötigen explizite Middleware-Typdeklaration
  • Sie möchten Typenkonflikte verhindern
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

Klassenbasierte Middleware ist nützlich für zustandsbehaftete Vorgänge oder komplexe Logik, die von objektorientierten Entwurfsmustern profitiert.

Agent Middleware-Klasse

Klassenbasierte Agent-Middleware verwendet eine process Methode, die dieselbe Signatur und dasselbe Verhalten wie funktionsbasierte Middleware aufweist. Die process Methode empfängt die gleichen context Parameter und next wird auf genau die gleiche Weise aufgerufen.

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

Middleware-Klasse für Funktionen

Klassenbasierte Funktions-Middleware verwendet auch eine process Methode mit der gleichen Signatur und demselben Verhalten wie funktionsbasierte Middleware. Die Methode empfängt dieselben context Parameter und next Parameter.

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

Middleware-Klasse für Chats

Klassenbasierte Chat-Middleware folgt demselben Muster mit einer process Methode, die identische Signatur und verhalten wie funktionsbasierte Chat-Middleware aufweist.

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

Middleware-Registrierung

Middleware kann auf zwei Ebenen mit unterschiedlichen Bereichen und Verhaltensweisen registriert werden.

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

Wichtige Unterschiede:

  • Agent-Ebene: Persistent für alle Ausgeführten, einmal beim Erstellen des Agents konfiguriert
  • Ausführungsebene: Nur auf bestimmte Ausführungen angewendet, ermöglicht Anpassungen pro Anforderung
  • Ausführungsreihenfolge: Agent Middleware (äußerst) → Middleware ausführen (innerstes) → Agentausführung

Middleware-Beendigung

Middleware kann die Ausführung frühzeitig beenden.context.terminate Dies ist nützlich für Sicherheitsüberprüfungen, Ratenbeschränkungen oder Überprüfungsfehler.

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)

Was eine Kündigung bedeutet:

  • Festlegen von context.terminate = True Signalen, dass die Verarbeitung beendet werden soll
  • Sie können ein benutzerdefiniertes Ergebnis vor dem Beenden bereitstellen, um Benutzern Feedback zu geben.
  • Die Agentausführung wird vollständig übersprungen, wenn Middleware beendet wird.

Außerkraftsetzung von Middleware-Ergebnissen

Middleware kann Ergebnisse sowohl in Nicht-Streaming- als auch Streamingszenarien außer Kraft setzen, sodass Sie Agent-Antworten ändern oder vollständig ersetzen können.

Der Ergebnistyp context.result hängt davon ab, ob der Agentaufruf Streaming oder Nicht-Streaming ist:

  • Nicht-Streaming: context.result enthält eine AgentRunResponse mit der vollständigen Antwort
  • Streaming: context.result enthält einen asynchronen AgentRunResponseUpdate Generator, der Blöcke liefert

Sie können verwenden context.is_streaming , um zwischen diesen Szenarien zu unterscheiden und Ergebnisüberschreibungen entsprechend zu behandeln.

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

Mit diesem Middleware-Ansatz können Sie anspruchsvolle Antworttransformationen, Inhaltsfilterung, Ergebnisverbesserungen und Streaminganpassungen implementieren und gleichzeitig die Agentlogik sauber und fokussiert halten.

Nächste Schritte

Agent-Hintergrundantworten

  • Last updated on