探索微软智能代理框架

简介 (Introduction)

本节课将涵盖以下内容：

• 了解 Microsoft Agent 框架：核心特性与价值
• 探索 Microsoft Agent 框架的关键概念
• 高级 MAF 模式：工作流、中间件和内存

学习目标 (Learning Goals)

完成本节课后，您将学会如何：

• 使用 Microsoft Agent 框架构建生产级别的 AI 智能体 (AI Agents)
• 将 Microsoft Agent 框架的核心特性应用于您的智能体用例
• 使用包括工作流、中间件和可观测性在内的高级模式

代码示例 (Code Samples)

Microsoft Agent 框架 (MAF) 的代码示例可以在本仓库的 xx-python-agent-framework 和 xx-dotnet-agent-framework 文件夹中找到。

了解 Microsoft Agent 框架 (Understanding Microsoft Agent Framework)

Microsoft Agent 框架 (MAF) 是微软用于构建 AI 智能体的统一框架。它提供了极高的灵活性，能够应对生产和研究环境中各种各样的智能体用例，包括：

• 顺序智能体编排 (Sequential Agent orchestration)： 适用于需要逐步执行工作流的场景。
• 并发编排 (Concurrent orchestration)： 适用于智能体需要同时完成任务的场景。
• 群聊编排 (Group chat orchestration)： 适用于智能体可以协同完成同一项任务的场景。
• 交接编排 (Handoff Orchestration)： 适用于智能体在子任务完成后将任务移交给其他智能体的场景。
• 磁性编排 (Magnetic Orchestration)： 适用于管理员智能体创建和修改任务列表，并协调子智能体以完成任务的场景。

为了在生产环境中交付 AI 智能体，MAF 还包含了以下特性的支持：

• 可观测性 (Observability)： 通过使用 OpenTelemetry，记录 AI 智能体的每一个动作（包括工具调用、编排步骤、推理流程），并通过 Microsoft Foundry 仪表板进行性能监控。
• 安全性 (Security)： 通过在 Microsoft Foundry 上原生托管智能体，提供基于角色的访问控制、私有数据处理和内置内容安全等安全控制。
• 持久性 (Durability)： 智能体线程和工作流可以暂停、恢复并从错误中恢复，从而支持长时间运行的流程。
• 控制力 (Control)： 支持“人类在环 (human in the loop)”工作流，可将任务标记为需要人工审批。
• Microsoft Agent 框架还致力于通过以下方式实现互操作性：
• 云平台无关 (Being Cloud-agnostic)： 智能体可以在容器中、本地环境以及跨多个不同云平台运行。
• 供应商无关 (Being Provider-agnostic)： 可以通过您偏好的 SDK（包括 Azure OpenAI 和 OpenAI）创建智能体。
• 集成开放标准 (Integrating Open Standards)： 智能体可以利用诸如 Agent-to-Agent (A2A) 和模型上下文协议 (MCP) 等协议来发现和使用其他智能体及工具。
• 插件和连接器 (Plugins and Connectors)： 可以连接到数据和内存服务，如 Microsoft Fabric、SharePoint、Pinecone 和 Qdrant。

让我们来看看这些特性是如何应用于 Microsoft Agent 框架的一些核心概念上的。

Microsoft Agent 框架的关键概念 (Key Concepts of Microsoft Agent Framework)

智能体 (Agents)

创建智能体 创建智能体需要定义推理服务（LLM 供应商）、一组供 AI 智能体遵循的指令以及一个指定的名称：

agent = AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent( instructions="You are good at recommending trips to customers based on their preferences.", name="TripRecommender" )

上面的示例使用的是 Azure OpenAI，但也可以使用各种服务来创建智能体，包括 Microsoft Foundry Agent 服务：

AzureAIAgentClient(async_credential=credential).create_agent( name="HelperAgent", instructions="You are a helpful assistant." ) as agent

OpenAI 响应、ChatCompletion API：

agent = OpenAIResponsesClient().create_agent( name="WeatherBot", instructions="You are a helpful weather assistant.", )

agent = OpenAIChatClient().create_agent( name="HelpfulAssistant", instructions="You are a helpful assistant.", )

或者使用 MiniMax，它提供与 OpenAI 兼容的 API 以及超大上下文窗口（高达 204K tokens）：

agent = OpenAIChatClient(base_url="https://api.minimax.io/v1", api_key=os.environ["MINIMAX_API_KEY"], model_id="MiniMax-M2.7").create_agent( name="HelpfulAssistant", instructions="You are a helpful assistant.", )

或者使用 A2A 协议的 远程智能体 (remote agents)：

agent = A2AAgent( name=agent_card.name, description=agent_card.description, agent_card=agent_card, url="https://your-a2a-agent-host" )

运行智能体 使用 .run 或 .run_stream 方法运行智能体，以获取非流式或流式响应。

result = await agent.run("What are good places to visit in Amsterdam?")
print(result.text)

async for update in agent.run_stream("What are the good places to visit in Amsterdam?"):
if update.text:
print(update.text, end="", flush=True)

每次运行智能体时，还可以选择自定义参数，例如智能体使用的 max_tokens（最大 token 数）、智能体能够调用的 tools（工具），甚至用于该智能体的 model（模型）本身。 这在需要特定模型或工具来完成用户任务的情况下非常有用。

工具 (Tools)

工具可以在定义智能体时指定：

def get_attractions( location: Annotated[str, Field(description="The location to get the top tourist attractions for")], ) -> str: """Get the top tourist attractions for a given location.""" return f"The top attractions for {location} are." 

# When creating a ChatAgent directly 

agent = ChatAgent( chat_client=OpenAIChatClient(), instructions="You are a helpful assistant", tools=[get_attractions]

也可以在运行智能体时指定：

result1 = await agent.run( "What's the best place to visit in Seattle?", tools=[get_attractions] # Tool provided for this run only )

智能体线程 (Agent Threads)

智能体线程用于处理多轮对话。可以通过以下两种方式创建线程：

使用 get_new_thread()，允许线程随着时间的推移被保存。

在运行智能体时自动创建线程，且该线程仅在当前运行期间有效。

创建线程的代码如下所示：

# Create a new thread. 
thread = agent.get_new_thread() # Run the agent with the thread. 
response = await agent.run("Hello, I am here to help you book travel. Where would you like to go?", thread=thread)

然后，您可以序列化该线程以将其存储备用：

# Create a new thread. 
thread = agent.get_new_thread() 

# Run the agent with the thread. 

response = await agent.run("Hello, how are you?", thread=thread) 

# Serialize the thread for storage. 

serialized_thread = await thread.serialize() 

# Deserialize the thread state after loading from storage. 

resumed_thread = await agent.deserialize_thread(serialized_thread)

智能体中间件 (Agent Middleware)

智能体通过与工具和 LLM 交互来完成用户任务。在某些场景中，我们希望在这些交互之间执行操作或进行追踪。智能体中间件使我们能够通过以下方式实现这一点：

函数中间件 (Function Middleware) 该中间件允许我们在智能体和它将要调用的函数/工具之间执行操作。一个典型的使用场景是，您可能希望在函数调用时进行一些日志记录。 在下面的代码中，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")

聊天中间件 (Chat Middleware) 该中间件允许我们在智能体与 LLM 的请求之间执行或记录操作。 它包含重要信息，例如发送到 AI 服务的消息 (messages)。

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

智能体内存 (Agent Memory)

正如在“智能体内存”课程中所介绍的，内存是使智能体能够在不同上下文中运行的重要元素。MAF 提供了几种不同类型的内存：

内存中存储 (In-Memory Storage) 这是在应用程序运行时存储在线程中的内存。

# Create a new thread. 
thread = agent.get_new_thread() # Run the agent with the thread. 
response = await agent.run("Hello, I am here to help you book travel. Where would you like to go?", thread=thread)

持久化消息 (Persistent Messages) 这种内存用于跨越不同会话存储对话历史记录。它是通过 chat_message_store_factory 定义的：

from agent_framework import ChatMessageStore

# Create a custom message store
def create_message_store():
    return ChatMessageStore()

agent = ChatAgent(
    chat_client=OpenAIChatClient(),
    instructions="You are a Travel assistant.",
    chat_message_store_factory=create_message_store
)

动态内存 (Dynamic Memory) 这种内存会在智能体运行之前添加到上下文中。这些内存可以存储在诸如 mem0 之类的外部服务中：

from agent_framework.mem0 import Mem0Provider

# Using Mem0 for advanced memory capabilities
memory_provider = Mem0Provider(
    api_key="your-mem0-api-key",
    user_id="user_123",
    application_id="my_app"
)

agent = ChatAgent(
    chat_client=OpenAIChatClient(),
    instructions="You are a helpful assistant with memory.",
    context_providers=memory_provider
)

智能体可观测性 (Agent Observability)

可观测性对于构建可靠且可维护的智能体系统至关重要。MAF 集成了 OpenTelemetry，以提供分布式追踪 (tracing) 和指标计量 (meters)，从而实现更好的可观测性。

from agent_framework.observability import get_tracer, get_meter

tracer = get_tracer()
meter = get_meter()
with tracer.start_as_current_span("my_custom_span"):
    # do something
    pass
counter = meter.create_counter("my_custom_counter")
counter.add(1, {"key": "value"})

工作流 (Workflows)

MAF 提供了工作流，即完成任务的预定义步骤，并将 AI 智能体作为这些步骤中的组件。

工作流由不同的组件构成，以实现更好的控制流。工作流还支持多智能体编排 (multi-agent orchestration) 和检查点 (checkpointing) 机制，用于保存工作流状态。

工作流的核心组件包括：

执行器 (Executors) 执行器接收输入消息，执行分配给它们的任务，然后生成输出消息。这推动了工作流向前发展，以完成更大的任务。执行器可以是 AI 智能体或自定义逻辑。

边 (Edges) 边用于定义工作流中消息的流向。它们可以是：

直接边 (Direct Edges) - 执行器之间简单的一对一连接：

 WorkflowBuilder

builder = WorkflowBuilder()
builder.add_edge(source_executor, target_executor)
builder.set_start_executor(source_executor)
workflow = builder.build()

• 条件边 (Conditional Edges) - 在满足特定条件后激活。例如，当酒店房间不可用时，执行器可以建议其他选项。
• Switch-case 边 - 根据定义的条件将消息路由到不同的执行器。例如，如果旅游客户拥有优先访问权，他们的任务将通过另一个工作流处理。
• 扇出边 (Fan-out Edges) - 将一条消息发送到多个目标。
• 扇入边 (Fan-in Edges) - 从不同的执行器收集多条消息并发送到一个目标。

事件 (Events) 为了提供对工作流更好的可观测性，MAF 提供了内置的执行事件，包括：

• WorkflowStartedEvent - 工作流执行开始
• WorkflowOutputEvent - 工作流产生输出
• WorkflowErrorEvent - 工作流遇到错误
• ExecutorInvokeEvent - 执行器开始处理
• ExecutorCompleteEvent - 执行器完成处理
• RequestInfoEvent - 发出一个请求

高级 MAF 模式 (Advanced MAF Patterns)

前面的部分介绍了 Microsoft Agent 框架的关键概念。在构建更复杂的智能体时，这里有一些值得考虑的高级模式：

• 中间件组合 (Middleware Composition)： 使用函数和聊天中间件链式连接多个中间件处理程序（日志记录、身份验证、速率限制），以实现对智能体行为的细粒度控制。
• 工作流检查点 (Workflow Checkpointing)： 使用工作流事件和序列化来保存和恢复长时间运行的智能体进程。
• 动态工具选择 (Dynamic Tool Selection)： 结合 RAG（检索增强生成）工具描述和 MAF 的工具注册功能，确保每次查询只提供相关的工具。
• 多智能体交接 (Multi-Agent Handoff)： 使用工作流边和条件路由来编排专业化智能体之间的任务交接。
• 代码示例 (Code Samples)

Microsoft Agent 框架的代码示例可以在本仓库的 xx-python-agent-framework 和 xx-dotnet-agent-framework 文件夹中找到。