🎯 前言

在前14篇的连载中，我们从单节点工作流、DAG编排，一路打磨到了断点持久化、全链路监控、高可用容错，彻底搞定了单智能体企业级落地的所有核心能力。

但真实复杂业务中，单Agent存在能力天花板：

• 单一模型既要规划、又要执行、还要审核，职责混杂、逻辑混乱

• 复杂综合性任务（方案撰写、项目分析、全维度复盘）输出粗糙、逻辑漏洞多

• 无法实现分工协作、互相校验、层层打磨的专业化工作模式

• 任务复杂度提升后，单Agent Prompt极其臃肿，难以维护迭代

想要搭建真正的高级自主AI系统，必须升级架构：从「单Agent全包」进化为「多Agent分工协同」。

本篇作为LangGraph高阶核心章节，零基础手把手搭建拆解Agent、执行Agent、审核Agent三套智能体协同体系，实现任务拆分、分工执行、互评校验、迭代优化的完整闭环，对标大厂多智能体工程落地方案。

一、多智能体核心原理（小白秒懂）

1.1 什么是多Agent协同？

模仿人类团队协作模式：不同Agent承担不同职责，各司其职、互相配合、互相监督，共同完成单一Agent无法搞定的复杂任务。

类比人类团队：项目经理（拆解）→ 执行员工（落地）→ 质检审核（校验），完美对应多智能体架构。

1.2 三大核心角色定义

• 拆解规划Agent：负责接收复杂需求、拆解细分任务、制定执行大纲，解决「不知道怎么做」的问题

• 内容执行Agent：根据拆解后的细分任务，专注落地输出内容，解决「内容生产」问题

• 审核优化Agent：校验执行结果、排查漏洞、提出优化建议、判定是否达标，解决「质量不可控」问题

1.3 多Agent相比单Agent的绝对优势

• 能力专业化：每个Agent职责单一、专精岗位，输出质量远超全能单Agent

• 逻辑更严谨：执行+审核双向制衡，杜绝逻辑漏洞、内容敷衍

• 架构更清晰：职责解耦，新增能力只需新增Agent，无需改动原有代码

• 可迭代性强：单独优化某个Agent的Prompt即可全局提升效果

• 适配超复杂任务：长文案、项目方案、数据分析、复盘报告等重度任务完美适配

二、本次实战整体架构

我们搭建一套工业级三智能体协同工作流，完整流程闭环：

1. 规划Agent接收用户复杂需求，拆解结构化执行任务大纲

2. 执行Agent根据大纲批量生产完整内容

3. 审核Agent全方位质检，判定达标/不达标并给出优化点

4. 不达标则自动返回执行Agent迭代优化，达标直接输出最终结果

5. 全程叠加断点续传、容错重试、LangSmith监控，纯生产级架构

三、完整可运行生产级代码

本篇代码为多智能体通用模板，可直接拓展为四Agent、五Agent复杂团队，兼容前序所有工程能力，2026最新版本完美适配。

from dotenv import load_dotenv
import os
from typing import TypedDict, Literal
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

# 加载环境变量
load_dotenv()

# ===================== 继承上篇工程能力：监控+持久化 =====================
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = os.getenv("LANGSMITH_API_KEY")
os.environ["LANGCHAIN_PROJECT"] = "LangGraph-多智能体协同实战"
# =========================================================================

# --------------------------
# 全局状态定义（多Agent共享状态）
# --------------------------
class MultiAgentState(TypedDict):
    user_task: str        # 用户原始复杂需求
    task_outline: str     # 规划Agent拆解的任务大纲
    content_result: str   # 执行Agent生成的内容
    audit_result: str     # 审核Agent质检结果
    iter_count: int       # 迭代优化次数

# --------------------------
# 模型初始化（各Agent可独立配置不同模型）
# --------------------------
llm_planner = ChatOpenAI(
    api_key=os.getenv("API_KEY"),
    base_url=os.getenv("BASE_URL"),
    model="gpt-3.5-turbo",
    temperature=0.2  # 规划类任务低随机，保证严谨
)

llm_executor = ChatOpenAI(
    api_key=os.getenv("API_KEY"),
    base_url=os.getenv("BASE_URL"),
    model="gpt-3.5-turbo",
    temperature=0.4  # 内容生成适度灵活
)

llm_auditor = ChatOpenAI(
    api_key=os.getenv("API_KEY"),
    base_url=os.getenv("BASE_URL"),
    model="gpt-3.5-turbo",
    temperature=0.1  # 审核质检极低随机，标准统一
)

# 断点持久化
memory = MemorySaver()

# --------------------------
# 智能体1：规划拆解Agent
# --------------------------
def planner_agent(state: MultiAgentState) -> MultiAgentState:
    """负责拆解复杂用户任务，输出结构化执行大纲"""
    prompt = f"""
    你是专业的任务规划拆解师，请对用户的复杂需求进行结构化拆解：
    需求：{state['user_task']}
    
    输出要求：
    1、拆解为清晰的执行步骤与核心写作模块
    2、明确每个模块的写作重点、风格、核心要点
    3、输出结构化大纲，逻辑清晰、可直接交给执行人员落地
    """
    res = llm_planner.invoke(prompt)
    state["task_outline"] = res.content.strip()
    state["iter_count"] = 0
    print("📋 【规划Agent】任务拆解完成")
    return state

# --------------------------
# 智能体2：内容执行Agent
# --------------------------
def executor_agent(state: MultiAgentState) -> MultiAgentState:
    """根据规划大纲/审核建议，生成/优化完整内容"""
    # 首次生成：根据大纲创作
    if state["iter_count"] == 0:
        prompt = f"""
        你是专业内容创作者，严格按照规划大纲完成优质内容输出：
        需求：{state['user_task']}
        执行大纲：{state['task_outline']}
        
        要求：内容完整、逻辑流畅、贴合主题、通俗易懂
        """
    # 迭代优化：根据审核问题修改
    else:
        prompt = f"""
        根据审核反馈优化内容，针对性修复问题，保留原有优质结构：
        原文内容：{state['content_result']}
        审核问题：{state['audit_result']}
        """
    res = llm_executor.invoke(prompt)
    state["content_result"] = res.content.strip()
    state["iter_count"] += 1
    print("✍️ 【执行Agent】内容生成/优化完成，当前迭代次数：", state["iter_count"])
    return state

# --------------------------
# 智能体3：审核质检Agent
# --------------------------
def auditor_agent(state: MultiAgentState) -> MultiAgentState:
    """全方位质检内容，判定达标或给出优化问题"""
    prompt = f"""
    你是严苛的内容审核专家，请全方位质检以下内容：
    用户需求：{state['user_task']}
    执行大纲：{state['task_outline']}
    生成内容：{state['content_result']}
    
    审核标准：
    1、是否贴合用户核心需求
    2、是否符合大纲结构，内容完整无缺失
    3、逻辑是否通顺、无病句、无逻辑漏洞
    4、内容质量是否优质、通俗易懂
    
    输出规则：
    完全达标输出：【达标】
    存在问题输出：【不达标】+ 逐条列出具体优化问题与修改建议
    """
    res = llm_auditor.invoke(prompt)
    state["audit_result"] = res.content.strip()
    print("✅ 【审核Agent】质检完成")
    return state

# --------------------------
# 协同路由逻辑：自动判断迭代/结束
# --------------------------
def agent_route(state: MultiAgentState) -> Literal["re_optimize", "finish"]:
    # 最大迭代3次，防止无限循环
    if state["iter_count"] >= 3:
        return "finish"
    # 审核达标直接结束
    if "达标" in state["audit_result"]:
        return "finish"
    # 不达标返回执行Agent优化
    return "re_optimize"

# --------------------------
# 搭建多智能体协同工作流
# --------------------------
graph = StateGraph(MultiAgentState)

# 注册三大智能体节点
graph.add_node("planner_agent", planner_agent)
graph.add_node("executor_agent", executor_agent)
graph.add_node("auditor_agent", auditor_agent)

# 流程拓扑：规划 → 执行 → 审核 → 条件循环
graph.add_edge(START, "planner_agent")
graph.add_edge("planner_agent", "executor_agent")
graph.add_edge("executor_agent", "auditor_agent")

# 条件分支：继续优化 / 任务结束
graph.add_conditional_edges(
    "auditor_agent",
    agent_route,
    {"re_optimize": "executor_agent", "finish": END}
)

# 编译工作流，绑定断点持久化
multi_agent_workflow = graph.compile(checkpointer=memory)

# --------------------------
# 运行测试
# --------------------------
if __name__ == "__main__":
    config = {"configurable": {"thread_id": "2026_multi_agent_001"}}
    # 复杂综合性需求（单Agent很难写好）
    complex_task = "写一篇零基础入门LangGraph的学习指南，包含核心价值、学习路线、实战场景、避坑要点，适合新手入门"

    result = multi_agent_workflow.invoke({"user_task": complex_task}, config=config)

    print("\n🎉 多智能体协同任务全部完成！")
    print("=" * 50)
    print("最终生成内容：")
    print(result["content_result"])
    print("\n最终审核结果：", result["audit_result"])
    print("总迭代优化次数：", result["iter_count"])

四、核心架构逐点深度拆解

4.1 职责完全解耦（核心精髓）

三个Agent各司其职、互不干扰：

• 规划Agent只负责「思考拆解」，不做内容落地，保证框架严谨

• 执行Agent只负责「落地输出」，专注内容撰写，保证内容饱满

• 审核Agent只负责「监督质检」，客观评判优劣，保证输出质量

彻底解决单Agent「又当裁判又当选手」的天然缺陷。

4.2 差异化模型配置

实战中根据职责差异化设置温度，是工业级最佳实践：

• 规划、审核类任务：温度0.1-0.2，追求稳定、严谨、标准统一

• 内容生成类任务：温度0.4左右，兼顾质量与灵活性

4.3 闭环迭代机制

形成拆解→生成→审核→优化→再审核的闭环打磨体系，相比单次生成，内容质量提升数倍，完美适配高精度输出场景。

4.4 全工程能力兼容

本架构天然兼容：

• Checkpoint断点续传：多Agent长流程中断可无缝恢复

• LangSmith监控：可清晰看到每个Agent的执行、入参、出参、耗时

• 容错重试机制：可快速给每个Agent节点叠加第十四篇容错装饰器

五、多智能体进阶拓展方案

5.1 角色拓展（四Agent/五Agent团队）

可根据业务自由新增专业Agent：

• 调研检索Agent：负责联网查询、资料搜集

• 格式排版Agent：负责Markdown规整、格式美化

• 润色优化Agent：专门负责语句通顺、文案润色

• 合规筛查Agent：负责内容合规、敏感词检测

5.2 分层协同架构

复杂项目可结合第十篇Subgraph：主Agent调度多个子Agent团队，实现超大型分层智能体系统。

5.3 自主决策协同

可升级为动态路由：Agent自主判断是否需要调用其他Agent、是否需要检索、是否需要迭代优化，实现完全自主智能协作。

六、新手避坑指南

坑1：所有Agent使用相同Prompt与温度

问题：角色同质化，失去分工意义，效果和单Agent无区别。

解决：根据职责定制专属Prompt、差异化温度参数。

坑2：无迭代次数限制

问题：审核一直不达标，导致无限循环、Token爆炸。

解决：必须设置最大迭代次数，强制兜底结束流程。

坑3：多Agent状态混乱

问题：状态字段定义混乱，Agent互相抢占数据。

解决：统一全局状态，明确每个字段归属与用途。

坑4：职责重叠冗余

问题：多个Agent做重复工作，资源浪费、流程拖沓。

解决：严格划分每个Agent的边界职责，互不越权。

七、零基础自测巩固

1、多智能体协同相比单智能体的核心优势是什么？

2、规划、执行、审核三大Agent各自的核心职责与边界是什么？

3、为什么不同职能的Agent需要设置不同的temperature温度？

✅ 本篇核心总结

1、多智能体协同是LangGraph高阶核心能力，是搭建复杂自主AI系统的必备架构；

2、通过职责拆分，实现规划、执行、审核专业化分工，解决单Agent能力瓶颈；

3、搭配迭代闭环机制，实现内容自主打磨、自我纠错，输出质量远超单次生成；

4、完美兼容前序所有工程化能力，可快速落地为企业级复杂AI应用。

📌 下一篇预告

第十六篇：【2026零基础AI教程16】LangGraph工具调用高阶封装、参数校验、自动适配，彻底解决工具调用失败、参数错乱、格式异常问题