【LangChain1.0】第九篇 Agent 架构设计

第九篇 Agent 架构设计

目标: 掌握 Multi-Agent 系统架构，理解三大通信协议（MCP / A2A / AG-UI），构建工业级智能体生态

在前面的篇章中，我们学习了单个 Agent 的构建方法。但在真实的生产环境中，复杂任务往往需要多个 Agent 协作、连接外部工具、与用户界面交互。本篇将深入探讨 Agent 架构的三个核心维度。

核心问题：

• 如何让多个 Agent 在同一进程内高效协作？（Swarm 模式）
• 如何让 Agent 连接外部工具和数据源？（MCP 协议）
• 如何让不同框架的 Agent 互相通信？（A2A 协议）
• 如何让 Agent 与用户界面实时交互？（AG-UI 协议）

三大协议矩阵：

第1章：架构演进与协议全景

本章目标: 理解 Agent 架构从单体到生态的演进路径，掌握协议选型策略

1.1 吴恩达的四种 Agentic 模式

在 2024 年的演讲中，Andrew Ng 总结了四种核心的 Agentic Workflow 模式：

本篇聚焦 Pattern 4：Multi-Agent Collaboration（多智能体协作）。

1.2 架构演进：从单体到生态

1.3 三大协议定位

💡 关键洞察: 这三个协议不是竞争关系，而是互补关系。MCP 负责"Agent↔工具"，A2A 负责"Agent↔Agent"，AG-UI 负责"Agent↔用户"。

1.4 技术选型决策树

第2章：Swarm 多 Agent 协作模式

本章目标: 掌握 LangGraph 官方推荐的 Multi-Agent 实现方式

2.1 核心定义：Swarm ≠ 蜂群算法

在 LangGraph 语境下，Swarm 不是指分布式群体智能算法，而是指：

多个 Agent 通过 Handoffs（控制权移交）进行去中心化协作的设计模式。

关键特性：

• 去中心化：没有中央路由器，每个 Agent 自己决定下一步该谁接手
• Handoff：不只是返回结果，而是传递完整的上下文和控制权
• 共享状态：所有 Agent 操作同一个 State 对象

2.2 官方推荐写法：Tool-Based Routing

LangGraph 官方推荐的 Multi-Agent 实现方式是：定义 transfer_to_X 工具，通过 Tool Call 触发条件边跳转。

实战场景：保险理赔系统

业务流程：

用户提交理赔
→ Triage Agent (分流)
→ Claim Processor (初审)
→ 风控 Agent / 法务 Agent (专业处理)
→ 结案

Step 1: 定义全局状态

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class InsuranceState(TypedDict):
"""全局共享状态"""
messages: Annotated[list, add_messages] # 对话历史（自动合并）
claim_id: str # 案件ID
risk_score: float # 风险评分
current_agent: str # 当前处理人

Step 2: 创建 Transfer 工具

from langchain_core.tools import tool

def create_transfer_tool(target_agent: str):
"""工厂函数：生成移交工具"""
@tool(f"transfer_to_{target_agent}")
def transfer() –> str:
f"""将任务移交给 {target_agent} 继续处理。"""
return target_agent
return transfer

Step 3: 定义 Agent 节点

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# === Triage Agent ===
triage_tools = [create_transfer_tool("ClaimProcessor")]

def triage_node(state: InsuranceState):
"""分流专员：收集信息并转发"""
system_msg = SystemMessage(content="""
你是理赔分流专员。职责：
1. 询问用户案件编号和基本情况
2. 记录到系统后，调用 transfer_to_ClaimProcessor 移交
""")
response = llm.bind_tools(triage_tools).invoke(
[system_msg] + state["messages"]
)
return {"messages": [response]}

# === Claim Processor ===
processor_tools = [
create_transfer_tool("FraudDetector"),
create_transfer_tool("LegalAdvisor")
]

def processor_node(state: InsuranceState):
"""初审专员：计算风险并分流"""
system_msg = SystemMessage(content="""
你是理赔初审员。职责：
1. 分析案件，计算风险评分（0-100）
2. 如果风险 > 70，调用 transfer_to_FraudDetector
3. 否则调用 transfer_to_LegalAdvisor
""")
response = llm.bind_tools(processor_tools).invoke(
[system_msg] + state["messages"]
)
return {"messages": [response]}

# === 终端节点 ===
def fraud_node(state: InsuranceState):
"""风控专员：终端节点"""
system_msg = SystemMessage(content="你是风控专员，进行最终审核。")
response = llm.invoke([system_msg] + state["messages"])
return {"messages": [response]}

def legal_node(state: InsuranceState):
"""法务专员：终端节点"""
system_msg = SystemMessage(content="你是法务顾问，提供法律意见。")
response = llm.invoke([system_msg] + state["messages"])
return {"messages": [response]}

Step 4: 构建路由图

from langgraph.graph import StateGraph, END

# 1. 初始化图
workflow = StateGraph(InsuranceState)

# 2. 添加节点
workflow.add_node("Triage", triage_node)
workflow.add_node("ClaimProcessor", processor_node)
workflow.add_node("FraudDetector", fraud_node)
workflow.add_node("LegalAdvisor", legal_node)

# 3. 定义路由函数
def router(state: InsuranceState):
"""根据最后一条消息的 Tool Call 决定路由"""
last_msg = state["messages"][–1]
if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
tool_name = last_msg.tool_calls[0]["name"]
if tool_name.startswith("transfer_to_"):
return tool_name.replace("transfer_to_", "")
return END

# 4. 添加条件边
workflow.add_conditional_edges("Triage", router)
workflow.add_conditional_edges("ClaimProcessor", router)
workflow.add_edge("FraudDetector", END)
workflow.add_edge("LegalAdvisor", END)

# 5. 设置入口并编译
workflow.set_entry_point("Triage")
app = workflow.compile()

2.3 分布式状态共享

当需要跨进程协作时，使用 Redis Checkpointer：

from langgraph.checkpoint.redis import RedisSaver
import redis

# 连接 Redis
redis_client = redis.Redis(host="localhost", port=6379, db=0)
checkpointer = RedisSaver(redis_client)

# 编译时传入
app = workflow.compile(checkpointer=checkpointer)

# 使用 thread_id 共享状态
config = {"configurable": {"thread_id": "claim-12345"}}
result = app.invoke(input_data, config=config)

第3章：MCP – Agent 与工具的标准化连接

本章目标: 掌握 MCP 协议，实现 “Write once, run anywhere” 的工具标准化

3.1 MCP 的定位

传统做法：为每个 Agent 框架（LangChain、LlamaIndex、AutoGPT）都写一遍工具代码。

MCP 的愿景：一次编写，所有 Agent 框架可用。

3.2 MCP 的三种原语

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("MyService")

# 1. Resources（资源）：只读数据源
@mcp.resource("user://profile/{user_id}")
def get_user_profile(user_id: str):
return {"name": "Alice", "age": 30}

# 2. Tools（工具）：可执行操作
@mcp.tool()
def send_email(to: str, subject: str, body: str):
"""发送邮件"""
return "Email sent"

# 3. Prompts（提示词模板）：预设提示
@mcp.prompt()
def code_review_prompt(language: str):
return f"Review this {language} code for security issues…"

3.3 实战：构建 SQLite MCP Server

# db_server.py
from mcp.server.fastmcp import FastMCP
import sqlite3

mcp = FastMCP("DatabaseService")

@mcp.tool()
def query_db(sql: str) –> str:
"""执行只读 SQL 查询"""
if not sql.strip().lower().startswith("select"):
return "Error: Only SELECT queries allowed"

conn = sqlite3.connect("claims.db")
cursor = conn.cursor()
cursor.execute(sql)
results = cursor.fetchall()
conn.close()
return str(results)

if __name__ == "__main__":
mcp.run()

客户端调用：

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient({
"database": {
"transport": "stdio",
"command": "python",
"args": ["db_server.py"]
}
})

tools = await client.get_tools() # 自动发现工具

第4章：A2A – Agent 间通信协议

本章目标: 掌握 Google 开源的 A2A 协议，实现跨框架 Agent 互通

4.1 A2A 协议概述

A2A (Agent2Agent) 是 Google 于 2025 年开源并捐赠给 Linux 基金会的协议，旨在解决：

不同 AI 框架构建的 Agent 如何互相通信和协作？

核心价值：

• 互操作性：LangGraph Agent 可以调用 CrewAI Agent
• 安全性：Agent 交互时不暴露内部状态、内存或工具
• 标准化：基于 JSON-RPC 2.0，使用 HTTP(S) 传输

4.2 核心概念

Agent Card：能力声明

每个 A2A Agent 必须发布一个 Agent Card，声明自己的能力：

{
"name": "Travel Planner Agent",
"description": "Helps you plan travel itineraries",
"version": "1.0.0",
"url": "https://travel-agent.example.com/a2a",
"capabilities": {
"streaming": true,
"pushNotifications": false
},
"skills": [
{
"id": "plan_trip",
"name": "Plan Trip",
"description": "Creates a detailed travel itinerary"
}
]
}

4.3 python-a2a 快速上手

pip install python-a2a

创建 A2A Server

from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState

@agent(
name="Weather Agent",
description="Provides weather information",
version="1.0.0"
)
class WeatherAgent(A2AServer):

@skill(
name="Get Weather",
description="Get current weather for a location",
tags=["weather", "forecast"]
)
def get_weather(self, location: str) –> str:
"""获取指定地点的天气"""
return f"It's sunny and 75°F in {location}"

def handle_task(self, task):
"""处理 A2A 任务请求"""
message_data = task.message or {}
content = message_data.get("content", {})
text = content.get("text", "") if isinstance(content, dict) else ""

if "weather" in text.lower():
location = text.split("in", 1)[1].strip().rstrip("?.")
weather_text = self.get_weather(location)
task.artifacts = [{
"parts": [{"type": "text", "text": weather_text}]
}]
task.status = TaskStatus(state=TaskState.COMPLETED)
return task

# 启动服务
if __name__ == "__main__":
agent = WeatherAgent()
run_server(agent, port=5000)

A2A Client 调用

from python_a2a import A2AClient

# 连接远程 Agent
client = A2AClient("http://localhost:5000")

# 获取 Agent Card
card = client.get_agent_card()
print(f"Connected to: {card.name}")

# 发送任务
task = client.send_task({
"message": {
"content": {"text": "What's the weather in Beijing?"}
}
})

print(f"Result: {task.artifacts[0]['parts'][0]['text']}")

4.4 fasta2a：Pydantic 团队的实现

fasta2a 是一个框架无关的 A2A 实现，专注于灵活性：

from fasta2a import FastA2A, Worker
from fasta2a.broker import InMemoryBroker
from fasta2a.storage import InMemoryStorage

class MyWorker(Worker):
async def run_task(self, params):
# 执行任务逻辑
task = await self.storage.get_task(params.id)

# 调用你的 Agent（可以是 LangGraph、CrewAI 等）
result = await my_agent.invoke(task.message)

# 更新任务状态
task.artifacts = [{"parts": [{"type": "text", "text": result}]}]
task.status = {"state": "completed"}
await self.storage.update_task(task)

# 创建服务
storage = InMemoryStorage()
broker = InMemoryBroker()
worker = MyWorker(storage=storage, broker=broker)

app = FastA2A(storage=storage, broker=broker)

4.5 A2A 与 LangGraph 集成

from langgraph.graph import StateGraph
from python_a2a import A2AServer, agent

# 创建 LangGraph 图
workflow = StateGraph(MyState)
# … 添加节点和边 …
langgraph_app = workflow.compile()

@agent(name="LangGraph Agent", version="1.0.0")
class LangGraphA2AAgent(A2AServer):

def handle_task(self, task):
# 将 A2A 任务转换为 LangGraph 输入
input_state = {
"messages": [{"role": "user", "content": task.message["content"]["text"]}]
}

# 调用 LangGraph
result = langgraph_app.invoke(input_state)

# 返回 A2A 格式的结果
task.artifacts = [{
"parts": [{"type": "text", "text": result["messages"][–1].content}]
}]
task.status = {"state": "completed"}
return task

4.6 A2A vs MCP

💡 最佳实践: MCP 用于连接"确定性工具"，A2A 用于连接"智能体服务"。

第5章：AG-UI – Agent 与用户界面交互

本章目标: 掌握 AG-UI 协议，实现 Agent 与前端的实时双向通信

5.1 AG-UI 协议概述

AG-UI (Agent-User Interaction Protocol) 是一个开放的、基于事件的协议，用于标准化 AI Agent 与用户界面之间的连接。

解决的问题：

• Agent 是长时运行的，需要流式传输中间结果
• Agent 具有非确定性，可能动态控制 UI
• 需要支持结构化和非结构化 IO（文本、语音、工具调用）

5.2 AG-UI 事件类型

AG-UI 定义了丰富的事件类型用于双向通信：

// Agent → UI 的事件
interface AgentEvents {
// 文本流式输出
TEXT_MESSAGE_START: { messageId: string }
TEXT_MESSAGE_CONTENT: { messageId: string, delta: string }
TEXT_MESSAGE_END: { messageId: string }

// 工具调用
TOOL_CALL_START: { toolCallId: string, toolName: string }
TOOL_CALL_ARGS: { toolCallId: string, delta: string }
TOOL_CALL_END: { toolCallId: string }

// 状态更新
STATE_SNAPSHOT: { snapshot: object }
STATE_DELTA: { delta: object[] }

// 思考过程
STEP_START: { stepName: string }
STEP_END: { stepName: string }
}

// UI → Agent 的事件
interface UIEvents {
RUN_STARTED: { threadId: string, runId: string }
RUN_CANCELED: { runId: string }
TOOL_RESULT: { toolCallId: string, result: any }
}

5.3 AG-UI 架构

5.4 Python 后端实现

使用 LangGraph 实现 AG-UI 兼容的 Agent：

from typing import AsyncGenerator
import json

async def stream_agent_events(
agent_app,
input_data: dict,
config: dict
) –> AsyncGenerator[str, None]:
"""将 LangGraph 输出转换为 AG-UI 事件流"""

message_id = f"msg_{config['thread_id']}"

# 发送消息开始事件
yield f"data: {json.dumps({'type': 'TEXT_MESSAGE_START', 'messageId': message_id})}\\n\\n"

async for event in agent_app.astream(input_data, config):
if "messages" in event:
for msg in event["messages"]:
if hasattr(msg, "content"):
# 流式输出文本内容
yield f"data: {json.dumps({'type': 'TEXT_MESSAGE_CONTENT', 'messageId': message_id, 'delta': msg.content})}\\n\\n"

if hasattr(msg, "tool_calls") and msg.tool_calls:
for tc in msg.tool_calls:
# 工具调用事件
yield f"data: {json.dumps({'type': 'TOOL_CALL_START', 'toolCallId': tc['id'], 'toolName': tc['name']})}\\n\\n"
yield f"data: {json.dumps({'type': 'TOOL_CALL_ARGS', 'toolCallId': tc['id'], 'delta': json.dumps(tc['args'])})}\\n\\n"
yield f"data: {json.dumps({'type': 'TOOL_CALL_END', 'toolCallId': tc['id']})}\\n\\n"

# 发送消息结束事件
yield f"data: {json.dumps({'type': 'TEXT_MESSAGE_END', 'messageId': message_id})}\\n\\n"

5.5 A2UI：声明式 Generative UI

A2UI 是 Google 提出的声明式 UI 规范，让 Agent 可以返回结构化的 UI 组件：

# Agent 返回 A2UI 格式的响应
a2ui_response = {
"type": "card",
"title": "Flight Options",
"content": [
{
"type": "list",
"items": [
{"text": "Flight 1: Beijing → Shanghai, $200"},
{"text": "Flight 2: Beijing → Shanghai, $180"}
]
},
{
"type": "button",
"label": "Book Now",
"action": "book_flight"
}
]
}

前端框架（如 CopilotKit）会自动将 A2UI JSON 渲染为对应的 UI 组件。

5.6 与 CopilotKit 集成

// 前端代码 (React + CopilotKit)
import { useAgent } from "@copilotkit/react-core";

function ChatInterface() {
const { agent, messages, sendMessage } = useAgent({
url: "http://localhost:8000/agent", // AG-UI 兼容的后端
protocol: "ag-ui"
});

return (
<div>
{messages.map(msg => (
<Message key={msg.id} content={msg.content} />
))}
<input onSubmit={sendMessage} />
</div>
);
}

5.7 三大协议协同工作

第6章：微服务化与生产部署

本章目标: 掌握 LangServe 微服务化和工程最佳实践

6.1 LangServe 微服务

from fastapi import FastAPI
from langserve import add_routes

api = FastAPI(title="Insurance Claims API")

add_routes(
api,
swarm_app,
path="/claims",
enabled_endpoints=["invoke", "stream"]
)

自动生成的端点：

• POST /claims/invoke：同步调用
• POST /claims/stream：流式输出（SSE）
• GET /claims/playground：可视化调试

6.2 技术栈对比

6.3 工程最佳实践

死循环检测

def router_with_ttl(state):
hop_count = state.get("_hop_count", 0)
if hop_count > 20:
return END # 强制终止
state["_hop_count"] = hop_count + 1
return normal_router(state)

分布式追踪

from langsmith import traceable

@traceable(run_type="agent")
def triage_node(state):
# LangSmith 自动记录
...

安全考虑

# Agent 权限隔离
triage_tools = await client.get_tools(server="readonly_db")
fraud_tools = await client.get_tools(server="full_access_db")

结语：协议即架构

从 Swarm 的进程内协作到 A2A 的跨框架通信，Agent 架构正在经历从"单体应用"到"分布式生态"的演进。

三大协议的定位：

• MCP：Agent 的"USB 接口"，标准化工具连接
• A2A：Agent 的"HTTP 协议"，实现互联网级互通
• AG-UI：Agent 的"显示器接口"，标准化人机交互

设计原则：

• 工具连接用 MCP
• Agent 协作用 A2A（跨框架）或 Swarm（同框架）
• 用户交互用 AG-UI

“Protocols are the new APIs.”

— 在 Agent 时代，标准协议比私有 API 更有价值。

参考资料：

• A2A Protocol Specification
• Model Context Protocol
• AG-UI Documentation
• LangGraph Multi-Agent
• python-a2a PyPI
• fasta2a GitHub