软件正在进入智能体时代。经过多年AI作为高级自动补全工具的发展——建议代码、总结文本、回答问题——我们正见证着一个向能够自主规划、执行和适应的系统的根本性转变。智能体工作流,即AI智能体以最少的人工干预协调复杂的多步骤任务,代表了应用人工智能的下一个前沿领域。
但构建可靠的多智能体系统比构建聊天机器人要困难得多。协调、错误处理、状态管理和信任方面的挑战以需要新架构思维的方式累积。本文探讨了在生产环境中构建智能体系统的团队所出现的模式、框架和来之不易的经验教训。
什么使工作流具有智能体特性
“智能体”这个术语被随意使用,因此精确性很重要。智能体工作流有几个定义特征,使其区别于传统自动化或简单的AI集成:
- 目标导向行为:智能体追求高层次目标,而不是执行固定脚本
- 工具使用:智能体可以调用外部工具、API和服务来实现其目标
- 规划和推理:智能体将复杂任务分解为子任务并确定执行顺序
- 适应能力:智能体根据中间结果和错误调整其方法
- 记忆能力:智能体在步骤之间以及跨会话保持上下文
一个回答问题的简单聊天机器人不是智能体。一个接收错误报告、搜索代码库中的相关文件、生成修复方案、编写测试、打开拉取请求并回应审查评论的系统——这才是智能体。
单智能体与多智能体架构
第一个架构决策是使用单个智能体还是多个协作智能体。每种方法都有不同的权衡取舍。
单智能体系统
单智能体系统使用一个具有多种工具访问权限的 LLM 实例。它在统一循环中规划和执行所有步骤。这更容易构建、调试和推理。对于大多数用例,设计良好的单智能体系统优于多智能体系统。
# 使用工具的单智能体模式
from anthropic import Anthropic
client = Anthropic()
tools = [
{"name": "search_codebase", "description": "搜索代码模式"},
{"name": "read_file", "description": "从仓库中读取文件"},
{"name": "write_file", "description": "将内容写入文件"},
{"name": "run_tests", "description": "执行测试套件"},
{"name": "create_pr", "description": "创建拉取请求"},
]
def agent_loop(task: str):
messages = [{"role": "user", "content": task}]
while True:
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=4096,
tools=tools,
messages=messages,
)
if response.stop_reason == "end_turn":
return extract_text(response)
# 处理工具调用
tool_results = []
for block in response.content:
if block.type == "tool_use":
result = execute_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result,
})
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
多智能体系统
多智能体系统使用多个协作完成任务的专门智能体。每个智能体都有专注的角色——一个可能是研究员,另一个是编码员,另一个是审查员。当任务确实需要不同的专业知识领域,或者出于安全原因需要关注点分离时,多智能体架构才有意义。
多智能体协调的开销——消息传递、共享状态、冲突解决——是相当大的。在你有证据表明单个智能体无法有效处理任务之前,不要采用多智能体架构。
编排模式
智能体如何协调工作是核心设计挑战。已经出现了三种主要的编排模式。
顺序(流水线)
代理按固定顺序执行,每个代理接收前一个代理的输出。这是最简单的模式,当任务具有自然的线性流程时效果很好。
# 顺序编排
async def sequential_pipeline(task):
# 步骤1:研究代理收集上下文
context = await research_agent.run(task)
# 步骤2:规划代理创建实施计划
plan = await planning_agent.run(task, context)
# 步骤3:编码代理实施计划
code = await coding_agent.run(plan)
# 步骤4:审查代理验证输出
review = await review_agent.run(code, plan)
if review.needs_revision:
code = await coding_agent.run(review.feedback)
return code
并行(扇出/扇入)
多个代理同时处理独立的子任务,并聚合它们的结果。当任务可以分解为独立部分时,这是理想的选择——例如,同时研究多个主题或针对多个环境进行测试。
# 并行编排
async def parallel_research(topics: list[str]):
# 扇出:多个代理同时研究
tasks = [research_agent.run(topic) for topic in topics]
results = await asyncio.gather(*tasks)
# 扇入:综合代理合并发现结果
synthesis = await synthesis_agent.run(results)
return synthesis
分层(管理器/工作器)
管理器代理分解任务,将子任务委托给工作器代理,评估它们的输出,并决定下一步。这是最灵活但也是最复杂的模式。管理器代理需要强大的规划能力和判断力,以知道工作器输出何时令人满意。
# 分层编排
async def hierarchical_workflow(objective: str):
manager = ManagerAgent(objective)
while not manager.is_complete():
# 管理器决定下一个子任务并分配
assignment = await manager.plan_next_step()
# 委托给适当的工作器
worker = get_worker(assignment.agent_type)
result = await worker.execute(assignment.task)
# 管理器评估并决定下一个行动
await manager.evaluate_result(result)
return manager.get_final_output()
工具使用和函数调用
工具是将语言模型从文本生成器转变为代理的关键。工具界面的设计是构建代理系统中最具决定性的决策之一。
良好工具设计的原则
原子操作:每个工具应该做好一件事。读取文件的工具不应同时解析它。可组合性来自于组合原子工具,而不是构建瑞士军刀式的工具。
明确的契约:工具描述应精确说明输入、输出和副作用。模糊的工具描述会导致误用和错误。
尽可能实现幂等性:可以安全重试的工具能极大地简化错误处理。当工具调用在执行过程中失败时,代理需要知道重试是否安全。
有限范围:工具应有限制条件。文件写入工具应限制在特定目录,数据库工具只能访问特定表。最小权限原则同样适用于代理,就像适用于人类用户一样。
# 设计良好的工具定义
tools = [
{
"name": "query_database",
"description": "对分析数据库执行只读SQL查询。只允许SELECT语句。结果限制在1000行以内。",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "要执行的SQL SELECT查询"
},
"timeout_seconds": {
"type": "integer",
"default": 30,
"description": "查询超时时间(秒)(最大60)"
}
},
"required": ["query"]
}
}
]
错误处理与恢复
错误处理是决定代理系统成败的关键。与传统软件遵循可预测的错误模式不同,代理系统面临的是故障模式的组合爆炸:工具故障、格式错误的输出、幻觉行为、无限循环、上下文窗口耗尽等等。
基本错误处理模式
带退避的重试:对于瞬时故障——网络超时、速率限制、临时服务不可用——应自动重试。指数退避可以防止级联故障。
备用策略:当主要方法失败时,代理应备有替代策略。如果代码搜索工具没有返回结果,代理可以尝试更广泛的搜索或读取目录结构来手动查找相关文件。
断路器:在相同类型的故障重复发生后,代理应停止重试,要么升级到人工处理,要么尝试根本不同的方法。无限重试相同失败的API调用的代理比没有用的更糟。
上下文管理: 当代理处理复杂任务时,它们会积累上下文。如果没有主动管理,它们将耗尽上下文窗口。有效的代理会总结中间结果并修剪无关的上下文。
# 使用断路器模式进行错误处理
class AgentExecutor:
def __init__(self, max_retries=3, max_consecutive_failures=5):
self.consecutive_failures = 0
self.max_consecutive_failures = max_consecutive_failures
async def execute_with_recovery(self, agent, task):
for attempt in range(self.max_retries):
try:
result = await agent.run(task)
self.consecutive_failures = 0
return result
except ToolExecutionError as e:
self.consecutive_failures += 1
if self.consecutive_failures >= self.max_consecutive_failures:
raise CircuitBreakerOpen(
f"连续失败次数过多: {e}"
)
await asyncio.sleep(2 ** attempt) # 指数退避
except ContextOverflowError:
# 总结并压缩上下文,然后重试
agent.compress_context()
continue
raise MaxRetriesExceeded(f"在 {self.max_retries} 次尝试后失败")
构建代理系统的框架
已经出现了多个框架来简化代理工作流程的开发。每个框架都有不同的权衡取舍。
LangGraph
LangGraph 由 LangChain 团队开发,将代理工作流程建模为有向图。节点表示处理步骤(LLM 调用、工具执行、条件逻辑),边定义它们之间的流程。LangGraph 在处理具有分支、循环和人机交互的复杂工作流程方面表现出色。它的状态管理非常出色,并提供了对持久化和流处理的内置支持。
CrewAI
CrewAI 专注于多代理协作。您可以定义具有特定角色、背景故事和目标的代理,然后将它们组装成在任务上一起工作的团队。CrewAI 处理代理之间的协调和通信。它特别适合那些自然映射到团队隐喻的工作流程——研究团队、内容创作团队、代码审查团队。
Anthropic Agent SDK
Anthropic Agent SDK 采用了刻意简约的方法。它没有提供一个沉重的框架,而是为构建代理提供了轻量级的原语:工具注册、代理循环、代理之间的交接以及护栏。其理念是,代理系统中的大部分复杂性是特定领域的,框架应该提供构建块而非主观意见。
# Anthropic Agent SDK 示例
from agents import Agent, Runner, function_tool
@function_tool
def search_web(query: str) -> str:
"""在网上搜索信息。"""
return web_search_api(query)
@function_tool
def write_report(title: str, content: str) -> str:
"""编写结构化报告。"""
return save_report(title, content)
researcher = Agent(
name="研究员",
instructions="您使用网络搜索彻底研究主题。",
tools=[search_web],
)
writer = Agent(
name="作者",
instructions="您编写清晰、结构良好的报告。",
tools=[write_report],
handoffs=[researcher], # 可以委托回研究员
)
result = Runner.run(writer, "编写一份关于 2026 年 WASM 采用情况的报告")
生产环境考量
将代理系统从原型迁移到生产环境会暴露出容易被低估的挑战。
可观测性
无法观测就无法操作。每个代理行动——每次 LLM 调用、工具调用、决策点和错误——都需要被记录和追踪。如果没有全面的可观测性,在多步骤代理工作流中调试生产问题几乎是不可能的。像 LangSmith、Braintrust 和 Arize Phoenix 这样的工具为代理系统提供专门的可观测性。
成本管理
代理工作流可能很昂贵。单个任务可能涉及数十次 LLM 调用,每次都会消耗 token。成本随复杂性而增加,失控的代理可能会迅速耗尽 API 预算。对每个任务的迭代次数、token 消耗和工具调用实施硬性限制。
延迟
代理工作流中的每个步骤都会增加延迟。一个五步代理循环,包含两秒的 LLM 调用和一秒的工具执行,至少需要 15 秒。对于面向用户的应用程序,这种延迟通常是不可接受的。策略包括并行化独立步骤、缓存常用工具结果,以及对简单决策使用更快的模型。
安全和护栏
能够采取现实世界行动的代理需要护栏。至少,对高影响操作(发送电子邮件、进行购买、修改生产数据)实施审批门、输出验证以捕获幻觉或格式错误的操作、工具调用的速率限制,以及代码执行的沙箱化。最危险的故障模式是代理大规模自信地采取错误行动。
何时使用代理工作流
并非每个问题都需要代理解决方案。代理工作流会增加复杂性、成本、延迟和不可预测性。当任务确实需要多步推理和中间决策时,当工作流程无法完全预先指定时,当任务需要与多个外部系统交互时,以及在决策点需要人类级别的判断(而不仅仅是模式匹配)时,代理解决方案才有意义。
如果一个任务可以通过一次 LLM 调用、一个简单的链或传统自动化来解决,那么这些更简单的方法几乎总是更可取。
展望未来
代理范式仍处于早期阶段。框架正在迅速成熟,模型在规划和工具使用方面变得越来越好,可靠性的工程模式也在不断巩固。但我们远未达到可以向代理模糊地给出目标并期望获得可靠结果的阶段。
构建最成功的代理系统的团队有一个共同特点:他们非常务实。他们从狭窄、定义明确的任务开始。他们在可观测性和测试方面投入大量资源。他们逐步建立信任,只有在证明可靠性后才扩展代理的权限。他们永远不会忘记代理是为人类目标服务的工具——而不是目的本身。
代理工作流的兴起不是为了取代人类判断,而是为了扩展它——让工程师和组织能够自动化那些以前无法委托给软件的复杂认知任务。学会构建可靠代理系统的组织将拥有显著的竞争优势。关键词是可靠。
