了解如何在几个简单的步骤中向代理添加中间件。 中间件允许截获和修改代理通信，以便记录、安全性和其他跨领域关注点。
先决条件
有关先决条件和安装 NuGet 包，请参阅本教程中的 “创建并运行简单代理 ”步骤。
步骤 1：创建简单代理
首先，使用函数工具创建基本代理。
using System;
using System.ComponentModel;
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;
[Description("The current datetime offset.")]
static string GetDateTime()
=> DateTimeOffset.Now.ToString();
AIAgent baseAgent = new AIProjectClient(
new Uri("<your-foundry-project-endpoint>"),
new DefaultAzureCredential())
.AsAIAgent(
model: "gpt-4o-mini",
instructions: "You are an AI assistant that helps people find information.",
tools: [AIFunctionFactory.Create(GetDateTime, name: nameof(GetDateTime))]);
警告
DefaultAzureCredential 对于开发来说很方便，但在生产中需要仔细考虑。 在生产环境中，请考虑使用特定凭据（例如）， ManagedIdentityCredential以避免延迟问题、意外凭据探测以及回退机制的潜在安全风险。
步骤 2：创建代理运行中间件
接下来，创建一个将在每次智能体运行时被调用的函数。 它允许检查代理的输入和输出。
除非打算使用中间件来停止运行，否则该函数应调用提供的 RunAsync 上的 innerAgent。
此示例中间件只检查代理运行的输入和输出，并输出传入和传出代理的消息数。
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
async Task<AgentResponse> CustomAgentRunMiddleware(
IEnumerable<ChatMessage> messages,
AgentSession? session,
AgentRunOptions? options,
AIAgent innerAgent,
CancellationToken cancellationToken)
{
Console.WriteLine($"Input: {messages.Count()}");
var response = await innerAgent.RunAsync(messages, session, options, cancellationToken).ConfigureAwait(false);
Console.WriteLine($"Output: {response.Messages.Count}");
return response;
}
步骤 3：将代理运行的中间件添加到您的软件代理中
若要将此中间件函数添加到您在步骤 1 中创建的 baseAgent，请使用构建器模式。
这会创建应用中间件的新代理。
baseAgent原始副本未修改。
var middlewareEnabledAgent = baseAgent
.AsBuilder()
.Use(runFunc: CustomAgentRunMiddleware, runStreamingFunc: null)
.Build();
现在，使用查询执行代理时，应调用中间件，输出输入消息数和响应消息数。
Console.WriteLine(await middlewareEnabledAgent.RunAsync("What's the current time?"));
步骤 4：创建函数调用中间件
注释
函数调用中间件目前仅支持使用 AIAgent 的 FunctionInvokingChatClient ，例如 ChatClientAgent。
您还可以创建一个中间件，它会在每个函数工具被调用时触发。 下面是函数调用中间件的示例，该中间件可以检查和/或修改所调用的函数以及函数调用的结果。
除非打算使用中间件来不执行函数工具，否则中间件应调用提供的 nextFunc。
using System.Threading;
using System.Threading.Tasks;
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;
}
步骤 5：将调用中间件的函数添加到代理
与添加代理运行中间件相同，可以添加函数调用中间件，如下所示：
var middlewareEnabledAgent = baseAgent
.AsBuilder()
.Use(CustomFunctionCallingMiddleware)
.Build();
现在，使用调用函数的查询执行代理时，应调用中间件，输出函数名称和调用结果。
Console.WriteLine(await middlewareEnabledAgent.RunAsync("What's the current time?"));
步骤 6：创建聊天客户端中间件
对于使用 IChatClient生成的代理，您可能需要截获从代理到 IChatClient的调用。
在这种情况下，可以将中间件用于IChatClient。
下面是聊天客户端中间件的示例，该中间件可以检查和/或修改聊天客户端向推理服务请求的输入和输出。
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
async Task<ChatResponse> CustomChatClientMiddleware(
IEnumerable<ChatMessage> messages,
ChatOptions? options,
IChatClient innerChatClient,
CancellationToken cancellationToken)
{
Console.WriteLine($"Input: {messages.Count()}");
var response = await innerChatClient.GetResponseAsync(messages, options, cancellationToken);
Console.WriteLine($"Output: {response.Messages.Count}");
return response;
}
注释
有关中间件的详细信息 IChatClient ，请参阅 自定义 IChatClient 中间件。
步骤 7：将聊天客户端中间软件添加到 IChatClient
若要将中间件添加到你的 IChatClient，可以使用生成器模式。
添加中间件后，可以像往常一样将 IChatClient 中间件与代理一起使用。
var chatClient = new AIProjectClient(
new Uri("<your-foundry-project-endpoint>"),
new DefaultAzureCredential())
.GetProjectOpenAIClient()
.GetProjectResponsesClient()
.AsIChatClient("gpt-4o-mini");
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 AIProjectClient(
new Uri("<your-foundry-project-endpoint>"),
new DefaultAzureCredential())
.AsAIAgent(
model: "gpt-4o-mini",
instructions: "You are a helpful assistant.",
clientFactory: (chatClient) => chatClient
.AsBuilder()
.Use(getResponseFunc: CustomChatClientMiddleware, getStreamingResponseFunc: null)
.Build());
步骤 1：创建简单代理
首先，创建基本代理：
import asyncio
from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from azure.identity.aio import AzureCliCredential
async def main():
credential = AzureCliCredential()
async with Agent(
client=FoundryChatClient(credential=credential),
name="GreetingAgent",
instructions="You are a friendly greeting assistant.",
) as agent:
result = await agent.run("Hello!")
print(result.text)
if __name__ == "__main__":
asyncio.run(main())
步骤 2：创建中间件
创建简单的日志记录中间件以查看代理何时运行：
from collections.abc import Awaitable, Callable
from agent_framework import AgentContext
async def logging_agent_middleware(
context: AgentContext,
call_next: Callable[[], Awaitable[None]],
) -> None:
"""Simple middleware that logs agent execution."""
print("Agent starting...")
# Continue to agent execution
await call_next()
print("Agent finished!")
步骤 3：将中间件添加到代理
创建代理时添加中间件：
async def main():
credential = AzureCliCredential()
async with Agent(
client=FoundryChatClient(credential=credential),
name="GreetingAgent",
instructions="You are a friendly greeting assistant.",
middleware=[logging_agent_middleware], # Add your middleware here
) as agent:
result = await agent.run("Hello!")
print(result.text)
步骤 4：创建函数中间件
如果代理使用函数，可以在工具执行之前截获函数调用并设置仅限工具的运行时值：
from collections.abc import Awaitable, Callable
from agent_framework import FunctionInvocationContext
def get_time(ctx: FunctionInvocationContext) -> str:
"""Get the current time."""
from datetime import datetime
source = ctx.kwargs.get("request_source", "direct")
return f"[{source}] {datetime.now().strftime('%H:%M:%S')}"
async def inject_function_kwargs(
context: FunctionInvocationContext,
call_next: Callable[[], Awaitable[None]],
) -> None:
"""Middleware that adds tool-only runtime values before execution."""
context.kwargs.setdefault("request_source", "middleware")
await call_next()
# Add both the function and middleware to your agent
async with Agent(
client=FoundryChatClient(credential=credential),
name="TimeAgent",
instructions="You can tell the current time.",
tools=[get_time],
middleware=[inject_function_kwargs],
) as agent:
result = await agent.run("What time is it?")
步骤 5：使用 Run-Level 中间件
还可以为特定运行添加中间件：
# Use middleware for this specific run only
result = await agent.run(
"This is important!",
middleware=[logging_function_middleware]
)
下一步是什么？
有关更高级的方案，请参阅 代理中间件用户指南，其中包括：
- 不同类型的中间件（代理、函数、聊天）。
- 基于类的复杂场景中间件。
- 中间件终止和结果重写。
- 高级中间件模式和最佳做法。
完整示例
基于类的中间件
# 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 import Agent
from agent_framework.foundry import FoundryChatClient
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,
Agent(
client=FoundryChatClient(credential=credential),
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())
基于函数的中间件
# 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 import Agent
from agent_framework.foundry import FoundryChatClient
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,
Agent(
client=FoundryChatClient(credential=credential),
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())
基于修饰器的中间件
# 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 import Agent
from agent_framework.foundry import FoundryChatClient
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,
Agent(
client=FoundryChatClient(credential=credential),
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())