AI Agent 工程实践全解：从 ReAct 规划到多智能体协作落地
什么是 AI Agent？从概念到工程落地
近年来，AI Agent 从学术概念迅速走向工程实践。不同于传统的问答模型，Agent 具备感知—规划—执行的闭环能力：它能接收外部环境信息、利用 LLM 进行推理规划，并通过调用工具完成真实世界的任务。
典型的 Agent 架构由以下几部分组成：
大模型核心（LLM Brain）：负责理解用户意图、制定行动计划，常用 GPT-4o、Claude 3.5、Qwen2.5 等。
工具集（Tool Set）：函数调用接口，包括搜索引擎、代码执行器、数据库查询、HTTP 请求等。
记忆模块（Memory）：短期记忆（对话上下文）+ 长期记忆（向量数据库），让 Agent 跨会话保持状态。
规划器（Planner）：ReAct、CoT、Tree-of-Thought 等推理范式，指导 LLM 分步推导。
执行器（Executor）：将规划产出的 Action 实际调度到工具或子 Agent。
理解这套架构，是构建生产级 AI Agent 系统的第一步。
ReAct 与 Plan-and-Execute：两种主流规划范式对比
目前工程实践中最常见的两种 Agent 规划模式是 ReAct 和 Plan-and-Execute，两者各有适用场景。
ReAct（Reasoning + Acting）
ReAct 将推理与行动交织进行，每一步都输出"思考→行动→观察"三元组，直到得出最终答案。优点是灵活、能根据中间结果实时调整策略；缺点是长任务下容易迷失方向，Token 消耗大。
Thought: 用户要查北京今天的天气，需要调用天气工具
Action: get_weather(city="北京")
Observation: {"temp": 32, "weather": "晴", "humidity": 45}
Thought: 已获取天气数据，可以回答用户
Answer: 北京今天晴天，气温32°C，湿度45%，适合外出。Plan-and-Execute
先由 Planner LLM 生成完整执行计划（有序步骤列表），再由 Executor 依次调用工具执行。优点是对复杂多步任务结构清晰、便于并发；缺点是计划生成后难以动态修正。
实际项目中，可以结合两者：先做高层规划，每一步内部用 ReAct 灵活执行，这也是 LangGraph、AutoGen 等框架的设计思路。
工具调用（Function Calling）的工程实现
Function Calling 是 Agent 与外部系统交互的核心机制。主流模型（OpenAI、Anthropic、Google）均支持结构化工具调用，本质是让 LLM 输出一段 JSON，由宿主程序解析并执行对应函数，再将结果回填上下文。
下面以 OpenAI SDK 为例，展示一个天气查询工具的完整注册与调用流程：
from openai import OpenAI
import json
client = OpenAI()
# 1. 定义工具 Schema
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的实时天气",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称，如北京、上海"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"], "default": "celsius"}
},
"required": ["city"]
}
}
}
]
# 2. 模拟工具实现
def get_weather(city: str, unit: str = "celsius") -> dict:
# 实际项目中调用天气 API
return {"city": city, "temp": 32, "weather": "晴", "unit": unit}
# 3. Agent 循环
messages = [{"role": "user", "content": "北京今天天气怎么样？"}]
while True:
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto"
)
msg = response.choices[0].message
if msg.tool_calls:
messages.append(msg)
for call in msg.tool_calls:
args = json.loads(call.function.arguments)
result = get_weather(**args)
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": json.dumps(result, ensure_ascii=False)
})
else:
print("最终回答:", msg.content)
break工程要点：工具 Schema 描述要精确（影响 LLM 调用准确率）、工具执行要有超时与异常捕获、敏感工具需加人工确认（Human-in-the-Loop）。
长期记忆与向量数据库集成
生产级 Agent 往往需要跨会话记忆能力。实现方式是将历史对话、用户偏好、任务结果等信息向量化后存入向量数据库（如 Chroma、Qdrant、Milvus），在新会话中检索相关片段注入上下文。
以下是基于 LangChain + Chroma 的简单记忆模块示例：
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.memory import VectorStoreRetrieverMemory
# 初始化向量数据库
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
collection_name="agent_memory",
embedding_function=embeddings,
persist_directory="./chroma_db"
)
# 创建检索式记忆
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
memory = VectorStoreRetrieverMemory(retriever=retriever)
# 存储记忆
memory.save_context(
{"input": "我的名字是张三，我是一名 Python 开发者"},
{"output": "好的，我记住了，张三，Python 开发者"}
)
# 检索相关记忆
relevant = memory.load_memory_variables({"prompt": "用户是做什么的？"})
print(relevant)
# {'history': '用户：我的名字是张三，我是一名 Python 开发者\nAI：好的，我记住了...'}在高并发场景下，还需考虑记忆的写入队列化（防止并发冲突）、定期压缩合并（控制向量库规模）以及隐私脱敏（避免敏感信息持久化）。
多 Agent 协作：从单体到分布式 Agent 网络
复杂任务通常需要多个专业 Agent 协同完成。多 Agent 架构有以下几种常见拓扑：
主从架构（Orchestrator-Worker）：一个主 Agent 负责任务分解和结果汇总，多个子 Agent 各司其职（如搜索 Agent、代码 Agent、写作 Agent）。
流水线架构（Pipeline）：Agent 之间串行传递中间结果，适合有明确顺序依赖的任务流。
辩论架构（Debate）：多个 Agent 对同一问题给出不同视角的答案，最终由裁判 Agent 综合评判，可显著提升答案质量。
群体智能（Swarm）：受蚁群算法启发，多个轻量 Agent 并行探索，共享知识图谱，适合大规模信息收集场景。
以 LangGraph 为例，多 Agent 工作流可通过有向图（DAG）描述节点间的数据流和控制流，支持条件分支、并发执行和人工干预节点，是目前工程化程度最高的多 Agent 框架之一。
from langgraph.graph import StateGraph, END
from typing import TypedDict
class AgentState(TypedDict):
task: str
search_result: str
final_answer: str
def search_agent(state: AgentState) -> AgentState:
# 执行搜索
state["search_result"] = f"搜索结果：关于 {state['task']} 的相关信息..."
return state
def writer_agent(state: AgentState) -> AgentState:
# 基于搜索结果生成报告
state["final_answer"] = f"综合报告：{state['search_result']}"
return state
# 构建 Agent 流水线
builder = StateGraph(AgentState)
builder.add_node("searcher", search_agent)
builder.add_node("writer", writer_agent)
builder.set_entry_point("searcher")
builder.add_edge("searcher", "writer")
builder.add_edge("writer", END)
graph = builder.compile()
result = graph.invoke({"task": "2025年AI Agent发展趋势"})Agent 的评估与可观测性
Agent 系统的评估远比传统 ML 模型复杂，因为其行为是动态的、多步的、与环境交互的。工程实践中常用的评估维度包括：
任务完成率：Agent 成功完成目标任务的比例，需构建多样化的评测集。
步骤效率：达成目标所需的平均工具调用次数，步骤越少越好。
幻觉率：Agent 伪造工具调用参数或捏造事实的频率。
成本控制：每次任务的 Token 消耗和工具调用费用。
可观测性方面，推荐接入 LangSmith、Langfuse 或自建 OpenTelemetry 采集链路，对每次 Agent 运行的 Trace（推理链、工具调用、耗时、Token 数）进行结构化记录，便于问题排查和效果迭代。
# 使用 Langfuse 追踪 Agent 调用链
from langfuse import Langfuse
from langfuse.openai import openai # 自动 patch OpenAI SDK
lf = Langfuse(
public_key="your-public-key",
secret_key="your-secret-key",
host="https://cloud.langfuse.com"
)
# 创建追踪会话
trace = lf.trace(name="weather-agent-run", user_id="user_123")
span = trace.span(name="tool-call-get_weather")
# ... 执行 Agent 逻辑 ...
span.end(output={"result": "北京32°C晴"})
trace.update(output="任务完成")生产部署注意事项与最佳实践
将 Agent 推向生产环境时，有几个关键工程问题需要重视：
超时与熔断：为每个工具调用设置最大超时（建议 10-30s），并实现熔断机制，防止某个工具失败导致整个 Agent 卡死。
重试策略：LLM API 限流（429）时使用指数退避重试，工具调用失败时允许 Agent 自主换用备选工具。
安全沙箱：代码执行类工具必须运行在隔离容器（Docker/gVisor）中，限制网络访问和文件系统权限。
Human-in-the-Loop：高风险操作（发送邮件、修改数据库、执行支付）前强制引入人工确认节点。
成本上限：为每次 Agent 运行设置 Token 预算上限，超出后自动降级到更简单的策略或直接返回兜底答案。
幂等性设计：工具调用尽量设计为幂等，避免网络重试导致重复操作（如重复下单、重复发消息）。
AI Agent 工程化之路道阻且长，但随着 LangGraph、AutoGen、CrewAI 等框架的成熟，以及大模型 Function Calling 能力的持续提升，2025-2026 年将是 Agent 从 Demo 走向规模化落地的关键窗口期。持续关注 Agent 的可靠性、可控性和可解释性，是每一位 AI 工程师的核心命题。
发布评论
热门评论区：