SingleAgent vs MultiAgent
单体Agent瓶颈：
| 瓶颈 | 表现 |
|---|---|
| 上下文窗口有限 | 单次对话塞不下大仓库、长文档、多轮工具调用记录，上下文被无关信息污染 |
| 角色混淆 | 一个 Agent 既要写代码又要审查又要部署，提示词越来越长，行为越来越不稳定 |
| 串行低效 | 任务只能一个接一个做，无法并行读文件、跑实验、查资料 |
| 单点失败 | 一个环节出错（工具报错、格式错乱）整个任务从头再来 |
| 不可组合 | 能力写死在提示词里，难以复用、难以扩展，维护成本高 |
MultiAgent解决了什么：
- 上下文隔离，关注点分离：每个子Agent只能看见自己那部分上下文，例如审查Agent只看diff
- 并行执行：多个SubAgent同时查询资料
- 任务分解与递归：大任务拆解给子Agent，子Agent还能继续拆解任务
- 失败隔离与重试：SubAgent失败后不影响全局，主Agent可以重启它
- 长期记忆/状态管理：Agent之间通过共享工作区的文件交换产物
实现架构
router架构
核心思想：把”决策”从”执行”中抽离出来。一个路由模块（可能是 LLM、规则或向量检索）决定请求交给哪个下游处理单元。
class MultiAgentState(TypedDict):
task: str
next_agent: str
file_report: str
code_report: str
final_answer: str
def main() -> None:
task = " ".join(sys.argv[1:]).strip() or DEFAULT_TASK
reset_workspace()
print("=== 02. MultiAgent：每个 Agent 是一个 Node，用 conditional_edge 路由 ===")
print("\n用户任务:")
print(task)
print("\n运行前 workspace:")
print(show_workspace())
llm = load_llm()
file_agent = create_agent(
llm,
tools=[list_files, move_file],
system_prompt=FILE_AGENT_PROMPT,
name="file_agent",
)
code_agent = create_agent(
llm,
tools=[write_file],
system_prompt=CODE_AGENT_PROMPT,
name="code_agent",
)
def supervisor_node(state: MultiAgentState) -> MultiAgentState:
print("\n[supervisor] 决定下一个 agent")
if not state["file_report"]:
next_agent = "file_agent"
elif not state["code_report"]:
next_agent = "code_agent"
else:
next_agent = "finish"
decision = llm.invoke(
[
SystemMessage(content=SUPERVISOR_PROMPT),
HumanMessage(
content=(
f"用户任务：{state['task']}\n\n"
f"file_report：{state['file_report'] or '(empty)'}\n\n"
f"code_report：{state['code_report'] or '(empty)'}\n\n"
f"请判断下一个 agent。建议答案：{next_agent}"
)
),
]
).content
decision = str(decision).strip()
if decision not in {"file_agent", "code_agent", "finish"}:
decision = next_agent
print(decision)
if decision == "finish":
final_answer = llm.invoke(
[
SystemMessage(content=SUMMARY_PROMPT),
HumanMessage(
content=(
f"file_agent:\n{state['file_report']}\n\n"
f"code_agent:\n{state['code_report']}"
)
),
]
).content
return {**state, "next_agent": "finish", "final_answer": str(final_answer)}
return {**state, "next_agent": decision}
def file_agent_node(state: MultiAgentState) -> MultiAgentState:
print("\n[file_agent node] 调用真正的 file_agent")
result = file_agent.invoke(
{
"messages": [
{
"role": "user",
"content": (
f"用户原始任务：{state['task']}\n\n"
"你是 file_agent，只处理其中和文件整理有关的部分。"
),
}
]
}
)
report = last_message_text(result)
print(report)
return {**state, "file_report": report}
def code_agent_node(state: MultiAgentState) -> MultiAgentState:
print("\n[code_agent node] 调用真正的 code_agent")
result = code_agent.invoke(
{
"messages": [
{
"role": "user",
"content": (
f"用户原始任务：{state['task']}\n\n"
f"file_agent 已完成的结果：\n{state['file_report']}\n\n"
"你是 code_agent，只处理其中和代码生成有关的部分。"
),
}
]
}
)
report = last_message_text(result)
print(report)
return {**state, "code_report": report}
def route_next(state: MultiAgentState) -> str:
if state["next_agent"] == "file_agent":
return "file_agent"
if state["next_agent"] == "code_agent":
return "code_agent"
return END
graph = StateGraph(MultiAgentState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("file_agent", file_agent_node)
graph.add_node("code_agent", code_agent_node)
graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route_next)
graph.add_edge("file_agent", "supervisor")
graph.add_edge("code_agent", "supervisor")
result = graph.compile().invoke(
{
"task": task,
"next_agent": "",
"file_report": "",
"code_report": "",
"final_answer": "",
}
)
print("\n[supervisor] 最终回答:")
print(result["final_answer"])
print("\n运行后 workspace:")
print(show_workspace())
Agent-as-Tool架构
核心思想：主 Agent 的工具循环（tool loop）里，子 Agent 被注册为一个普通 tool——和 search_web、read_file 平级。LLM 自己决定”什么时候调用这个工具”。
FILE_AGENT_PROMPT = """
你是 file_agent，只负责文件整理。
你可以使用 list_files 和 move_file。
目标：先查看 workspace，再把 inbox/a.txt 移动到 archive/a.txt，最后汇报目录变化。
不要写代码。
""".strip()
CODE_AGENT_PROMPT = """
你是 code_agent，只负责生成代码文件。
你可以使用 write_file。
请创建 check_archive.py，代码用于检查 archive/a.txt 是否存在。
不要移动文件。
""".strip()
SUPERVISOR_PROMPT = """
你是 supervisor agent，负责协调两个 specialist agent。
你不能直接整理文件，也不能直接写文件。
你只能调用这两个工具：
1. call_file_agent：让 file_agent 整理文件。
2. call_code_agent：让 code_agent 生成代码。
必须先调用 call_file_agent，再调用 call_code_agent，最后用中文总结两个 agent 的结果。
""".strip()
def last_message_text(result: dict) -> str:
return str(result["messages"][-1].content)
def main() -> None:
task = " ".join(sys.argv[1:]).strip() or DEFAULT_TASK
reset_workspace()
print("=== 01. MultiAgent：Supervisor 把 Sub-Agent 当作 Tool ===")
print("\n用户任务:")
print(task)
print("\n运行前 workspace:")
print(show_workspace())
llm = load_llm()
file_agent = create_agent(
llm,
tools=[list_files, move_file],
system_prompt=FILE_AGENT_PROMPT,
name="file_agent",
)
code_agent = create_agent(
llm,
tools=[write_file],
system_prompt=CODE_AGENT_PROMPT,
name="code_agent",
)
@tool
def call_file_agent(instruction: str) -> str:
"""Delegate a file-management task to file_agent."""
print("\n[supervisor -> file_agent]")
print(instruction)
result = file_agent.invoke({"messages": [{"role": "user", "content": instruction}]})
answer = last_message_text(result)
print("\n[file_agent -> supervisor]")
print(answer)
return answer
@tool
def call_code_agent(instruction: str) -> str:
"""Delegate a code-generation task to code_agent."""
print("\n[supervisor -> code_agent]")
print(instruction)
result = code_agent.invoke({"messages": [{"role": "user", "content": instruction}]})
answer = last_message_text(result)
print("\n[code_agent -> supervisor]")
print(answer)
return answer
supervisor = create_agent(
llm,
tools=[call_file_agent, call_code_agent],
system_prompt=SUPERVISOR_PROMPT,
name="supervisor",
)
result = supervisor.invoke({"messages": [{"role": "user", "content": task}]})
print("\n[supervisor] 最终回答:")
print(last_message_text(result))
print("\n运行后 workspace:")
print(show_workspace())
评论（0）
暂无评论