Ü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:

Ü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ó.
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.
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 DefaultAzureCredential())
    .GetChatClient(deploymentName)
    .AsIChatClient();

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

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

Figyelmeztetés

DefaultAzureCredential a fejlesztéshez kényelmes, de az éles környezetben gondos megfontolást igényel. Éles környezetben fontolja meg egy adott hitelesítő adat (pl. ) használatát a ManagedIdentityCredentialkésési problémák, a nem szándékos hitelesítő adatok próbaüzeme és a tartalék mechanizmusok esetleges biztonsági kockázatainak 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 AzureOpenAIClient(new Uri(endpoint), new DefaultAzureCredential())
    .GetChatClient(deploymentName)
    .AsAIAgent("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.

Figyelmeztetés

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

Jótanács

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:

Ü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.
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.
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 hívhatót a next feldolgozás folytatásához.

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

Függvényalapú

async def logging_agent_middleware(
    context: AgentContext,
    next: Callable[[AgentContext], 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")

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,
        next: Callable[[AgentContext], Awaitable[None]],
    ) -> None:
        print("[Agent Class] Starting execution")
        await next(context)
        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
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.

Függvényalapú

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

Osztályalapú

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:
        print(f"[Function Class] Calling {context.function.name}")
        await next(context)
        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:

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.

Függvényalapú

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

Osztályalapú

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:
        print(f"[Chat Class] Sending {len(context.messages)} messages to AI")
        await next(context)
        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, next):
    print("Before agent execution")
    await next(context)
    print("After agent execution")

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

@chat_middleware
async def simple_chat_middleware(context, next):
    print(f"Processing {len(context.messages)} chat messages")
    await next(context)
    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.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).as_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: AgentContext,
    next: Callable[[AgentContext], 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: AgentContext,
    next: Callable[[AgentContext], 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=[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.azure import AzureAIAgentClient
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,
        AzureAIAgentClient(credential=credential).as_agent(
            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.azure import AzureAIAgentClient
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,
        AzureAIAgentClient(credential=credential).as_agent(
            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.azure import AzureAIAgentClient
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,
        AzureAIAgentClient(credential=credential).as_agent(
            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övetkező lépések

Ügynök háttérválaszai

Visszacsatolás

Hasznos volt ez az oldal?

Last updated on 2026-02-13

Megosztás:

Ügynök köztes szoftvere

Ügynök futtatása Köztes szoftver

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

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

IChatClient köztes szoftver

Ügynök köztes szoftvere

Függvényalapú

Osztályalapú

Függvény köztes szoftver

Függvényalapú

Osztályalapú

Csevegés köztes szoftver

Függvényalapú

Osztályalapú

Middleware Decorators

Köztes szoftver regisztrációja

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

Köztes szoftver leállítása

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

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

Osztályalapú köztes szoftver

Függvényalapú köztes szoftver

Dekorátoralapú köztes szoftver

Következő lépések

Visszacsatolás

További források