Middleware agenta

Middleware v architektuře agentů poskytuje účinný způsob, jak zachytit, upravit a vylepšit interakce agentů v různých fázích provádění. Middleware můžete použít k implementaci průřezových aspektů, jako je protokolování, ověřování zabezpečení, zpracování chyb a transformace výsledků beze změny základního agenta nebo logiky funkce.

Rozhraní agenta lze přizpůsobit pomocí tří různých typů middlewaru:

Spuštění middlewaru agenta: Umožňuje zachytit všechna spuštění agenta, aby bylo možné v případě potřeby zkontrolovat vstup a výstup nebo upravit.
Middleware volání funkce: Umožňuje zachytit všechna volání funkcí spuštěná agentem, aby bylo možné podle potřeby kontrolovat a upravovat vstup a výstup.
IChatClient middleware: Umožňuje zachycení volání do IChatClient implementace, kde agent používá IChatClient pro odvozování volání, například při použití ChatClientAgent.

Všechny typy middlewaru se implementují prostřednictvím zpětného volání funkce a když je registrováno více instancí middlewaru stejného typu, tvoří řetěz, kde se očekává, že každá instance middlewaru bude volat další v řetězci prostřednictvím poskytnuté nextFunc.

Spuštění agenta a volání typů middlewaru lze v agentu zaregistrovat pomocí tvůrce agentů s existujícím objektem agenta.

var middlewareEnabledAgent = originalAgent
    .AsBuilder()
        .Use(CustomAgentRunMiddleware)
        .Use(CustomFunctionCallingMiddleware)
    .Build();

IChatClient middleware lze zaregistrovat na před IChatClient použitím s nástrojem ChatClientAgent, pomocí vzoru tvůrce chatovacího klienta.

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 lze také zaregistrovat pomocí metody továrny při vytváření agenta prostřednictvím jedné z pomocných metod na klientech SADY 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());

Spuštění middlewaru agenta

Tady je příklad spuštění middlewaru agenta, který může kontrolovat a/nebo upravovat vstup a výstup ze spuštění agenta.

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 volání funkcí

Poznámka:

Middleware pro volání funkcí se v současné době podporuje pouze v případě AIAgent , že se používá Microsoft.Extensions.AI.FunctionInvokingChatClient, například ChatClientAgent.

Tady je příklad middlewaru volání funkce, který může zkontrolovat a/nebo upravit volanou funkci a výsledek volání funkce.

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

Smyčku volání funkce je možné ukončit pomocí middlewaru volání funkce nastavením zadané FunctionInvocationContext.Terminate hodnoty true. Tím zabráníte volání funkce ve volání požadavku na službu odvozování obsahující výsledky volání funkce po vyvolání funkce. Pokud bylo během této iterace k dispozici více než jedna funkce pro vyvolání, může také zabránit spuštění všech zbývajících funkcí.

Výstraha

Ukončení smyčky volání funkce může vést k tomu, že vlákno zůstane v nekonzistentním stavu, například obsahující obsah volání funkce bez obsahu výsledku funkce. To může vést k nepoužitelnému vláknu pro další spuštění.

Middleware IChatClient

Tady je příklad middlewaru chatovacího klienta, který může zkontrolovat nebo upravit vstup a výstup požadavku na službu odvozování, kterou chatovací klient poskytuje.

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

Poznámka:

Další informace o IChatClient middlewaru najdete v tématu Vlastní middleware IChatClient v Microsoft.Extensions.AI dokumentaci.

middleware Function-Based

Middleware založený na funkcích je nejjednodušší způsob implementace middlewaru pomocí asynchronních funkcí. Tento přístup je ideální pro bezstavové operace a poskytuje jednoduché řešení pro běžné scénáře middlewaru.

Middleware agenta

Zachytávání middlewaru agenta a úpravy spuštění agenta. Používá následující AgentRunContext funkce:

agent: Vyvolání agenta
messages: Seznam chatovacích zpráv v konverzaci
is_streaming: Logická hodnota označující, jestli je odpověď streamovaná
metadata: Slovník pro ukládání dalších dat mezi middlewarem
result: Odpověď agenta (lze upravit)
terminate: Příznak pro zastavení dalšího zpracování
kwargs: Další argumenty klíčových slov předané metodě spuštění agenta

next Volatelný pokračuje v řetězu middlewaru nebo provádí agenta, pokud se jedná o poslední middleware.

Tady je jednoduchý příklad protokolování s logikou před a po next volání:

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 funkce

Middleware funkcí zachycuje volání funkcí v rámci agentů. Používá následující FunctionInvocationContext funkce:

function: Vyvolána funkce
arguments: Ověřené argumenty funkce
metadata: Slovník pro ukládání dalších dat mezi middlewarem
result: Návratová hodnota funkce (lze upravit)
terminate: Příznak pro zastavení dalšího zpracování
kwargs: Další argumenty klíčových slov předané metodě chatu, která vyvolala tuto funkci

next Volatelný pokračuje k dalšímu middlewaru nebo provede skutečnou funkci.

Tady je jednoduchý příklad protokolování s logikou před a po next volání:

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 chatu

Middleware chatu zachycuje žádosti o chat odeslané do modelů AI. Používá následující ChatContext funkce:

chat_client: Vyvolání chatovacího klienta
messages: Seznam zpráv odesílaných do služby AI
chat_options: Možnosti žádosti o chat
is_streaming: Logická hodnota označující, jestli se jedná o vyvolání streamování
metadata: Slovník pro ukládání dalších dat mezi middlewarem
result: Odpověď na chat z umělé inteligence (lze upravit)
terminate: Příznak pro zastavení dalšího zpracování
kwargs: Další argumenty klíčových slov předané chatovacímu klientovi

Volající next pokračuje v dalším middlewaru nebo odešle požadavek do služby AI.

Tady je jednoduchý příklad protokolování s logikou před a po next volání:

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

Dekorátory middlewaru funkcí

Dekorátory poskytují explicitní deklaraci typu middlewaru bez nutnosti psaní poznámek. Jsou užitečné v případech, kdy:

Nepoužíváte poznámky typu.
Potřebujete explicitní deklaraci typu middlewaru.
Chcete zabránit neshodám typů

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

Middleware založený na třídách je užitečný pro stavové operace nebo komplexní logiku, která využívá vzory návrhu orientovaného na objekty.

Třída middlewaru agenta

Middleware agenta založený na třídách používá metodu process , která má stejný podpis a chování jako middleware založený na funkcích. Metoda process přijímá stejné context a next parametry a je vyvolána úplně stejným způsobem.

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

Třída middlewaru funkcí

Middleware funkcí založený na třídě také používá metodu process se stejným podpisem a chováním jako middleware založený na funkcích. Metoda přijímá stejné context parametry a next parametry.

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

Třída middlewaru chatu

Middleware chatu založený na třídě se řídí stejným vzorem s metodou process , která má stejný podpis a chování pro chatovací middleware založený na funkcích.

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

Registrace middlewaru

Middleware je možné zaregistrovat na dvou úrovních s různými obory a chováním.

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

Klíčové rozdíly:

Úroveň agenta: Trvalá napříč všemi spuštěními nakonfigurovaná jednou při vytváření agenta
Úroveň spuštění: Vztahuje se pouze na konkrétní spuštění, umožňuje přizpůsobení jednotlivých požadavků.
Pořadí spuštění: Middleware agenta (vnější) → Spuštění middlewaru (nejvnitřnější) → spuštění agenta

Ukončení middlewaru

Middleware může ukončit spouštění brzy pomocí context.terminate. To je užitečné pro kontroly zabezpečení, omezování rychlosti nebo chyby ověřová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)

Co znamená ukončení:

Nastavení context.terminate = True signálů, které by mělo zastavit zpracování
Před ukončením můžete poskytnout vlastní výsledek, abyste uživatelům poskytli zpětnou vazbu.
Při ukončení middlewaru se úplně přeskočí spuštění agenta.

Přepsání výsledku middlewaru

Middleware může přepsat výsledky ve scénářích bez streamování i ve scénářích streamování, což vám umožní změnit nebo zcela nahradit odpovědi agenta.

Typ context.result výsledku závisí na tom, jestli je vyvolání agenta streamované nebo nestreamové:

Bez streamování: context.result obsahuje AgentRunResponse úplnou odpověď.
Streamování: context.result obsahuje asynchronní generátor, který poskytuje AgentRunResponseUpdate bloky dat.

Můžete použít context.is_streaming k rozlišení těchto scénářů a zpracování přepsání výsledků odpovídajícím způsobem.

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

Tento přístup middlewaru umožňuje implementovat sofistikované transformace odpovědí, filtrování obsahu, vylepšení výsledků a přizpůsobení streamování a přitom udržovat logiku agenta čistou a zaměřenou.

Další kroky

Rozšíření generování agenta (RAG)

Váš názor

Byla tato stránka užitečná?

Last updated on 2025-10-01

Sdílet prostřednictvím

Middleware agenta

Spuštění middlewaru agenta

Middleware volání funkcí

Middleware IChatClient

middleware Function-Based

Middleware agenta

Middleware funkce

Middleware chatu

Dekorátory middlewaru funkcí

middleware Class-Based

Třída middlewaru agenta

Třída middlewaru funkcí

Třída middlewaru chatu

Registrace middlewaru

Agent-Level vs. middleware Run-Level

Ukončení middlewaru

Přepsání výsledku middlewaru

Další kroky

Váš názor

Další materiály