Oprogramowanie pośredniczące agenta

Oprogramowanie pośredniczące w strukturze agentów zapewnia zaawansowany sposób przechwytywania, modyfikowania i ulepszania interakcji agentów na różnych etapach wykonywania. Za pomocą oprogramowania pośredniczącego można zaimplementować kwestie związane z wycinaniem krzyżowym, takie jak rejestrowanie, walidacja zabezpieczeń, obsługa błędów i przekształcanie wyników bez modyfikowania podstawowego agenta lub logiki funkcji.

Program Agent Framework można dostosować przy użyciu trzech różnych typów oprogramowania pośredniczącego:

1. Oprogramowanie pośredniczące Uruchom agenta: umożliwia przechwycenie wszystkich uruchomień agenta, dzięki czemu dane wejściowe i wyjściowe mogą być sprawdzane i/lub modyfikowane zgodnie z potrzebami.
2. Wywoływanie oprogramowania pośredniczącego funkcji: umożliwia przechwycenie wszystkich wywołań funkcji wykonywanych przez agenta, dzięki czemu dane wejściowe i wyjściowe mogą być sprawdzane i modyfikowane zgodnie z potrzebami.
3. IChatClient oprogramowanie pośredniczące: umożliwia przechwycenie wywołań do IChatClient implementacji, w której agent jest używany IChatClient na potrzeby wywołań wnioskowania, na przykład w przypadku używania metody ChatClientAgent.

Wszystkie typy oprogramowania pośredniczącego są implementowane za pośrednictwem wywołania zwrotnego funkcji, a gdy wiele wystąpień oprogramowania pośredniczącego tego samego typu jest zarejestrowanych, tworzą łańcuch, w którym każde wystąpienie oprogramowania pośredniczącego ma wywołać następny w łańcuchu za pośrednictwem podanego nextFuncelementu .

Uruchamianie agenta i wywoływanie typów oprogramowania pośredniczącego można zarejestrować na agencie przy użyciu konstruktora agenta z istniejącym obiektem agenta.

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

Ważne

W idealnym przypadku należy podać zarówno, jak runFunc i runStreamingFunc . W przypadku udostępniania tylko oprogramowania pośredniczącego spoza przesyłania strumieniowego agent będzie używać go zarówno do wywołań przesyłania strumieniowego, jak i nieprzesyłania strumieniowego. Przesyłanie strumieniowe będzie działać tylko w trybie przesyłania strumieniowego, aby wystarczyć oczekiwaniom oprogramowania pośredniczącego.

Uwaga / Notatka

Istnieje dodatkowe przeciążenie , Use(sharedFunc: ...)które umożliwia zapewnienie tego samego oprogramowania pośredniczącego dla nieprzesyłania strumieniowego i przesyłania strumieniowego bez blokowania przesyłania strumieniowego. Jednak udostępnione oprogramowanie pośredniczące nie będzie mogło przechwycić ani zastąpić danych wyjściowych. To przeciążenie powinno być używane w scenariuszach, w których trzeba tylko sprawdzić lub zmodyfikować dane wejściowe przed dotarciem do agenta.

IChatClient Oprogramowanie pośredniczące można zarejestrować na elementy IChatClient przed użyciem ChatClientAgentelementu , przy użyciu wzorca konstruktora klienta czatu.

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

Ostrzeżenie

DefaultAzureCredential jest wygodne do programowania, ale wymaga starannego rozważenia w środowisku produkcyjnym. W środowisku produkcyjnym rozważ użycie określonego poświadczenia (np ManagedIdentityCredential. ) w celu uniknięcia problemów z opóźnieniami, niezamierzonego sondowania poświadczeń i potencjalnego ryzyka związanego z bezpieczeństwem z mechanizmów rezerwowych.

IChatClient Oprogramowanie pośredniczące można również zarejestrować za pomocą metody fabrycznej podczas tworzenia agenta przy użyciu jednej z metod pomocniczych na klientach 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());

Uruchamianie oprogramowania pośredniczącego agenta

Oto przykład oprogramowania pośredniczącego uruchamiania agenta, który może sprawdzać i/lub modyfikować dane wejściowe i wyjściowe z uruchomienia 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;
}

Agent uruchamia oprogramowanie pośredniczące przesyłania strumieniowego

Oto przykład agenta uruchamia oprogramowanie pośredniczące przesyłania strumieniowego, które może sprawdzać i/lub modyfikować dane wejściowe i wyjściowe z uruchomienia przesyłania strumieniowego 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);
}

Wywoływanie oprogramowania pośredniczącego przez funkcję

Uwaga / Notatka

Średnie oprogramowanie do wywoływania funkcji jest obecnie obsługiwane tylko w przypadku komponentu AIAgent, który używa FunctionInvokingChatClient, na przykład ChatClientAgent.

Oto przykład wywołania oprogramowania pośredniczącego, które może sprawdzać i/lub modyfikować wywoływaną funkcję oraz wynik tego wywołania.

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

Istnieje możliwość przerwania pętli wywołań funkcji za pomocą wywołania funkcji wywołującego oprogramowanie pośredniczące przez ustawienie podanej FunctionInvocationContext.Terminate wartości true. Dzięki temu pętla wywoływania funkcji nie będzie wysyłać żądania do usługi wnioskowania zawierającej wyniki wywołania funkcji po wywołaniu funkcji. Jeśli podczas tej iteracji jest dostępnych więcej niż jedna funkcja, może również uniemożliwić wykonywanie pozostałych funkcji.

Ostrzeżenie

Zakończenie pętli wywołań funkcji może spowodować pozostawienie historii czatu w niespójnym stanie, na przykład zawierającej zawartość wywołania funkcji bez zawartości wyniku funkcji. Może to spowodować, że historia czatu będzie bezużyteczna dla dalszych przebiegów.

Oprogramowanie pośredniczące IChatClient

Oto przykład oprogramowania pośredniczącego dla klienta czatu, które może sprawdzać i/lub modyfikować dane wejściowe i wyjściowe w ramach żądania do usługi analizy, którą zapewnia klient czatu.

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

Zobacz przykłady dla platformy .NET , aby uzyskać pełne przykłady możliwych do uruchomienia.

Uwaga / Notatka

Aby uzyskać więcej informacji na temat IChatClient oprogramowania pośredniczącego, zobacz Niestandardowe oprogramowanie pośredniczące IChatClient.

Program Agent Framework można dostosować przy użyciu trzech różnych typów oprogramowania pośredniczącego:

1. Oprogramowanie pośredniczące agenta: przechwytuje wykonywanie uruchamiania agenta, co umożliwia inspekcję i modyfikowanie danych wejściowych, danych wyjściowych i przepływu sterowania.
2. Oprogramowanie pośredniczące funkcji: przechwytuje wywołania funkcji (narzędzia) wykonywane podczas wykonywania agenta, włączając walidację danych wejściowych, przekształcanie wyników i kontrolę wykonywania.
3. Oprogramowanie pośredniczące czatu: przechwytuje podstawowe żądania czatu wysyłane do modeli sztucznej inteligencji, zapewniając dostęp do nieprzetworzonych wiadomości, opcji i odpowiedzi.

Wszystkie typy obsługują implementacje oparte na funkcjach i oparte na klasach. Gdy zarejestrowano wiele oprogramowania pośredniczącego tego samego typu, tworzą łańcuch, w którym każde wywołanie next umożliwia kontynuowanie przetwarzania.

Oprogramowanie pośredniczące agenta

Oprogramowanie pośredniczące agenta przechwytuje i modyfikuje wykonywanie uruchamiania agenta. Używa elementu AgentContext , który zawiera:

• agent: Wywoływany agent
• messages: Lista wiadomości czatu w konwersacji
• is_streaming: wartość logiczna wskazująca, czy odpowiedź jest przesyłana strumieniowo
• metadata: Słownik do przechowywania dodatkowych danych między oprogramowaniem pośredniczącym
• result: Odpowiedź agenta (można zmodyfikować)
• terminate: Flaga, aby zatrzymać dalsze przetwarzanie
• kwargs: Dodatkowe argumenty słowa kluczowego przekazane do metody uruchamiania agenta

Obiekt next wywołujący kontynuuje łańcuch oprogramowania pośredniczącego lub wykonuje agenta, jeśli jest to ostatnie oprogramowanie pośredniczące.

Oparte na funkcjach

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

Oparte na klasach

Oprogramowanie pośredniczące agenta opartego na klasach używa process metody, która ma ten sam podpis i zachowanie co oprogramowanie pośredniczące oparte na funkcjach.

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

Oprogramowanie pośredniczące funkcji

Oprogramowanie pośredniczące funkcji przechwytuje wywołania funkcji w agentach. Używa elementu FunctionInvocationContext , który zawiera:

• function: wywoływana funkcja
• arguments: Zweryfikowane argumenty funkcji
• metadata: Słownik do przechowywania dodatkowych danych między oprogramowaniem pośredniczącym
• result: zwracana wartość funkcji (można zmodyfikować)
• terminate: Flaga, aby zatrzymać dalsze przetwarzanie
• kwargs: Dodatkowe argumenty słowa kluczowego przekazane do metody czatu, która wywołała tę funkcję

Wywołanie next jest kontynuowane do następnego oprogramowania pośredniczącego lub wykonuje rzeczywistą funkcję.

Oparte na funkcjach

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

Oparte na klasach

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

Oprogramowanie pośredniczące czatu

Oprogramowanie pośredniczące czatu przechwytuje żądania czatu wysyłane do modeli sztucznej inteligencji. Używa elementu ChatContext , który zawiera:

• chat_client: Wywoływany klient czatu
• messages: Lista komunikatów wysyłanych do usługi sztucznej inteligencji
• options: opcje żądania czatu
• is_streaming: wartość logiczna wskazująca, czy jest to wywołanie przesyłania strumieniowego
• metadata: Słownik do przechowywania dodatkowych danych między oprogramowaniem pośredniczącym
• result: Odpowiedź czatu ze sztucznej inteligencji (można zmodyfikować)
• terminate: Flaga, aby zatrzymać dalsze przetwarzanie
• kwargs: Dodatkowe argumenty słowa kluczowego przekazane do klienta czatu

Wywołanie next jest kontynuowane do następnego oprogramowania pośredniczącego lub wysyła żądanie do usługi sztucznej inteligencji.

Oparte na funkcjach

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

Oparte na klasach

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

Dekoratory oprogramowania pośredniczącego

Dekoratory zapewniają jawną deklarację typu oprogramowania pośredniczącego bez konieczności adnotacji typu. Są one przydatne, gdy nie używasz adnotacji typu lub chcesz zapobiec niezgodności typów:

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

Rejestracja oprogramowania pośredniczącego

Oprogramowanie pośredniczące można zarejestrować na dwóch poziomach z różnymi zakresami i zachowaniami.

oprogramowanie pośredniczące Agent-Level a oprogramowanie pośredniczące 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?")

Kluczowe różnice:

• Poziom agenta: trwały we wszystkich przebiegach, skonfigurowany raz podczas tworzenia agenta
• Poziom uruchamiania: stosowany tylko do określonych przebiegów, umożliwia dostosowanie poszczególnych żądań
• Kolejność wykonywania: oprogramowanie pośredniczące agenta (najbardziej zewnętrzne) → uruchamianie oprogramowania pośredniczącego (najbardziej wewnętrzne) → wykonywanie agenta

Kończenie oprogramowania pośredniczącego

Oprogramowanie pośredniczące może zakończyć wykonywanie na wczesnym etapie przy użyciu polecenia context.terminate. Jest to przydatne w przypadku kontroli zabezpieczeń, ograniczania szybkości lub błędów walidacji.

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 oznacza zakończenie:

• Ustawianie context.terminate = True sygnałów, że przetwarzanie powinno zostać zatrzymane
• Przed zakończeniem możesz podać wynik niestandardowy, aby przekazać użytkownikom opinię
• Wykonywanie agenta jest całkowicie pomijane po zakończeniu działania oprogramowania pośredniczącego

Przesłonięcia wyniku oprogramowania pośredniczącego

Oprogramowanie pośredniczące może zastąpić wyniki zarówno w scenariuszach przesyłania strumieniowego, jak i przesyłania strumieniowego, co umożliwia modyfikowanie lub całkowite zastępowanie odpowiedzi agenta.

Typ wyniku zależy context.result od tego, czy wywołanie agenta jest przesyłane strumieniowo, czy nie przesyłane strumieniowo:

• Brak przesyłania strumieniowego: context.result zawiera element AgentResponse z pełną odpowiedzią
• Przesyłanie strumieniowe: context.result zawiera generator asynchroniczny, który daje AgentResponseUpdate fragmenty

Można użyć context.is_streaming metody , aby odpowiednio rozróżnić te scenariusze i obsługiwać przesłonięcia wyników.

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

To podejście oprogramowania pośredniczącego umożliwia zaimplementowanie zaawansowanego przekształcania odpowiedzi, filtrowania zawartości, ulepszania wyników i dostosowywania przesyłania strumieniowego przy jednoczesnym zachowaniu czystej i skoncentrowanej logiki agenta.

Kompletne przykłady oprogramowania pośredniczącego

Oprogramowanie pośredniczące oparte na klasach

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

Oprogramowanie pośredniczące oparte na funkcjach

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

Oprogramowanie pośredniczące oparte na dekoratorze

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

Dalsze kroki

Odpowiedzi w tle agenta