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

Important

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 AIProjectClient(
    new Uri("<your-foundry-project-endpoint>"),
    new DefaultAzureCredential())
        .GetProjectOpenAIClient()
        .GetProjectResponsesClient()
        .AsIChatClient(deploymentName);

var middlewareEnabledChatClient = chatClient
    .AsBuilder()
        .Use(getResponseFunc: CustomChatClientMiddleware, getStreamingResponseFunc: null)
    .Build();

var agent = new ChatClientAgent(middlewareEnabledChatClient, instructions: "You are a helpful assistant.");

Warning

DefaultAzureCredential a fejlesztéshez kényelmes, de a termelési környezetben gondos megfontolást igényel. Éles környezetben fontolja meg egy adott hitelesítő adat (pl. ManagedIdentityCredential) használatát a késési problémák elkerülése, a hitelesítő adatok nem szándékos próbálgatásának és a tartalék mechanizmusokból eredő esetleges biztonsági kockázatok elkerülése érdekében.

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 AIProjectClient(
    new Uri("<your-foundry-project-endpoint>"),
    new DefaultAzureCredential())
        .AsAIAgent(
            model: deploymentName,
            instructions: "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,
    AgentSession? session,
    AgentRunOptions? options,
    AIAgent innerAgent,
    CancellationToken cancellationToken)
{
    Console.WriteLine(messages.Count());
    var response = await innerAgent.RunAsync(messages, session, 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,
    AgentSession? session,
    AgentRunOptions? options,
    AIAgent innerAgent,
    [EnumeratorCancellation] CancellationToken cancellationToken)
{
    Console.WriteLine(messages.Count());
    List<AgentResponseUpdate> updates = [];
    await foreach (var update in innerAgent.RunStreamingAsync(messages, session, 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.

Warning

A függvényhívási ciklus megszüntetése esetén előfordulhat, hogy a csevegési előzmények inkonzisztens állapotban maradnak, például függvényhívási tartalmat tartalmaznak függvényeredmény-tartalom nélkül. Ez azt eredményezheti, hogy a csevegési előzmények nem lesznek használhatatlanok 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;
}

Tip

A teljes futtatható példákért tekintse meg a .NET-mintákat .

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.

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

1. Ügynök köztes szoftver: Elfogja az ügynök futtatásának végrehajtását, lehetővé téve a bemenetek, kimenetek és vezérlőfolyamatok vizsgálatát és módosítását.
2. Függvényközvetítő: Elfogja az ügynök végrehajtása során végrehajtott függvényhívásokat( eszköz), lehetővé téve a bemeneti érvényesítést, az eredményátalakítást és a végrehajtás-vezérlést.
3. Csevegés köztes szoftver: Elfogja az AI-modelleknek küldött mögöttes csevegési kéréseket, és hozzáférést biztosít a nyers üzenetekhez, a lehetőségekhez és a válaszokhoz.

Minden típus támogatja a függvényalapú és az osztályalapú implementációkat is. Ha több azonos típusú köztes szoftver van regisztrálva, azok egy láncot alkotnak, amelyben mindegyik meghívja a visszahívást a call_next feldolgozás folytatásához. call_nextnem argumentumként veszi fel a kontextust; a köztes szoftver közvetlenül mutálja a megosztott környezet objektumát, majd várja.call_next()

Megjegyzés:

Köztes szoftverrendelés vegyes regisztrációs hatókörökkel:

• Az ügynökszintű köztes szoftver futtatási szintű köztes szoftvereket burkol.
• Az ügynök köztes szoftvere [A1, A2] és a köztes szoftver [R1, R2]futtatása esetén a végrehajtási sorrend a következő: A1 -> A2 -> R1 -> R2 -> Agent -> R2 -> R1 -> A2 -> A1.
• A függvény/csevegés köztes szoftver ugyanazt a burkoló elvet követi eszköz/csevegés-hívás idején.

Tip

Az ügynökre, csevegésre és függvényközvetítőre kiterjedő szabványosított, sikertelen vezérlési határért tekintse meg az Ügynökhookok című témakört. Az Agent Hooks koordinálja az alapvető streamelési és adatmegőrzési viselkedést is, amelyet a szokásos köztes szoftver önmagában nem tud biztosítani.

Ü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 AgentContext használja:

• agent: A meghívandó ügynök
• messages: A beszélgetésben lévő csevegőüzenetek listája
• session: Az aktuális ügynök munkamenete, ha van ilyen
• options: Ügynökfuttatási beállítások ehhez a meghíváshoz
• stream: 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ó)
• kwargs: Örökölt futtatókörnyezeti kulcsszóargumentumok továbbítása az ügynökfuttatási metódusnak
• client_kwargs: Ügyfélspecifikus futtatókörnyezeti értékek alsóbb rétegbeli csevegőügyfelek számára
• function_invocation_kwargs: Futtatókörnyezeti értékek, amelyek az eszközökre lesznek továbbítva

A call_next visszahívás folytatja a köztes szoftverláncot, vagy végrehajtja az ügynököt, ha ez az utolsó köztes szoftver.

Függvényalapú

async def inject_tool_runtime_defaults(
    context: AgentContext,
    call_next: Callable[[], Awaitable[None]],
) -> None:
    """Agent middleware that sets tool-only runtime defaults."""
    print("[Agent] Starting execution")
    context.function_invocation_kwargs.setdefault("tenant", "contoso")
    context.function_invocation_kwargs.setdefault("request_source", "agent-middleware")

    await call_next()

    print("[Agent] Execution completed")

Osztályalapú

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.

from agent_framework import AgentMiddleware, AgentContext

class LoggingAgentMiddleware(AgentMiddleware):
    """Agent middleware that logs execution."""

    async def process(
        self,
        context: AgentContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        print("[Agent Class] Starting execution")
        await call_next()
        print("[Agent Class] 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
• session: Az aktuális ügynök munkamenete, ha van ilyen
• 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ó)
• kwargs: Futtatókörnyezeti kulcsszóargumentumok, amelyeket a rendszer továbbít az eszköz meghívásának

A call_next visszahívás folytatódik a következő köztes szoftverre, vagy végrehajtja a tényleges függvényt.

Függvényalapú

async def inject_function_kwargs(
    context: FunctionInvocationContext,
    call_next: Callable[[], Awaitable[None]],
) -> None:
    """Function middleware that enriches tool runtime values."""
    context.kwargs.setdefault("tenant", "contoso")
    context.kwargs.setdefault("request_source", "function-middleware")

    await call_next()

Osztályalapú

from agent_framework import FunctionMiddleware, FunctionInvocationContext

class LoggingFunctionMiddleware(FunctionMiddleware):
    """Function middleware that logs function execution."""

    async def process(
        self,
        context: FunctionInvocationContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        print(f"[Function Class] Calling {context.function.name}")
        await call_next()
        print(f"[Function Class] {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:

• 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
• stream: 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ó)
• kwargs: A csevegőügyfélnek átadott további kulcsszóargumentumok
• function_invocation_kwargs: Csak eszközalapú futásidejű értékek, amelyeket a csevegőréteg továbbít

A call_next visszahívás folytatódik a következő köztes szoftverre, vagy elküldi a kérést az AI szolgáltatásnak.

Megjegyzés:

A csevegési köztes szoftver a függvényhívási cikluson belül fut. Ez azt jelenti, hogy minden modellhíváshoz végrehajtja azokat a hívásokat is, amelyek egy többfordulós eszközhívási sorozat során küldik vissza az eszköz eredményeit a modellnek.

Függvényalapú

async def logging_chat_middleware(
    context: ChatContext,
    call_next: Callable[[], 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 call_next()

    # Post-processing: Log after AI response
    print("[Chat] AI response received")

Osztályalapú

from agent_framework import ChatMiddleware, ChatContext

class LoggingChatMiddleware(ChatMiddleware):
    """Chat middleware that logs AI interactions."""

    async def process(
        self,
        context: ChatContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        print(f"[Chat Class] Sending {len(context.messages)} messages to AI")
        await call_next()
        print("[Chat Class] AI response received")

Middleware Decorators

A dekorátorok explicit köztes szoftvertípus-deklarációt biztosítanak anélkül, hogy típusjegyzeteket kellene megadniuk. Hasznosak, ha nem használ típusjegyzeteket, vagy meg szeretné akadályozni a típuseltéréseket:

from agent_framework import agent_middleware, function_middleware, chat_middleware

@agent_middleware
async def simple_agent_middleware(context, call_next):
    print("Before agent execution")
    await call_next()
    print("After agent execution")

@function_middleware
async def simple_function_middleware(context, call_next):
    print(f"Calling function: {context.function.name}")
    await call_next()
    print("Function call completed")

@chat_middleware
async def simple_chat_middleware(context, call_next):
    print(f"Processing {len(context.messages)} chat messages")
    await call_next()
    print("Chat processing completed")

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 import Agent
from agent_framework.foundry import FoundryChatClient
from azure.identity.aio import AzureCliCredential

# Agent-level middleware: Applied to ALL runs of the agent
async with Agent(
    client=FoundryChatClient(credential=credential),
    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 beállítással context.result és a felemeléssel MiddlewareTerminationkorán leállítja. Ez biztonsági ellenőrzések, sebességkorlátozás vagy érvényesítési hibák esetén hasznos.

from agent_framework import AgentContext, AgentResponse, Message, MiddlewareTermination

async def blocking_middleware(
    context: AgentContext,
    call_next: Callable[[], 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.result = AgentResponse(
                messages=[Message(role="assistant", contents=["This request was blocked by middleware."])]
            )
            raise MiddlewareTermination(result=context.result)

    # If no issues, continue normally
    await call_next()

Mit jelent a megszűnés:

• Beállítás context.result a felemelés MiddlewareTermination előtt, ha egyéni választ szeretne visszaadni
• A felemelés MiddlewareTermination leállítja a köztes szoftverlánc többi részét, és kihagyja a normál végrehajtási útvonalat
• Ez a minta az ügynök, a függvény és a csevegőközvetítő szoftver esetében működik

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.stream .

async def weather_override_middleware(
    context: AgentContext,
    call_next: Callable[[], Awaitable[None]]
) -> None:
    """Middleware that overrides weather results for both streaming and non-streaming."""

    # Execute the original agent logic
    await call_next()

    # 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.stream:
            # Streaming override
            async def override_stream() -> AsyncIterable[AgentResponseUpdate]:
                for chunk in custom_message_parts:
                    yield AgentResponseUpdate(contents=[Content.from_text(text=chunk)])

            context.result = override_stream()
        else:
            # Non-streaming override
            custom_message = "".join(custom_message_parts)
            context.result = AgentResponse(
                messages=[Message(role="assistant", contents=[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.

Middleware-példák teljes körű használata

Osztályalapú köztes szoftver

# Copyright (c) Microsoft. All rights reserved.

import asyncio
import time
from collections.abc import Awaitable, Callable
from random import randint
from typing import Annotated

from agent_framework import (
    AgentContext,
    AgentMiddleware,
    AgentResponse,
    FunctionInvocationContext,
    FunctionMiddleware,
    Message,
    tool,
)
from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from azure.identity.aio import AzureCliCredential
from pydantic import Field

"""
Class-based MiddlewareTypes Example

This sample demonstrates how to implement middleware using class-based approach by inheriting
from AgentMiddleware and FunctionMiddleware base classes. The example includes:

- SecurityAgentMiddleware: Checks for security violations in user queries and blocks requests
  containing sensitive information like passwords or secrets
- LoggingFunctionMiddleware: Logs function execution details including timing and parameters

This approach is useful when you need stateful middleware or complex logic that benefits
from object-oriented design patterns.
"""


# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
    location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
    """Get the weather for a given location."""
    conditions = ["sunny", "cloudy", "rainy", "stormy"]
    return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."


class SecurityAgentMiddleware(AgentMiddleware):
    """Agent middleware that checks for security violations."""

    async def process(
        self,
        context: AgentContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        # Check for potential security violations in the query
        # Look at the last user message
        last_message = context.messages[-1] if context.messages else None
        if last_message and last_message.text:
            query = last_message.text
            if "password" in query.lower() or "secret" in query.lower():
                print("[SecurityAgentMiddleware] Security Warning: Detected sensitive information, blocking request.")
                # Override the result with warning message
                context.result = AgentResponse(
                    messages=[Message("assistant", ["Detected sensitive information, the request is blocked."])]
                )
                # Simply don't call call_next() to prevent execution
                return

        print("[SecurityAgentMiddleware] Security check passed.")
        await call_next()


class LoggingFunctionMiddleware(FunctionMiddleware):
    """Function middleware that logs function calls."""

    async def process(
        self,
        context: FunctionInvocationContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        function_name = context.function.name
        print(f"[LoggingFunctionMiddleware] About to call function: {function_name}.")

        start_time = time.time()

        await call_next()

        end_time = time.time()
        duration = end_time - start_time

        print(f"[LoggingFunctionMiddleware] Function {function_name} completed in {duration:.5f}s.")


async def main() -> None:
    """Example demonstrating class-based middleware."""
    print("=== Class-based MiddlewareTypes Example ===")

    # For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
    # authentication option.
    async with (
        AzureCliCredential() as credential,
        Agent(
            client=FoundryChatClient(credential=credential),
            name="WeatherAgent",
            instructions="You are a helpful weather assistant.",
            tools=get_weather,
            middleware=[SecurityAgentMiddleware(), LoggingFunctionMiddleware()],
        ) as agent,
    ):
        # Test with normal query
        print("\n--- Normal Query ---")
        query = "What's the weather like in Seattle?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")

        # Test with security-related query
        print("--- Security Test ---")
        query = "What's the password for the weather service?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")


if __name__ == "__main__":
    asyncio.run(main())

Függvényalapú köztes szoftver

# Copyright (c) Microsoft. All rights reserved.

import asyncio
import time
from collections.abc import Awaitable, Callable
from random import randint
from typing import Annotated

from agent_framework import (
    AgentContext,
    AgentMiddleware,
    AgentResponse,
    FunctionInvocationContext,
    FunctionMiddleware,
    Message,
    tool,
)
from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from azure.identity.aio import AzureCliCredential
from pydantic import Field

"""
Class-based MiddlewareTypes Example

This sample demonstrates how to implement middleware using class-based approach by inheriting
from AgentMiddleware and FunctionMiddleware base classes. The example includes:

- SecurityAgentMiddleware: Checks for security violations in user queries and blocks requests
  containing sensitive information like passwords or secrets
- LoggingFunctionMiddleware: Logs function execution details including timing and parameters

This approach is useful when you need stateful middleware or complex logic that benefits
from object-oriented design patterns.
"""


# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
    location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
    """Get the weather for a given location."""
    conditions = ["sunny", "cloudy", "rainy", "stormy"]
    return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."


class SecurityAgentMiddleware(AgentMiddleware):
    """Agent middleware that checks for security violations."""

    async def process(
        self,
        context: AgentContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        # Check for potential security violations in the query
        # Look at the last user message
        last_message = context.messages[-1] if context.messages else None
        if last_message and last_message.text:
            query = last_message.text
            if "password" in query.lower() or "secret" in query.lower():
                print("[SecurityAgentMiddleware] Security Warning: Detected sensitive information, blocking request.")
                # Override the result with warning message
                context.result = AgentResponse(
                    messages=[Message("assistant", ["Detected sensitive information, the request is blocked."])]
                )
                # Simply don't call call_next() to prevent execution
                return

        print("[SecurityAgentMiddleware] Security check passed.")
        await call_next()


class LoggingFunctionMiddleware(FunctionMiddleware):
    """Function middleware that logs function calls."""

    async def process(
        self,
        context: FunctionInvocationContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        function_name = context.function.name
        print(f"[LoggingFunctionMiddleware] About to call function: {function_name}.")

        start_time = time.time()

        await call_next()

        end_time = time.time()
        duration = end_time - start_time

        print(f"[LoggingFunctionMiddleware] Function {function_name} completed in {duration:.5f}s.")


async def main() -> None:
    """Example demonstrating class-based middleware."""
    print("=== Class-based MiddlewareTypes Example ===")

    # For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
    # authentication option.
    async with (
        AzureCliCredential() as credential,
        Agent(
            client=FoundryChatClient(credential=credential),
            name="WeatherAgent",
            instructions="You are a helpful weather assistant.",
            tools=get_weather,
            middleware=[SecurityAgentMiddleware(), LoggingFunctionMiddleware()],
        ) as agent,
    ):
        # Test with normal query
        print("\n--- Normal Query ---")
        query = "What's the weather like in Seattle?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")

        # Test with security-related query
        print("--- Security Test ---")
        query = "What's the password for the weather service?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")


if __name__ == "__main__":
    asyncio.run(main())

Dekorátoralapú köztes szoftver

# Copyright (c) Microsoft. All rights reserved.

import asyncio
import time
from collections.abc import Awaitable, Callable
from random import randint
from typing import Annotated

from agent_framework import (
    AgentContext,
    AgentMiddleware,
    AgentResponse,
    FunctionInvocationContext,
    FunctionMiddleware,
    Message,
    tool,
)
from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from azure.identity.aio import AzureCliCredential
from pydantic import Field

"""
Class-based MiddlewareTypes Example

This sample demonstrates how to implement middleware using class-based approach by inheriting
from AgentMiddleware and FunctionMiddleware base classes. The example includes:

- SecurityAgentMiddleware: Checks for security violations in user queries and blocks requests
  containing sensitive information like passwords or secrets
- LoggingFunctionMiddleware: Logs function execution details including timing and parameters

This approach is useful when you need stateful middleware or complex logic that benefits
from object-oriented design patterns.
"""


# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
    location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
    """Get the weather for a given location."""
    conditions = ["sunny", "cloudy", "rainy", "stormy"]
    return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."


class SecurityAgentMiddleware(AgentMiddleware):
    """Agent middleware that checks for security violations."""

    async def process(
        self,
        context: AgentContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        # Check for potential security violations in the query
        # Look at the last user message
        last_message = context.messages[-1] if context.messages else None
        if last_message and last_message.text:
            query = last_message.text
            if "password" in query.lower() or "secret" in query.lower():
                print("[SecurityAgentMiddleware] Security Warning: Detected sensitive information, blocking request.")
                # Override the result with warning message
                context.result = AgentResponse(
                    messages=[Message("assistant", ["Detected sensitive information, the request is blocked."])]
                )
                # Simply don't call call_next() to prevent execution
                return

        print("[SecurityAgentMiddleware] Security check passed.")
        await call_next()


class LoggingFunctionMiddleware(FunctionMiddleware):
    """Function middleware that logs function calls."""

    async def process(
        self,
        context: FunctionInvocationContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        function_name = context.function.name
        print(f"[LoggingFunctionMiddleware] About to call function: {function_name}.")

        start_time = time.time()

        await call_next()

        end_time = time.time()
        duration = end_time - start_time

        print(f"[LoggingFunctionMiddleware] Function {function_name} completed in {duration:.5f}s.")


async def main() -> None:
    """Example demonstrating class-based middleware."""
    print("=== Class-based MiddlewareTypes Example ===")

    # For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
    # authentication option.
    async with (
        AzureCliCredential() as credential,
        Agent(
            client=FoundryChatClient(credential=credential),
            name="WeatherAgent",
            instructions="You are a helpful weather assistant.",
            tools=get_weather,
            middleware=[SecurityAgentMiddleware(), LoggingFunctionMiddleware()],
        ) as agent,
    ):
        # Test with normal query
        print("\n--- Normal Query ---")
        query = "What's the weather like in Seattle?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")

        # Test with security-related query
        print("--- Security Test ---")
        query = "What's the password for the weather service?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")


if __name__ == "__main__":
    asyncio.run(main())

Köztes szoftver áttekintése

A Middleware in Go elfogja és módosítja az ügynök viselkedését a futtatási szinten. Minden köztes szoftver implementálja a agent.Middleware felületet.

Beépített és keretrendszer által biztosított köztes szoftver

Komponens	Registration	Réteg	Alkalmazás célja
Automatikus hívás	agent/harness/toolautocall	Szolgáltató köztes szoftvere	Automatikusan meghívja a függvényekhez tartozó eszközöket
Strukturált kimenet	agent.WithStructuredOutput	Szolgáltató köztes szoftvere	Strukturált kimeneti elemzés kezelése
OpenTelemetry	provider/otelprovider	Ügynök köztes szoftvere	Nyomkövetési ügynökhívások
Naplózó futtatása	agent.Config.Logger	Ügynök köztes szoftvere	Naplózza az ügynök interakcióit

A környezetszolgáltatók nem implementációk, hanem agent.Middleware szomszédos életciklus-összetevők. Azok azután futnak, hogy az egyéni ügynök köztes szoftvere belépett a futtatásba, és mielőtt a szolgáltató köztes szoftver meghívja a modellt.

Köztes szoftver regisztrálása

import otelprovider "github.com/microsoft/agent-framework-go/provider/otelprovider"

a := foundryprovider.NewAgent(endpoint, token, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
    Config: agent.Config{
        Middlewares: []agent.Middleware{
            otelprovider.NewMiddleware(otelprovider.MiddlewareConfig{}),
            myCustomMiddleware,
        },
    },
})

A regisztrált agent.Config.Middlewares köztes szoftver a deklarált sorrendben van alkalmazva; az első köztes szoftver a legkülső egyéni réteget burkolja. Ez az egyéni réteg körbefuttatja az előzményszolgáltatókat, a környezetszolgáltatókat és a szolgáltatói köztes szoftvereket.

Köztes szoftver létrehozása

Akkor használható agent.MiddlewareFunc , ha a teljes szerkezettípus szükségtelen:

addGuidance := agent.MiddlewareFunc(
    func(next agent.RunFunc, ctx context.Context, messages []*message.Message, options ...agent.Option) iter.Seq2[*agent.ResponseUpdate, error] {
        guided := append([]*message.Message{message.NewText("Keep the response concise and avoid exposing secrets.")}, messages...)
        return next(ctx, guided, options...)
    },
)

a := foundryprovider.NewAgent(endpoint, token, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
    Config: agent.Config{
        Middlewares: []agent.Middleware{addGuidance},
    },
})

Következő lépések

Köztes szoftver definiálása