对比单 Agent 的局限,展示规划、研究、执行、批评等专业化智能体分工模式。涵盖 LangGraph/CrewAI 等框架对比和实战代码。
单 Agent 聊天机器人很快就会遇到瓶颈——让一个模型同时负责规划、研究、写作和验证,结果往往是四件事都做不好。本文将拆解一套真正可用的多 Agent 架构:由多个专业 Agent 相互交接工作,并给出背后的实际编排代码。
试图通过单次 LLM 调用完成所有工作,会遇到一些意料之中的问题:
将工作拆分给多个专业 Agent——Planner、Researcher、Executor 和 Critic——可以解决这三个问题,因为每个 Agent 只会看到自己真正需要的信息。
在从头构建自己的编排层之前,值得先了解一下市面上已有的方案。在决定完全自研之前,可以通过软件聚合平台并排比较 LangGraph、CrewAI、AutoGen 等 Agent 框架。
每个 Agent 只负责一项工作,并配备范围严格限定的 system prompt——任何 Agent 都不应该试图包办一切。
AGENTS = {
"planner": {
"role": "Break the user's goal into a numbered list of concrete subtasks.",
"model": "gpt-4o-mini"
},
"researcher": {
"role": "Given a subtask, gather relevant facts using available tools. Do not write final answers.",
"model": "gpt-4o-mini"
},
"executor": {
"role": "Given research and a subtask, produce the final output for that subtask.",
"model": "gpt-4o"
},
"critic": {
"role": "Review the executor's output against the original goal. Approve or request revisions.",
"model": "gpt-4o-mini"
}
}
from openai import OpenAI
client = OpenAI()
def run_agent(role, user_input, context=""):
system_prompt = AGENTS[role]["role"]
response = client.chat.completions.create(
model=AGENTS[role]["model"],
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"{context}\n\nTask: {user_input}"}
]
)
return response.choices[0].message.content
def plan(goal):
output = run_agent("planner", goal)
steps = [line.strip() for line in output.split("\n") if line.strip()]
return steps
steps = plan("Write a technical comparison of vector databases for a RAG pipeline")
print(steps)
Researcher 负责收集事实,之后才由 Executor 将这些事实整理成最终答案——让“查找信息”和“撰写答案”各自成为独立的关注点。
def research(subtask):
return run_agent("researcher", subtask)
def execute(subtask, research_notes):
context = f"Research notes:\n{research_notes}"
return run_agent("executor", subtask, context)
results = []
for step in steps:
notes = research(step)
draft = execute(step, notes)
results.append({"step": step, "draft": draft})
这是大多数演示都会跳过的步骤,却也是最重要的一步。如果没有验证环节,Agent 自信满满地交付错误答案的概率,与交付正确答案一样高。
def critique(goal, draft):
context = f"Original goal: {goal}"
verdict = run_agent("critic", draft, context)
return "approve" in verdict.lower()
def refine_until_approved(goal, step, notes, max_retries=2):
draft = execute(step, notes)
for attempt in range(max_retries):
if critique(goal, draft):
return draft
notes += "\nPrevious attempt was rejected. Improve accuracy and completeness."
draft = execute(step, notes)
return draft # return best effort after max retries
使用一个简单的状态机将所有环节串联起来——要让整个流程端到端运行,并不需要引入重量级框架。
def run_workflow(goal):
steps = plan(goal)
final_output = []
for step in steps:
notes = research(step)
approved_draft = refine_until_approved(goal, step, notes)
final_output.append(approved_draft)
return "\n\n".join(final_output)
result = run_workflow("Write a technical comparison of vector databases for a RAG pipeline")
print(result)
每个 Agent 只会看到与自身工作相关的那一部分上下文——Planner 永远看不到研究笔记,Researcher 也永远看不到 Critic 的反馈。正是这种隔离机制,避免了多 Agent 系统退化成一个塞满混乱信息的巨型上下文窗口。
Agent 工作流的失败方式与单次调用应用不同——Agent 可能陷入循环、幻觉出一个工具调用,或者直接超时。务必限制重试次数,并记录每一次交接。
import logging
logging.basicConfig(level=logging.INFO)
def run_agent_safe(role, user_input, context="", timeout_retries=2):
for attempt in range(timeout_retries):
try:
output = run_agent(role, user_input, context)
logging.info(f"[{role}] succeeded on attempt {attempt + 1}")
return output
except Exception as e:
logging.warning(f"[{role}] failed: {e}")
raise RuntimeError(f"Agent '{role}' failed after {timeout_retries} attempts")
每项任务运行四个 Agent,而不是只调用一次模型,token 成本很快就会累积起来。不过,技术栈中的每个部分并不都需要使用付费版本——大量编排框架、链路追踪工具和评估库都是免费且开源的。在为商业 Agent 监控平台付费之前,先查看免费软件替代方案列表;LangSmith 的免费版本或开源链路追踪工具,通常已经能够满足小型团队的实际需求。
一个华而不实的 Agent 演示,与一套能够可靠完成真实任务的系统之间,差别归根结底在于结构:清晰的角色分工、先研究后执行的交接机制,以及能在错误交付前将其发现的 Critic 循环。只要缺少这三者中的任何一个,整个系统就会退化成一个困惑的单 Agent,只能一路猜测着完成任务。
Agent 框架和底层模型变化很快——上个月还能正常运行的方案,现在可能已经被弃用。在将多 Agent 系统部署到生产环境之前,请通过软件版本更新网站检查最新信息,确保使用的框架和模型版本仍然有效。
如需采取进一步措施,你可以考虑屏蔽此人和/或举报滥用行为。