Ügynök köztes szoftvere

Az Agent Framework köztes szoftvere hatékony módot kínál az ügynök interakcióinak elfogására, módosítására és javítására a végrehajtás különböző szakaszaiban. A köztes szoftverrel olyan átfogó problémákat valósíthat meg, mint a naplózás, a biztonsági ellenőrzés, a hibakezelés és az eredményátalakítás az alapvető ügynök vagy a függvénylogika módosítása nélkül.

Az Agent Framework három különböző köztes szoftvertípussal szabható testre:

1. Ügynökfuttatási köztes szoftver: Lehetővé teszi az ügynökfuttatások elfogását, így a bemenet és a kimenet szükség szerint vizsgálható és/vagy módosítható.
2. Függvényhívás köztes szoftver: Lehetővé teszi az ügynök által végrehajtott összes függvényhívás elfogását, hogy a bemenet és a kimenet szükség szerint vizsgálható és módosítható legyen.
3. IChatClientköztes szoftver: Lehetővé teszi a hívások elfogását egy IChatClient implementációba, ahol az ügynök következtetési hívásokat használIChatClient, például a .ChatClientAgent

Az összes köztes szoftvertípust függvényvisszahívással implementáljuk, és ha több azonos típusú köztes szoftverpéldány van regisztrálva, akkor egy láncot alkotnak, ahol minden köztes szoftverpéldánynak a következőt kell meghívnia a láncban egy megadott nextFuncmódon.

Az ügynökfuttatás és a köztes szoftvertípusokat hívó függvény regisztrálható egy ügynökön az ügynökszerkesztővel egy meglévő ügynökobjektummal.

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

Fontos

Ideális esetben mindkettőt runFuncrunStreamingFunc meg kell adni. Ha csak a nem streamelt köztes szoftvereket adja meg, az ügynök a streameléshez és a nem streameléses meghívásokhoz is használni fogja. A streamelés csak nem streamelési módban fog futni, hogy megfeleljen a köztes szoftver elvárásainak.

Megjegyzés:

Van egy további túlterhelés, Use(sharedFunc: ...)amely lehetővé teszi, hogy ugyanazt a köztes szoftvert biztosítsa a nem streameléshez és a streameléshez anélkül, hogy blokkolja a streamelést. A megosztott köztes szoftver azonban nem fogja tudni elfogni vagy felülbírálni a kimenetet. Ezt a túlterhelést olyan forgatókönyvekhez kell használni, ahol csak az ügynök elérése előtt kell megvizsgálnia vagy módosítania a bemenetet.

IChatClient A köztes szoftver regisztrálható a IChatClient csevegőügyfél-készítői minta használatával, mielőtt használva ChatClientAgentlenne.

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 A köztes szoftver gyári módszerrel is regisztrálható, amikor egy ügynököt az SDK-ügyfelek egyik segédmetódusán keresztül hoz létre.

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

Ügynök futtatása Köztes szoftver

Íme egy példa az ügynök által futtatott köztes szoftverre, amely megvizsgálhatja és/vagy módosíthatja az ügynökfuttatás bemenetét és kimenetét.

async Task<AgentResponse> 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;
}

Az ügynök streamelési köztes szoftver futtatása

Íme egy példa az ügynök által futtatott streamelési köztes szoftverre, amely képes megvizsgálni és/vagy módosítani az ügynök streamelési futtatásának bemenetét és kimenetét.

    async IAsyncEnumerable<AgentResponseUpdate> CustomAgentRunStreamingMiddleware(
    IEnumerable<ChatMessage> messages,
    AgentThread? thread,
    AgentRunOptions? options,
    AIAgent innerAgent,
    [EnumeratorCancellation] CancellationToken cancellationToken)
{
    Console.WriteLine(messages.Count());
    List<AgentResponseUpdate> updates = [];
    await foreach (var update in innerAgent.RunStreamingAsync(messages, thread, options, cancellationToken))
    {
        updates.Add(update);
        yield return update;
    }

    Console.WriteLine(updates.ToAgentResponse().Messages.Count);
}

Köztes szoftver meghívása függvény

Megjegyzés:

A függvényszólításokat végző köztes réteg jelenleg csak olyan AIAgent támogatja, amely FunctionInvokingChatClient-t használ, például ChatClientAgent.

Íme egy példa a köztes szoftver meghívására, amely megvizsgálhatja és/vagy módosíthatja a meghívott függvényt, valamint a függvényhívás eredményét.

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

A függvényhívási ciklust a függvényhívó köztes szoftverrel is meg lehet szüntetni a megadott FunctionInvocationContext.Terminate érték igaz értékre állításával. Ez megakadályozza, hogy a függvényhívási ciklus kérést küldjön a függvényhívás eredményeit tartalmazó következtetési szolgáltatásnak a függvényhívás után. Ha az iteráció során több függvény is rendelkezésre áll a meghíváshoz, az megakadályozhatja a fennmaradó függvények végrehajtását is.

Figyelmeztetés

A függvényhívási ciklus megszakítása azt eredményezheti, hogy a szál inkonzisztens állapotban marad, például olyan függvényhívási tartalmat tartalmaz, amely nem tartalmaz függvényeredmény-tartalmat. Ez azt eredményezheti, hogy a szál nem használható a további futtatásokhoz.

IChatClient köztes szoftver

Íme egy példa a csevegőügyfél köztes szoftverére, amely megvizsgálhatja és/vagy módosíthatja a kérés bemenetét és kimenetét a csevegőügyfél által biztosított következtetési szolgáltatásba.

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

Megjegyzés:

A köztes szoftverről IChatClient további információt az Egyéni IChatClient köztes szoftver című témakörben talál.

köztes szoftver Function-Based

A függvényalapú köztes szoftver a legegyszerűbb módja a köztes szoftver implementálásának az aszinkron függvények használatával. Ez a megközelítés ideális állapot nélküli műveletekhez, és egyszerű megoldást kínál a köztes szoftverek gyakori forgatókönyveihez.

Ügynök köztes szoftvere

Az ügynök köztes szoftver elfogja és módosítja az ügynök futtatási végrehajtását. A következőt AgentRunContext használja:

• agent: A meghívandó ügynök
• messages: A beszélgetésben lévő csevegőüzenetek listája
• is_streaming: Logikai érték, amely azt jelzi, hogy a válasz streamelt-e
• metadata: Szótár a köztes szoftverek közötti további adatok tárolásához
• result: Az ügynök válasza (módosítható)
• terminate: Megjelölés a további feldolgozás leállításához
• kwargs: Az ügynökfuttatási metódusnak átadott további kulcsszóargumentumok

A next hívható folytatja a köztes szoftverláncot, vagy végrehajtja az ügynököt, ha az utolsó köztes szoftver.

Íme egy egyszerű naplózási példa a hívható előtti és utáni next logikával:

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

Függvény köztes szoftver

A függvényközvetítő az ügynökökön belül elfogja a függvényhívásokat. A következőt FunctionInvocationContext használja:

• function: A meghívandó függvény
• arguments: A függvény érvényesített argumentumai
• metadata: Szótár a köztes szoftverek közötti további adatok tárolásához
• result: A függvény visszatérési értéke (módosítható)
• terminate: Megjelölés a további feldolgozás leállításához
• kwargs: A függvényt meghívó csevegési metódusnak átadott további kulcsszóargumentumok

A next hívható továbbra is a következő köztes szoftverre lép, vagy végrehajtja a tényleges függvényt.

Íme egy egyszerű naplózási példa a hívható előtti és utáni next logikával:

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

Csevegés köztes szoftver

A csevegési köztes szoftver elfogja az AI-modelleknek küldött csevegési kéréseket. A következőt ChatContext használja:

• chat_client: A meghívandó csevegőügyfél
• messages: Az AI szolgáltatásnak küldött üzenetek listája
• options: A csevegési kérelem beállításai
• is_streaming: Logikai érték, amely azt jelzi, hogy ez egy streamhívás
• metadata: Szótár a köztes szoftverek közötti további adatok tárolásához
• result: Az AI csevegési válasza (módosítható)
• terminate: Megjelölés a további feldolgozás leállításához
• kwargs: A csevegőügyfélnek átadott további kulcsszóargumentumok

A next hívható továbbra is a következő köztes szoftver, vagy elküldi a kérelmet az AI szolgáltatásnak.

Íme egy egyszerű naplózási példa a hívható előtti és utáni next logikával:

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

Függvény middleware decorators

A dekorátorok explicit köztes szoftvertípus-deklarációt biztosítanak anélkül, hogy típusjegyzeteket kellene megadniuk. A következő esetekben hasznosak:

• Nem használ gépelt széljegyzeteket
• Explicit köztes szoftvertípus-deklarációra van szükség
• Meg szeretné akadályozni a típuseltéréseket

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

köztes szoftver Class-Based

Az osztályalapú köztes szoftver olyan állapotalapú műveletekhez vagy összetett logikákhoz hasznos, amelyek az objektumorientált tervezési minták előnyeit élvezik.

Ügynök köztes szoftver osztálya

Az osztályalapú ügynökközvetítő olyan metódust process használ, amely ugyanazzal az aláírással és működéssel rendelkezik, mint a függvényalapú köztes szoftver. A process metódus ugyanazokat context és next paramétereket kap, és a meghívás pontosan ugyanúgy történik.

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

Függvény köztes szoftver osztálya

Az osztályalapú köztes szoftver egy olyan metódust process is használ, amely ugyanazzal az aláírással és működéssel rendelkezik, mint a függvényalapú köztes szoftver. A metódus ugyanazokat context és next paramétereket kapja.

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

Csevegő köztes szoftver osztály

Az osztályalapú csevegőközvetíteni kívánt szoftver ugyanazt a mintát követi egy process olyan metódussal, amely azonos aláírással és viselkedéssel rendelkezik a függvényalapú csevegőközvetíteni kívánt szoftverhez.

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

Köztes szoftver regisztrációja

A köztes szoftver két különböző hatókörrel és viselkedéssel rendelkező szinten regisztrálható.

Agent-Level és Run-Level Köztes szoftver

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

Főbb különbségek:

• Ügynökszintű: Állandó minden futtatáson, egyszer konfigurálva az ügynök létrehozásakor
• Futtatási szint: Csak adott futtatásokra alkalmazható, lehetővé teszi a kérelemenkénti testreszabást
• Végrehajtási sorrend: Ügynök köztes szoftver (legkülső) → Köztes szoftver futtatása (legbelső) → ügynök végrehajtása

Köztes szoftver leállítása

A köztes szoftver a végrehajtást a korai használat során context.terminateleállítja. Ez biztonsági ellenőrzések, sebességkorlátozás vagy érvényesítési hibák esetén hasznos.

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)

Mit jelent a megszűnés:

• A feldolgozás leállítására vonatkozó jelek beállítása context.terminate = True
• A felhasználók visszajelzése érdekében a megszüntetés előtt egyéni eredményt adhat meg
• A köztes szoftver leállásakor a program teljesen kihagyja az ügynök végrehajtását

Köztes szoftver eredményének felülbírálása

A Köztes szoftver felülbírálhatja az eredményeket a nem streamelési és streamelési forgatókönyvekben is, így módosíthatja vagy teljesen lecserélheti az ügynök válaszait.

Az eredmény típusa context.result attól függ, hogy az ügynök meghívása streamelt vagy nem streamelt:

• Nem streamelés: context.result teljes választ tartalmaz AgentResponse
• Streamelés: context.result egy adattömböket eredményező AgentResponseUpdate aszinkron generátort tartalmaz

A forgatókönyvek megkülönböztetésére és az eredmény felülbírálásának megfelelő kezelésére használható context.is_streaming .

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[AgentResponseUpdate]:
                for chunk in custom_message_parts:
                    yield AgentResponseUpdate(contents=[TextContent(text=chunk)])

            context.result = override_stream()
        else:
            # Non-streaming override
            custom_message = "".join(custom_message_parts)
            context.result = AgentResponse(
                messages=[ChatMessage(role=Role.ASSISTANT, text=custom_message)]
            )

Ezzel a köztes szoftveres megközelítéssel kifinomult válaszátalakítást, tartalomszűrést, eredményfejlesztést és streamelési testreszabást valósíthat meg, miközben az ügynöklogika tiszta és koncentrált marad.

Következő lépések

Ügynök háttérválaszai