Middleware agenta

Middleware v rozhraní Agent Framework 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í Agent Framework je možné přizpůsobit pomocí tří různých typů middlewaru:

1. 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.
2. 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.
3. 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(runFunc: CustomAgentRunMiddleware, runStreamingFunc: CustomAgentRunStreamingMiddleware)
        .Use(CustomFunctionCallingMiddleware)
    .Build();

Důležité

runFunc V ideálním případě by runStreamingFunc mělo být k dispozici obojí. Při poskytování pouze nesílovaného middlewaru ho agent použije pro streamování i volání bez streamování. Streamování se bude spouštět pouze v režimu bez streamování, aby stačit očekávání middlewaru.

Poznámka:

Existuje další přetížení, Use(sharedFunc: ...)které vám umožní poskytnout stejný middleware pro non-streaming a streamování bez blokování streamování. Sdílený middleware ale nebude moct zachytit ani přepsat výstup. Toto přetížení by se mělo použít ve scénářích, kdy potřebujete pouze zkontrolovat nebo upravit vstup před dosažením agenta.

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

Výstraha

DefaultAzureCredential je vhodný pro vývoj, ale vyžaduje pečlivé zvážení v produkčním prostředí. V produkčním prostředí zvažte použití konkrétních přihlašovacích údajů (např ManagedIdentityCredential. ) k zabránění problémům s latencí, neúmyslnému testování přihlašovacích údajů a potenciálním bezpečnostním rizikům z náhradních mechanismů.

IChatClient middleware lze také zaregistrovat pomocí factory metody při vytváření agenta prostřednictvím jedné z pomocných metod na klientských knihovnách SDK.

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

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

Spuštění middlewaru streamování agenta

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

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

Middleware volání funkcí

Poznámka:

Funkce volání middlewaru se v současné době podporuje pouze v AIAgent případě, že se používá FunctionInvokingChatClientnapří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 během této iterace bylo 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 historie chatu zůstane v nekonzistentním stavu, například obsahující obsah volání funkce bez obsahu výsledků funkce. To může vést k nepoužitelné historii chatu 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;
}

Návod

Kompletní runnable příklady najdete v ukázkách .NET .

Poznámka:

Další informace o middlewaru naleznete v IChatClient tématu Vlastní middleware IChatClient.

Rozhraní Agent Framework je možné přizpůsobit pomocí tří různých typů middlewaru:

1. Middleware agenta: Zachytí spuštění agenta, což vám umožní kontrolovat a upravovat vstupy, výstupy a tok řízení.
2. Middleware funkce: Zachytává volání funkce (nástroj) provedená během provádění agenta, povolení ověřování vstupu, transformace výsledku a řízení provádění.
3. Middleware chatu: Zachytí podkladové žádosti o chat odeslané do modelů AI a poskytuje přístup k nezpracovaným zprávám, možnostem a odpovědím.

Všechny typy podporují implementace založené na funkcích i na třídách. Pokud je zaregistrovaný více middlewarů stejného typu, vytvoří řetěz, ve kterém každý volá next volající, aby pokračoval ve zpracování.

Middleware agenta

Zachytávání middlewaru agenta a úpravy spuštění agenta. Používá následující AgentContext 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.

Založené na funkcích

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

Založené na třídách

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.

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

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.

Založené na funkcích

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

Založené na třídách

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

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

Založené na funkcích

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

Založené na třídá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:
        print(f"[Chat Class] Sending {len(context.messages)} messages to AI")
        await next(context)
        print("[Chat Class] AI response received")

Dekorátory middlewaru

Dekorátory poskytují explicitní deklaraci typu middlewaru bez nutnosti psaní poznámek. Jsou užitečné, když nepoužíváte poznámky k typu nebo chcete zabránit neshodám typů:

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

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

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

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 AgentResponse úplnou odpověď.
• Streamování: context.result obsahuje asynchronní generátor, který poskytuje AgentResponseUpdate 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: 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])]
            )

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.

Kompletní příklady middlewaru

Middleware založený na třídě

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

Middleware založený na funkcích

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

Middleware založený na dekorátoru

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

Další kroky

Odpovědi na pozadí agenta