代理程式中介軟體

Agent Framework 中的中介軟體提供了一種強大的方式，能攔截、修改及增強執行各階段的代理互動。您可以使用中介軟體來實作跨領域關注點，例如日誌記錄、安全驗證、錯誤處理和結果轉換，而無需修改核心代理程式或函數邏輯。

代理框架可透過三種不同類型的中介軟體進行客製化：

代理程式執行中介件：允許攔截所有代理程式執行，以便可以根據需要檢查和/或修改輸入和輸出。
函數呼叫中間件：允許攔截代理執行的所有函數調用，以便根據需要檢查和修改輸入和輸出。
IChatClient中介軟體：允許攔截執行執行IChatClientIChatClient者進行推論呼叫時的呼叫，例如當使用 ChatClientAgent。

所有類型的中間件都是透過函數回呼實現的，當註冊多個相同類型的中間件實例時，它們會形成一個鏈，其中每個中間件實例都期望透過提供的 nextFunc.

代理程式執行及函數呼叫中介軟體類型可以在代理程式上登錄，方法是將代理程式建置器與現有代理程式物件搭配使用。

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

這很重要

理想狀況下runFuncrunStreamingFunc兩者都應該提供。當只提供非串流中介軟體時，代理程式會同時使用它來執行串流與非串流的呼叫。串流只會在非串流模式下運行，以符合中介軟體的預期。

備註

還有一個額外的過載， Use(sharedFunc: ...)允許你提供相同的中介軟體，適用於非串流和串流，而不阻擋串流。不過，共用的中介軟體無法攔截或覆蓋輸出。這種過載應該用於只需檢查或修改輸入，然後才會到達代理的情境。

IChatClient 中介軟體可以在上 IChatClient 註冊，然後 ChatClientAgent再使用，方法是使用聊天用戶端產生器模式。

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 在透過 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());

代理程式執行中介軟體

以下是代理程式執行中介軟體的範例，可檢查和/或修改代理程式執行的輸入和輸出。

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

代理運行串流中介軟體

這裡有一個代理執行串流中介軟體的範例，可以檢查和/或修改代理串流執行的輸入與輸出。

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

函數呼叫中間件

備註

目前僅支援具備AIAgent的函式呼叫中介軟體，例如FunctionInvokingChatClient。

以下是函數呼叫中介軟體的範例，它可以檢查和/或修改正在呼叫的函數，以及函數呼叫的結果。

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

可以透過將提供的 FunctionInvocationContext.Terminate 設定為 true 來終止函數呼叫中介軟體的函數呼叫迴圈。這可防止函式呼叫迴圈在函式呼叫之後，向包含函式呼叫結果的推論服務發出要求。如果在此迭代中有多個可調用函式，也可能阻止剩餘函式的執行。

警告

終止函式呼叫迴圈可能會導致執行緒處於不一致狀態，例如包含函式呼叫內容卻沒有函式結果內容。這可能導致執行緒無法繼續執行。

IChatClient 中介軟體

以下是聊天用戶端中介軟體的範例，可以檢查和/或修改聊天用戶端提供的推論服務請求的輸入和輸出。

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

備註

如需中介軟體的詳細資訊 IChatClient ，請參閱自訂 IChatClient 中介軟體。

Function-Based 中間件

基於函數的中間件是使用非同步函數實現中間件的最簡單方法。這種方法非常適合無狀態操作，並為常見的中間件場景提供了輕量級的解決方案。

代理程式中介軟體

代理程式中介軟體會攔截並修改代理程式執行執行。它使用 which AgentRunContext 包含：

agent：呼叫的代理程式
messages：對話中的聊天訊息清單
is_streaming：布林值，指出回應是否為串流
metadata：用於在中間件之間儲存額外資料的字典
result：客服專員的回應（可修改）
terminate：旗標以停止進一步處理
kwargs：傳遞至代理程式執行方法的其他關鍵字引數

可呼叫者 next 會繼續中介軟體鏈結，或執行代理程式（如果它是最後一個中介軟體）。

這是一個簡單的日誌記錄範例，其中包含可呼叫之前和之後 next 的邏輯：

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

函數中間件

函數中介件會攔截代理程式內的函數呼叫。它使用 which FunctionInvocationContext 包含：

function：正在呼叫的函數
arguments：函數的已驗證引數
metadata：用於在中間件之間儲存額外資料的字典
result：函數的傳回值（可修改）
terminate：旗標以停止進一步處理
kwargs：傳遞至叫用此函式之聊天方法的其他關鍵字引數

可呼叫繼續 next 到下一個中間件或執行實際函數。

這是一個簡單的日誌記錄範例，其中包含可呼叫之前和之後 next 的邏輯：

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

聊天中間件

聊天中間件攔截發送給 AI 模型的聊天請求。它使用 which ChatContext 包含：

chat_client：正在叫用的聊天用戶端
messages：傳送至 AI 服務的訊息清單
chat_options：聊天請求的選項
is_streaming：布林值，指出這是否是串流呼叫
metadata：用於在中間件之間儲存額外資料的字典
result：來自 AI 的聊天回應（可修改）
terminate：旗標以停止進一步處理
kwargs：傳遞至聊天用戶端的其他關鍵字引數

可呼叫項目會繼續 next 到下一個中介軟體，或將要求傳送至 AI 服務。

這是一個簡單的日誌記錄範例，其中包含可呼叫之前和之後 next 的邏輯：

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

函數中介軟體裝飾器

裝飾器提供明確的中介軟體類型宣告，而不需要類型註解。它們在以下情況下很有幫助：

您不會使用類型註釋
您需要明確的中介軟體類型宣告
您想要防止類型不相符

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

Class-Based 中間件

基於類別的中介軟體對於受益於物件導向設計模式的有狀態操作或複雜邏輯非常有用。

代理程式中介軟體類別

以類別為基礎的代理程式中介軟體使用的方法 process ，其特徵與以函數為基礎的中介軟體具有相同的簽章和行為。該 process 方法接收相同的 context 和 next 參數，並以完全相同的方式調用。

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

函數中介件類

基於類別的函數中間件也使用與 process 基於函數的中間件具有相同簽名和行為的方法。該方法接收相同的 context 和 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")

聊天中間件類

以類別為基礎的聊天中介軟體遵循相同的模式，其 process 方法具有與函數型聊天中介相同的簽章和行為。

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

中間件註冊

中間件可以在兩個層級註冊，具有不同的範圍和行為。

Agent-Level 與 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?")

主要區別：

代理程式層級：在所有執行中持續存在，在建立代理程式時設定一次
執行層級：僅套用至特定執行，允許每個要求自訂
執行順序：代理中間件（最外層）→運行中間件（最內層）→代理執行

中間件終止

中間件可以使用提前終止執行 context.terminate。這對於安全性檢查、速率限制或驗證失敗非常有用。

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)

終止的意義：

設定 context.terminate = True 處理應停止的訊號
您可以在終止之前提供自訂結果，以提供使用者意見反應
當中間件終止時，會完全略過代理程式執行

中介軟體結果覆寫

中介軟體可以覆寫非串流和串流案例中的結果，可讓您修改或完全取代代理程式回應。

結果 context.result 類型取決於代理程式呼叫是串流還是非串流：

非串流：context.result包含具有AgentRunResponse完整回應的
串流： context.result 包含產生區塊的 AgentRunResponseUpdate 非同步產生器

您可以使用來 context.is_streaming 區分這些案例，並適當地處理結果覆寫。

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

這種中間件方法允許您實施複雜的響應轉換、內容過濾、結果增強和流自定義，同時保持代理邏輯乾淨且集中。

後續步驟

探員背景回應

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-01-08

共用方式為

代理程式中介軟體

代理程式執行中介軟體

代理運行串流中介軟體

函數呼叫中間件

IChatClient 中介軟體

Function-Based 中間件

代理程式中介軟體

函數中間件

聊天中間件

函數中介軟體裝飾器

Class-Based 中間件

代理程式中介軟體類別

函數中介件類

聊天中間件類

中間件註冊

Agent-Level 與 Run-Level 中間件

中間件終止

中介軟體結果覆寫

後續步驟

意見反應

其他資源