Oprogramowanie pośredniczące agenta

Struktura agenta 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 do wywołań wnioskowania, np. 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();

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 AzureCliCredential())
    .GetChatClient(deploymentName)
    .AsIChatClient();

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

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

IChatClient Oprogramowanie pośredniczące można również zarejestrować przy użyciu metody fabryki podczas konstruowania agenta za pośrednictwem jednej z metod pomocnika na klientach zestawu SDK.

var agent = new AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential())
    .GetChatClient(deploymentName)
    .CreateAIAgent("You are a helpful assistant.", clientFactory: (chatClient) => chatClient
        .AsBuilder()
            .Use(getResponseFunc: CustomChatClientMiddleware, getStreamingResponseFunc: null)
        .Build());

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<AgentRunResponse> CustomAgentRunMiddleware(
    IEnumerable<ChatMessage> messages,
    AgentThread? thread,
    AgentRunOptions? options,
    AIAgent innerAgent,
    CancellationToken cancellationToken)
{
    Console.WriteLine(messages.Count());
    var response = await innerAgent.RunAsync(messages, thread, options, cancellationToken).ConfigureAwait(false);
    Console.WriteLine(response.Messages.Count);
    return response;
}

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

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

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

Oto przykład funkcji wywołującej oprogramowanie pośredniczące, które mogą sprawdzać i/lub modyfikować wywoływaną funkcję oraz wynik wywołania funkcji.

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ępna więcej niż jedna funkcja, może również uniemożliwić wykonywanie pozostałych funkcji.

Oprogramowanie pośredniczące IChatClient

Oto przykład oprogramowania pośredniczącego klienta czatu, który może sprawdzać i/lub modyfikować dane wejściowe i wyjściowe żądania do usługi wnioskowania, 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;
}

oprogramowanie pośredniczące Function-Based

Oprogramowanie pośredniczące oparte na funkcjach to najprostszy sposób implementowania oprogramowania pośredniczącego przy użyciu funkcji asynchronicznych. Takie podejście jest idealne w przypadku operacji bezstanowych i zapewnia lekkie rozwiązanie dla typowych scenariuszy oprogramowania pośredniczącego.

Oprogramowanie pośredniczące agenta

Oprogramowanie pośredniczące agenta przechwytuje i modyfikuje wykonywanie uruchamiania agenta. Używa elementu AgentRunContext , 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.

Oto prosty przykład rejestrowania z logiką przed wywołaniem i po next wywołaniu:

async def logging_agent_middleware(
    context: AgentRunContext,
    next: Callable[[AgentRunContext], Awaitable[None]],
) -> None:
    """Agent middleware that logs execution timing."""
    # Pre-processing: Log before agent execution
    print("[Agent] Starting execution")

    # Continue to next middleware or agent execution
    await next(context)

    # Post-processing: Log after agent execution
    print("[Agent] Execution completed")

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

Oto prosty przykład rejestrowania z logiką przed wywołaniem i po next wywołaniu:

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

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

Oto prosty przykład rejestrowania z logiką przed wywołaniem i po next wywołaniu:

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

Dekoratory oprogramowania pośredniczącego funkcji

Dekoratory zapewniają jawną deklarację typu oprogramowania pośredniczącego bez konieczności adnotacji typu. Są one przydatne, gdy:

• Nie używasz adnotacji typu
• Potrzebna jest jawna deklaracja typu oprogramowania pośredniczącego
• Chcesz zapobiec niezgodnościom typów

from agent_framework import agent_middleware, function_middleware, chat_middleware

@agent_middleware  # Explicitly marks as agent middleware
async def simple_agent_middleware(context, next):
    """Agent middleware with decorator - types are inferred."""
    print("Before agent execution")
    await next(context)
    print("After agent execution")

@function_middleware  # Explicitly marks as function middleware
async def simple_function_middleware(context, next):
    """Function middleware with decorator - types are inferred."""
    print(f"Calling function: {context.function.name}")
    await next(context)
    print("Function call completed")

@chat_middleware  # Explicitly marks as chat middleware
async def simple_chat_middleware(context, next):
    """Chat middleware with decorator - types are inferred."""
    print(f"Processing {len(context.messages)} chat messages")
    await next(context)
    print("Chat processing completed")

oprogramowanie pośredniczące Class-Based

Oprogramowanie pośredniczące oparte na klasach jest przydatne w przypadku operacji stanowych lub złożonej logiki, która korzysta z wzorców projektowych zorientowanych na obiekty.

Klasa oprogramowania pośredniczącego agenta

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. Metoda process odbiera te same context parametry i next jest wywoływana w dokładnie taki sam sposób.

from agent_framework import AgentMiddleware, AgentRunContext

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

    async def process(
        self,
        context: AgentRunContext,
        next: Callable[[AgentRunContext], Awaitable[None]],
    ) -> None:
        # Pre-processing: Log before agent execution
        print("[Agent Class] Starting execution")

        # Continue to next middleware or agent execution
        await next(context)

        # Post-processing: Log after agent execution
        print("[Agent Class] Execution completed")

Klasa oprogramowania pośredniczącego funkcji

Oprogramowanie pośredniczące funkcji oparte na klasach używa process również metody z tym samym podpisem i zachowaniem co oprogramowanie pośredniczące oparte na funkcjach. Metoda otrzymuje te same context parametry i next .

from agent_framework import FunctionMiddleware, FunctionInvocationContext

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

    async def process(
        self,
        context: FunctionInvocationContext,
        next: Callable[[FunctionInvocationContext], Awaitable[None]],
    ) -> None:
        # Pre-processing: Log before function execution
        print(f"[Function Class] Calling {context.function.name}")

        # Continue to next middleware or function execution
        await next(context)

        # Post-processing: Log after function execution
        print(f"[Function Class] {context.function.name} completed")

Klasa oprogramowania pośredniczącego czatu

Oprogramowanie pośredniczące czatu oparte na klasach jest zgodne z tym samym wzorcem process z metodą, która ma identyczny podpis i zachowanie w oprogramowaniu pośredniczącym czatu opartego na funkcjach.

from agent_framework import ChatMiddleware, ChatContext

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

    async def process(
        self,
        context: ChatContext,
        next: Callable[[ChatContext], Awaitable[None]],
    ) -> None:
        # Pre-processing: Log before AI call
        print(f"[Chat Class] Sending {len(context.messages)} messages to AI")

        # Continue to next middleware or AI service
        await next(context)

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

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).create_agent(
    name="WeatherAgent",
    instructions="You are a helpful weather assistant.",
    tools=get_weather,
    middleware=[
        SecurityAgentMiddleware(),  # Applies to all runs
        TimingFunctionMiddleware(),  # Applies to all runs
    ],
) as agent:

    # This run uses agent-level middleware only
    result1 = await agent.run("What's the weather in Seattle?")

    # This run uses agent-level + run-level middleware
    result2 = await agent.run(
        "What's the weather in Portland?",
        middleware=[  # Run-level middleware (this run only)
            logging_chat_middleware,
        ]
    )

    # This run uses agent-level middleware only (no run-level)
    result3 = await agent.run("What's the weather in Vancouver?")

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: AgentRunContext,
    next: Callable[[AgentRunContext], Awaitable[None]],
) -> None:
    """Middleware that blocks execution based on conditions."""
    # Check for blocked content
    last_message = context.messages[-1] if context.messages else None
    if last_message and last_message.text:
        if "blocked" in last_message.text.lower():
            print("Request blocked by middleware")
            context.terminate = True
            return

    # If no issues, continue normally
    await next(context)

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 AgentRunResponse z pełną odpowiedzią
• Przesyłanie strumieniowe: context.result zawiera generator asynchroniczny, który daje AgentRunResponseUpdate 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: AgentRunContext,
    next: Callable[[AgentRunContext], Awaitable[None]]
) -> None:
    """Middleware that overrides weather results for both streaming and non-streaming."""

    # Execute the original agent logic
    await next(context)

    # Override results if present
    if context.result is not None:
        custom_message_parts = [
            "Weather Override: ",
            "Perfect weather everywhere today! ",
            "22°C with gentle breezes. ",
            "Great day for outdoor activities!"
        ]

        if context.is_streaming:
            # Streaming override
            async def override_stream() -> AsyncIterable[AgentRunResponseUpdate]:
                for chunk in custom_message_parts:
                    yield AgentRunResponseUpdate(contents=[TextContent(text=chunk)])

            context.result = override_stream()
        else:
            # Non-streaming override
            custom_message = "".join(custom_message_parts)
            context.result = AgentRunResponse(
                messages=[ChatMessage(role=Role.ASSISTANT, text=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.