AI 多智能体协作：从单 Agent 到多 Agent 工作流

一个 Agent 干不了的活，就拆给多个 Agent 干。多智能体就是把任务分工，让每个 Agent 专精一件事：一个写代码、一个审查、一个跑测试，配合起来比单打独斗强得多。这篇把多智能体讲透：基础、架构模式、通信机制、实战、框架选型、状态与容错、成本与性能、评估、安全。

一、多智能体基础：Agent 由什么构成

一个 Agent 由四部分构成：模型（大脑，负责推理和决策）、工具（手，负责执行操作）、记忆（短期记忆是当前对话，长期记忆是知识库和历史记录）、规划（把大任务拆成小步骤）。多智能体就是多个这样的 Agent 协作，每个 Agent 有自己的角色和工具，通过消息传递配合。

• 什么时候需要多 Agent：任务太复杂，一个提示词装不下；需要分工，写代码和审查代码是两种思维；需要校验，关键操作要第二双眼睛复核。
• 什么时候不需要：简单问答、单步操作，单 Agent 更快更便宜，别为了多而多。
• 判断标准：单 Agent 跑通之后，确实遇到瓶颈（准确率上不去、任务太复杂、需要校验），再拆。先单后多，别一上来就搭五六个 Agent 的架构。

二、架构模式：五种协作方式

三、通信机制：Agent 之间怎么说话

• 共享上下文：各 Agent 看到的任务背景要一致，不然一个以为在测试环境，一个以为在生产环境。任务背景写进每个 Agent 的系统提示词里，别只告诉管理者。
• 消息格式：上游输出要结构化（JSON），下游才能稳定解析，别传自然语言。自然语言看着方便，解析起来全是坑。
• 冲突解决：两个 Agent 结论不一致时，定个规则：高风险操作听审查的，低风险操作听执行的，拿不准就问人。规则要事先定好，别等冲突了再吵。
• 状态管理：长流程要把中间结果存下来，挂了能从断点续跑，别从头再来。状态存数据库或者文件，别只放内存里。

四、实战：写代码 Agent + 审查 Agent

# 两个 Agent 协作写脚本（示意，伪代码）
def llm_call(system: str, user: str) -> str:
"""调模型的封装，实际用 OpenAI 客户端"""
...

def coder_agent(task: str) -> str:
"""写代码的 Agent：根据任务写出脚本"""
return llm_call(
system="你是一名运维工程师，只输出可运行的 Shell 脚本，不要解释。",
user=task,
)

def reviewer_agent(code: str) -> str:
"""审查的 Agent：检查脚本的安全问题"""
return llm_call(
system="你是代码审查员，只找安全问题：rm -rf 未判空、变量未加引号、"
"硬编码密码。有问题列出来，没问题只回 OK。",
user=code,
)

def run(task: str, max_rounds: int = 3) -> str:
"""协作流程：写 → 查 → 改，最多 3 轮"""
code = coder_agent(task)
for round_no in range(1, max_rounds + 1):
review = reviewer_agent(code)
print(f"第 {round_no} 轮审查：{review}")
if review.strip() == "OK":
return code
code = coder_agent(f"按以下意见修改脚本：\n{review}\n原脚本：\n{code}")
raise RuntimeError(f"审查 {max_rounds} 轮仍未通过，转人工处理")

code = run("写一个清理 30 天前日志的脚本")
print("最终脚本：\n", code)

写和查分开，是多智能体最经典的用法。审查 Agent 不用懂业务，只要懂安全规范，就能拦住大部分低级错误。关键操作再加人工确认，就是一道完整的防线。注意 max_rounds：不设上限，两个 Agent 可能反复修改不收敛，必须设最大轮数和超时。

五、实战：巡检多智能体（分工并行）

# 巡检场景：三个 Agent 并行检查，最后汇总
import concurrent.futures

def check_license() -> str:
return llm_call(
system="你是 license 巡检员，只汇报占用率超过 80% 的 feature。",
user="检查 VCS、Verdi 的 license 占用",
)

def check_disk() -> str:
return llm_call(
system="你是磁盘巡检员，只汇报使用率超过 85% 的挂载点。",
user="检查 /eda/proj 和 /home 的磁盘使用",
)

def check_queue() -> str:
return llm_call(
system="你是任务队列巡检员，只汇报排队超过 1 小时的任务。",
user="检查 LSF 队列的排队情况",
)

def summarize(results: dict) -> str:
"""汇总 Agent：把三个检查结果拼成报告"""
report = "\n".join(f"\n{v}" for k, v in results.items())
return llm_call(
system="你是巡检报告员，把下面三项检查结果写成一份报告，"
"异常项标出来并写处理建议。",
user=report,
)

# 并行执行三个检查
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
futures = {
"license": pool.submit(check_license),
"disk": pool.submit(check_disk),
"queue": pool.submit(check_queue),
}
results = {k: f.result() for k, f in futures.items()}

print(summarize(results))

六、框架选型

选型建议：流程固定、要求可控的场景，自研函数加并发最省事；流程复杂、要断点续跑，上 LangGraph；快速验证想法，用 CrewAI。别为了用框架而用框架，框架是手段，不是目的。

七、状态与容错

• Checkpoint：每步执行完存一次状态，挂了从断点续跑，别从头再来。LangGraph 原生支持，自研的话每步写文件或者数据库。
• 重试：工具调用失败重试 2 到 3 次，间隔递增。重试要设上限，无限重试等于死循环。
• 超时：每步设超时，超时就跳过或者转人工，别让一个卡住的步骤拖垮整个流程。
• 死循环防护：设最大轮数，超过就停下来转人工。两个 Agent 互相等对方先动手，或者反复修改不收敛，最大轮数是最后一道防线。
• 降级：某个 Agent 挂了，流程能不能继续？事先定好：非关键步骤跳过，关键步骤转人工。

八、成本与性能

• Token 预算：Agent 越多 token 烧得越快。每个任务设 token 预算，超了就停下来，提示用户确认。别让 Agent 无限制地烧钱。
• 并行度：能并行的步骤并行跑，别串行等。巡检场景三个检查并行，总耗时等于最慢的那个，不是三个加起来。
• 缓存：重复的查询结果缓存起来，别每次都调模型。系统提示词用前缀缓存，省 token 也省时间。
• 模型分级：简单步骤用小模型，复杂推理用大模型。审查 Agent 用小模型就够，写代码 Agent 用大模型。

九、评估：多智能体好不好，要能量化

• 任务成功率：端到端跑下来，多少任务完成了，多少转人工了。成功率低于 90% 就要拆任务或者改流程。
• 协作效率：平均几轮完成，多少轮是无效返工。返工率高说明分工或者审查标准有问题。
• 成本：每个任务平均多少 token、多少钱。成本失控先看是不是 Agent 太多或者轮数太多。
• 对比基线：多智能体和单 Agent 跑同一批任务，对比成功率和成本。多智能体成功率没高多少、成本翻倍，就是过度设计。

十、安全：权限边界与人工确认

• 权限边界：每个 Agent 只给必要的工具和权限。写代码的 Agent 不能删文件，审查的 Agent 不能执行命令。权限按角色分，别全开。
• 人工确认：高危操作（删数据、重启服务、改配置）先返回计划，等用户确认再执行。确认环节不能省，这是最后一道防线。
• 审计：每次工具调用记日志：哪个 Agent 调的、调了什么、参数是什么、结果是什么。出了问题能追溯到具体哪一步。
• 沙箱：Agent 跑命令先在沙箱里试，确认安全再在生产跑。别让 Agent 直接在生产环境试错。

多智能体的本质是分工：把大任务拆成小任务，让每个 Agent 只干自己最擅长的一件。先写查分开这一个模式跑通，再谈更复杂的协作。分工越清晰，协作越顺畅；分工越模糊，扯皮越多。

Mr丶点心

打赏 海报 链接