2026 多智能体协作实战：用 CrewAI 实现自主纠错与跨 Agent 通信的智能工作流

摘要

随着大语言模型推理与工具调用能力的成熟，企业自动化需求已从单一任务迈向“搜索—分析—写作—审核”多步骤协同。单 Agent 模式在复杂流程中显露出难以维护、缺乏纠错等短板。2026 年，以 CrewAI 为代表的多智能体框架通过角色化 Agent、结构化任务与灵活的流程编排，使复杂工作流可定义、可纠错、可跨 Agent 通信。本文以新闻聚合与审核系统为例，完整演示如何基于 CrewAI 构建支持自主纠错与反馈闭环的内容生产线。

问题背景：当单一 Agent 面对多步骤流程

过去两年，基于大模型的单体 Agent 在对话、文本生成等场景表现优异，但真实业务需求往往不止于此。以内容运营为例，一条“监控热点 → 聚合信息 → 生成摘要 → 审核事实 → 最终发布”的流水线，涉及搜索、总结、校验等多个子任务。传统做法是在一个 Agent 里串行调用工具，常出现以下问题：

• 单点故障难恢复：一个步骤失败导致整条链崩溃，缺乏重试或备用路径。
• 质量无闭环：生成的内容没有独立审核环节，错误信息直接交付。
• 逻辑耦合沉重：将搜索、写作、审核混杂在同一 Prompt 中，调试困难，扩展性差。
• 并行受限：无法让多个 Agent 同时工作并交换结果。

这些痛点表明，我们需要一种能够天然支持角色分工、任务编排、自主纠错与通信的架构，多智能体自主协作框架由此走向生产前端。

技术方案：角色化 Agent + 结构化任务 + 灵活流程

当前主流的 Agentic Workflow 框架（如 CrewAI、AutoGen）都遵循相似的核心设计哲学：

• Agent：拥有独立角色（研究员/写手/审核员）、目标、背景故事，并可以绑定工具（搜索引擎、API 等）。
• Task：是工作的最小单元，包含明确描述、期望输出、分配的执行 Agent，还可指定依赖关系。
• Crew：负责组装 Agent 和 Task，并决定执行流程——顺序（sequential）、层级（hierarchical，有管理者 Agent）或自定义图结构。

此外，框架普遍内置了纠错与迭代机制：当某个 Task 的输出不符合预期时，能触发重试、引入人工反馈，或将结果传给上一级 Agent 重新生成。跨 Agent 通信则通过共享上下文或直接传递 Task 输出来实现，让协作更顺畅。

本文选用 CrewAI 进行演示，它 API 简洁，能快速搭建出具备自主纠错与多轮反馈的工作流。

核心实现：构建一个新闻摘要与事实审核系统

场景设计

我们希望自动完成以下流程：

1. 研究员 Agent 使用搜索工具获取最新科技新闻链接与概要。
2. 写手 Agent 基于研究员提供的资料生成一篇简讯。
3. 审核员 Agent 检查简讯中的关键事实是否与原始资料一致，给出修改建议或通过标志。
4. 如果未通过，写手 Agent 根据反馈重新修订，直到审核通过或到达最大迭代次数。

这是一个典型的双反馈闭环，体现跨 Agent 通信与自主纠错。注意：CrewAI 的原生顺序流程会将所有任务依次执行，无法直接根据审核结果跳过修订任务。为实现“未通过才修订”的条件分支，我们将在 Python 层面解析审核输出，通过单独启动一次仅包含修订任务的 Crew（或手动调用 Agent）来触发修正，并将迭代次数限制在外部循环中。

代码实现

首先安装依赖（示例使用 CrewAI v0.14+，如使用更新版本请根据 API 调整导入路径）：

pip install crewai crewai-tools

定义工具、Agent 与 Task（crew_app.py）：

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool

# 初始化搜索工具（需要 SERPER_API_KEY 环境变量）
search_tool = SerperDevTool()

# --- 定义 Agents ---
researcher = Agent(
    role='新闻研究员',
    goal='查找今日最重要的科技新闻，提供原始链接与关键摘要',
    backstory='你是一名资深科技媒体人，擅长从海量信息中提取最有价值的内容。',
    tools=[search_tool],
    verbose=True,
    allow_delegation=False
)

writer = Agent(
    role='内容写手',
    goal='基于研究员提供的资料，撰写一篇 200 字以内的新闻简讯',
    backstory='你是一名专业的科技记者，文笔流畅，能够将复杂信息转化为大众易读的文章。',
    verbose=True,
    allow_delegation=False
)

reviewer = Agent(
    role='事实审核员',
    goal='核对简讯中的关键数据、名称、结论是否与原始资料一致，并指出错误或给出通过意见',
    backstory='你是一名严谨的编辑，对事实性错误零容忍。',
    verbose=True,
    allow_delegation=False
)

# --- 定义 Tasks ---
search_task = Task(
    description='搜索今日最热门的 3 条科技新闻，每条提供标题、来源链接和 50 字摘要。',
    expected_output='一份包含 3 条新闻的列表，每条有 title、url、summary 字段。',
    agent=researcher
)

write_task = Task(
    description='根据研究员提供的新闻列表，整合成一篇连贯的简讯，字数不超过 200。',
    expected_output='一篇中文简讯，标题自拟，内容包含 3 条新闻要点。',
    agent=writer,
    context=[search_task]   # 接收研究员任务的输出作为上下文
)

review_task = Task(
    description='''严格核对简讯中的事实数据（如公司名称、数字、人物）是否与研究员提供的原始摘要一致。
        若发现错误，列出具体修改建议（以“更正：”开头）；若无错误，请只输出 PASS。''',
    expected_output='审核意见：错误列表或"PASS"。',
    agent=reviewer,
    context=[search_task, write_task]   # 同时参考原始资料和简讯
)

# --- 组装主 Crew（只执行搜索、写作、审核）---
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[search_task, write_task, review_task],
    process=Process.sequential,
    verbose=True
)

# 启动主流程
result = crew.kickoff()
print("========== 审核结果 ==========")
print(result)

# --- 条件纠错：仅在审核不通过时触发修订 ---
revise_task = Task(
    description='''根据审核员的修改建议修正简讯，确保所有事实准确。
        若无错误则无需更改，直接返回原简讯。最终输出修改后的完整简讯。''',
    expected_output='修改后的简讯，仍控制在 200 字以内。',
    agent=writer,
    context=[write_task, review_task]  # 参考原稿和审核意见
)

MAX_REVISIONS = 2   # 最多修正两次
revision_count = 0

while "PASS" not in str(result).strip() and revision_count < MAX_REVISIONS:
    print(f"--- 检测到审核不通过，开始第 {revision_count+1} 次修订 ---")
    # 单独运行修订任务：创建一个只包含修订任务的 Crew
    revise_crew = Crew(
        agents=[writer],
        tasks=[revise_task],
        process=Process.sequential,
        verbose=True
    )
    revised_output = revise_crew.kickoff()
    print(f"--- 修订后输出 ---\n{revised_output}")

    # 再次审核修订后的内容
    review_again_task = Task(
        description='''再次核对以下简讯中的事实数据是否与研究员提供的原始摘要一致。
            若发现错误，列出具体修改建议（以“更正：”开头）；若无错误，请输出 PASS。''',
        expected_output='审核意见：错误列表或"PASS"。',
        agent=reviewer,
        context=[search_task, write_task]   # 注意此处 write_task 的上下文仍是原稿，但审核对象通过新任务描述传入修订后内容？实际需要将修订结果传入。
    )

    # 简便做法：将修订后的文本直接作为 review_again_task 的描述追加
    review_again_task.description += f"\n\n待审核简讯内容：\n{revised_output}"

    review_crew = Crew(
        agents=[reviewer],
        tasks=[review_again_task],
        process=Process.sequential,
        verbose=True
    )
    result = review_crew.kickoff()
    print(f"--- 复核结果 ---\n{result}")

    revision_count += 1

if revision_count == MAX_REVISIONS and "PASS" not in str(result).strip():
    print("警告：已达到最大修正次数，但审核仍未通过。输出最后一次修订结果。")

print("========== 最终输出 ==========")
print(result)

条件分支说明：上述代码将顺序流程拆分为两个阶段：第一阶段执行搜索、写作、审核，获取审核意见；第二阶段在 Python 层判断审核输出中是否包含 PASS，若不包含则循环启动修订任务并再次审核，直到通过或达到最大迭代次数。这样便实现了真正的“未通过才修订”的条件触发闭环，而非无条件顺序执行。若希望减少 Crew 对象创建开销，也可以直接通过 agent.execute_task() 方法在循环中调用任务。CrewAI 的层级流程（Hierarchical）同样可以委托 Manager Agent 根据审核结果决定是否指派写手修订，但其行为更依赖模型判断，可控性略低，本文采用外部循环模式更利于演示明确的分支逻辑。

关键设计点解析

• 跨 Agent 通信：通过 context 参数，write_task 可以获得 search_task 的实际输出结果；review_task 可以同时引用研究员和写手的输出；修订任务则直接阅读原稿和审核意见。CrewAI 上下文传递机制会将前序 Task 的实际执行结果（output）注入到当前 Agent 的系统消息中，而非 expected_output 文本，开发者可放心依赖任务间数据的可靠流转。
• 自主纠错闭环：修订任务仅在外部循环判定审核未通过时触发，若首次审核即通过则完全跳过修订，避免了资源浪费。通过 MAX_REVISIONS 控制最大修正轮次，防止因模型幻觉导致的无限重试。注意：CrewAI 原生 Task 或 Crew 对象没有直接提供 max_iter 参数用于自定义重试循环，迭代次数和退出条件需在外部 Python 逻辑或自定义 Process 中实现，本文的做法即清晰展示了这一控制模式。
• 工具集成：SerperDevTool 是 Google 搜索的轻量包装，让 Agent 能实时获取外部信息，保证了输出内容的新鲜度。
• 内存共享：示例中未显式启用 memory。若使用的是较早期版本（如 v0.14.x）且确实需要跨 Crew 运行保留对话历史，可以通过设置 memory=True 或配置文件启用；但在 v0.50+ 版本中，memory 机制已重构为通过 embedder 配置实现持久化，建议升级后查阅对应文档，或简单通过任务上下文传递关键信息，生产环境可在此基础上接入向量数据库实现长期记忆。

运行效果与生产实践建议

实际运行日志片段

执行脚本后，控制台会依次输出各 Agent 的思考与动作日志：

[研究员] 开始搜索科技新闻...
[研究员] Thought: 我需要查找今天的重要科技新闻，使用搜索工具...
[研究员] Observation: 返回3条结果：OpenAI发布多模态模型，估值90亿美元；特斯拉人形机器人量产推迟...
[写手] 收到研究员资料，开始撰写简讯...
[写手] 输出：今日简讯：OpenAI推出新一代多模态模型，公司估值已达90亿美元；特斯拉人形机器人量产计划推迟至2027年...
[审核员] 开始核对...
[审核员] 发现错误：研究员资料中显示 OpenAI 本轮估值约为 85 亿美元，简讯中为 90 亿美元。
[审核员] 输出：更正：将“90亿美元”改为“约85亿美元”。

接着检测到未通过，触发修订，写手重新生成并修正数字，复核输出“PASS”。整个流程无需人工干预，体现出自驱的协作与纠错能力。

生产落地的关键考量

在实际业务中直接部署此类工作流还需考虑以下几点：

1. API 限流与并发控制
   搜索工具调用受 Serper 的 QPS 限制，写作与审核使用的 LLM API 也有 RPM/TPM 上限。建议在 Agent 的工具调用层加入重试与指数退避，避免因单一环节失败导致整条流水线停滞。对于高并发场景，可引入异步任务队列（如 Celery）并发执行搜索，再聚合结果。

2. Token 成本控制
   一次完整流程可能消耗数万 Token，尤其是当审核员需要阅读大量原始资料时。成本优化措施包括：为每个 Agent 单独设置较低的最大 Token 数、对搜索返回内容进行摘要压缩、仅在修订时传入差异片段（而非全文）。实测中，一次包含两次修订的完整过程 GPT-4 的费用约为 $0.3-$0.5，日报级别尚可接受。

3. LLM 幻觉兜底
   审核员本身也可能出错（漏判或误判）。建议添加“最终人工抽检开关”：当审核员连续 N 轮仍判定不通过，或置信度较低时，自动将争议内容推送至 IM 通知群或审核后台，由人工裁决。也可引入多模型投票机制提高审核准确性。

4. 状态持久化与可观测性
   生产环境应为每个工作流实例生成唯一 ID，记录每轮任务的输入、输出、审核意见及耗时，便于回溯与调优。CrewAI 的回调机制（step_callback）可用于将日志写入数据库或监控系统。

5. 安全与内容合规
   当自动发布内容到公开平台时，务必增加敏感词过滤与合规检查步骤。可在审核员之后增加一个轻量级的安全 Agent，或直接调用内容安全 API 进行终审。

总结与展望

多智能体自主协作框架正从概念走向生产，它们的核心价值在于将复杂的业务逻辑拆解为可管理、可观测、可纠错的角色网络。本文通过 CrewAI 展示了搜索、撰写、审核、修正四步走的内容生成流水线，并详细说明了如何用条件循环实现真正的按需纠错，开发体验直观，纠错能力显著。

展望 2026 年下半年，这类框架将进一步与低代码平台、企业数据中间件深度集成，支持动态任务规划、人机协同审核、跨组织 Agent 通信等高级功能。对后端/全栈开发者而言，现在正是学习并应用多智能体架构的最佳时机——用更少代码构建更可靠、更智能的自动化服务。

（全文完）