一、什么是多智能体系统？

多智能体系统（Multi-Agent System, MAS）是由多个独立的、具有特定角色和能力的智能体组成，通过相互通信、协作共同完成一个复杂任务的系统。

在 LangGraph 中，多智能体的本质就是：每个智能体是一个独立的子图，主图作为调度器，管理智能体之间的交互和任务流转。

多智能体 vs 单智能体 vs 子图

表格

类型	核心特点	适用场景
单智能体	一个大脑做所有事，全能但不专业	简单工具调用、单轮问答
子图	按功能拆分模块，代码复用	功能模块化、代码解耦
多智能体	按角色拆分，每个智能体有独立的人格、目标和能力，相互协作	复杂写作、项目开发、团队协作、多角色决策

多智能体的核心优势

1. 角色专业化：每个智能体只做自己擅长的事，输出质量远高于单智能体
2. 可扩展性：新增角色只需新增一个子图，不影响现有系统
3. 容错性：一个智能体出错，其他智能体可以纠正或兜底
4. 复杂任务分解：天然适配需要多角色分工的复杂任务（写报告、做方案、代码开发）

二、双智能体协作系统架构

我们要实现的研究员 + 作家双智能体系统，是最经典、最实用的多智能体架构：

1. 角色分工

表格

智能体	核心职责	输出
研究员	围绕主题收集资料、整理要点、提供事实依据	结构化的研究报告、资料要点
作家	基于研究员的资料，进行内容创作、润色、排版	完整的文章、报告、文案

2. 标准协作流程

用户输入写作任务 → 研究员收集整理资料 → 作家基于资料写初稿
    ↓（需要补充资料）
研究员补充资料 → 作家修改完善 → 最终输出完成

3. 核心 State 设计

多智能体的 State 必须包含所有智能体共享的全局信息，是智能体之间通信的唯一载体：

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
from langgraph.graph import add_messages

class MultiAgentState(TypedDict):
    # 全局任务信息
    task: str  # 用户的原始写作任务
    topic: str  # 写作主题
    
    # 智能体输出
    research_notes: str  # 研究员的研究笔记
    draft: str  # 作家的初稿
    final_article: str  # 最终完成的文章
    
    # 交互控制
    current_agent: Literal["researcher", "writer", "user"]  # 当前轮到谁工作
    turn_count: int  # 交互轮次，防死循环
    max_turns: int  # 最大交互轮次
    
    # 对话历史（智能体之间的通信记录）
    messages: Annotated[Sequence[BaseMessage], add_messages]
    
    # 错误处理
    error: str | None

---

三、完整可运行代码实现

基于你之前的代码风格，实现了角色明确、交互自然、带错误处理、带持久化的双智能体协作系统，完全兼容智谱 API 和国内网络环境。

第一步：环境准备

bash

运行

pip install -U langgraph  langchain_community pydantic_settings pyjwt

第二步：完整代码（使用智普模型 = “glm-4-flash”）

from dotenv import load_dotenv
from typing import TypedDict, Literal, Annotated, Sequence

from langchain_community.chat_models import ChatZhipuAI
from langchain_core.messages import BaseMessage, AIMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate
from langgraph.checkpoint.memory import MemorySaver
from langgraph.constants import START, END
from langgraph.graph import add_messages, StateGraph
from pydantic_settings import BaseSettings, SettingsConfigDict

# 加载环境变量
load_dotenv()

class Settings(BaseSettings):
    ZHIPU_API_KEY: str
    ZHIPU_BASE_URL: str
    LLM_MODEL: str
    LLM_BACKUP_MODEL: str
    LLM_TIMEOUT: int=300
    MAX_TURNS: int=5    # 最大交互轮次 防死循环

    model_config = SettingsConfigDict(
        env_file='.env',
        env_file_encoding='utf-8',
        extra="ignore"
    )

settings = Settings()

# 全局状态定义
class MultiAgentState(TypedDict):
    # 全局任务信息
    task: str   # 用户的原始写作任务
    topic: str  # 写作主题
    # 智能体输出
    research_notes: str  # 研究员的研究笔记
    draft: str # 作家的初稿
    final_article: str # 最终完成的文章

    # 交互控制
    current_agent: Literal["researcher", "writer"]    # 当前轮到谁工作
    turn_count: int # 交互轮次，防死循环
    max_turns: int # 最大交互轮次

    # 对话历史 （智能体之间的通信记录）
    messages: Annotated[Sequence[BaseMessage], add_messages]

    # 错误处理
    error: str | None


# LLM 初始化
# 通用llm 两个智能体共享
llm = ChatZhipuAI(
    api_key=settings.ZHIPU_API_KEY,
    base_url=settings.ZHIPU_BASE_URL,
    model=settings.LLM_BACKUP_MODEL,
    temperature=0.7,
    timeout=settings.LLM_TIMEOUT,
    max_retries=3,  # 内置重试次数
)

# 智能体 研究员
def create_researcher_agent():
    """创建研究员智能体：负责收集资料、整理要点"""
    research_prompt = ChatPromptTemplate.from_messages([
        ("system", """
        你是专业的资深研究员，擅长围绕特定主题收集、整理、提炼准确、全面的资料。
        你的职责：
        1. 严格围绕用户的写作主题，收集相关的核心概念、发展历程、关键技术、应用场景、优缺点等信息
        2. 输出结构化的研究笔记，分点清晰，逻辑严谨
        3. 只提供事实性的资料，不要进行创作或润色
        4. 如果资料不足，明确指出需要补充的内容
        5. 研究笔记要足够详细，能够支撑作家完成一篇高质量的文章
        当前写作任务：{task}
        """),
        ("human", "{input}")
    ])

    def researcher_node(state: MultiAgentState):
        print("\n-- 🔬 研究员开始工作 --")
        try:
            # 构建提示词
            prompt = research_prompt.format(
                task=state["task"],
                topic=state["topic"],
                input="请基于当前主题，提供详情的研究笔记"
            )

            # 请用LLM生成研究笔记
            response = llm.invoke(prompt)
            research_notes = response.content

            print("✅️ 研究完成研究笔记")
            print(f"研究笔记长度：{len(research_notes)}字符")

            return {
                "research_notes": research_notes,
                "current_agent": "writer",
                "turn_count": state["turn_count"] + 1,
                "messages": [AIMessage(content=f"研究员已完成研究笔记：\n{research_notes}")]
            }
        except Exception as e:
            print(f"❌️ 研究院工作失败：{str(e)}")
            return {
                "error": str(e),
                "turn_count": state["turn_count"] + 1,
            }

    # 研究员是一个简单的单节点子图
    builder = StateGraph(MultiAgentState)
    builder.add_node("researcher", researcher_node)
    builder.set_entry_point("researcher")
    return builder.compile()

# 智能体2 作家
def create_writer_agent():
    """创建作家智能体：负责基于研究笔记进行内容创作"""
    writer_prompt = ChatPromptTemplate.from_messages([
        ("system", """
        你是专业的资深作家，擅长基于研究员提供的资料，创作高质量、结构清晰、语言流畅的文章。
        你的职责：
        1. 严格基于研究员提供的研究笔记进行创作，绝对禁止编造内容
        2. 文章结构要完整，包含引言、正文、结论三个部分
        3. 语言要通顺、专业、有逻辑性，适合目标读者阅读
        4. 如果研究笔记中有不足或缺失的内容，明确指出需要研究员补充的信息
        5. 输出完整的文章初稿，格式规范，段落清晰
        当前写作任务：{task}
        写作主题：{topic}
        研究员提供的研究笔记：
        {research_notes}
        """),
        ("human", "{input}")
    ])
    def writer_node(state: MultiAgentState):
        print(f"\n-- ✍️ 作家开始工作 --")
        try:
            prompt = writer_prompt.format(
                task=state["task"],
                topic=state["topic"],
                research_notes=state["research_notes"],
                input="请基于研究笔记，写一篇完整的文章初稿"
            )
            # 调用llm生成初稿
            response = llm.invoke(prompt)
            draft = response.content

            print("✅️ 作家完成文章初稿")
            print(f"初稿长度：{len(draft)}字符")

            return {
                "draft": draft,
                "current_agent": "researcher",
                "turn_count": state["turn_count"] + 1,
                "messages": [AIMessage(content=f"作家已完成文章初稿：\n{draft}")]
            }
        except Exception as e:
            print(f"❌ 作家工作失败：{str(e)}")
            return {
                "error": str(e),
                "turn_count": state["turn_count"] + 1,
            }
    builder = StateGraph(MultiAgentState)
    builder.add_node("writer", writer_node)
    builder.set_entry_point("writer")
    return builder.compile()

# 智能体调度器
def build_multi_agent_system():
    """"构建双智能体协作系统"""
    # 创建两个智能体子图
    researcher_agent = create_researcher_agent()
    writer_agent = create_writer_agent()

    # 主图构建
    builder = StateGraph(MultiAgentState)
    # 把子图作为节点加入主图
    builder.add_node("researcher", researcher_agent)
    builder.add_node("writer", writer_agent)

    # 最终汇总节点：生成最终文章
    def finalize_node(state: MultiAgentState):
        print(f"\n-- 🎉 任务完成，生成最终文章 --")
        return {
            "final_article": state["draft"],
            "messages": [AIMessage(content=f"最终文章：\n{state['draft']}")]
        }
    builder.add_node("finalize", finalize_node)

    # 条件路由：决定下一步由哪个智能体工作 还是结束
    def router(state: MultiAgentState) -> Literal["researcher", "writer", "finalize", "end"]:
        # 优先处理错误
        if state.get("error"):
            print(f"❌ 系统出错：{state['error']}，任务结束")
            return "end"

        # 防止死循环 超过最大轮次 强制结束
        if state.get("turn_count") >= state["max_turns"]:
            print("⚠️ 超过最大交互轮次，强制结束任务")
            return "finalize"

        # 根据当前智能体决定下一步
        current_agent = state["current_agent"]
        if current_agent == "researcher":
            # 如果还没研究笔记，研究员先工作
            if not state.get("research_notes"):
                return "researcher"
            # 如果已经有初稿   研究员研究后需要补充资料
            elif state.get("draft"):
                # 简单判断 如果初稿长度足够 直接结束 否则让研究员补充资料
                if len(state["draft"]) > 500:
                    return "finalize"
                else:
                    print(f"🔄 初稿内容不足，研究员补充资料")
                    return "researcher"
            else:
                return "researcher"
        elif current_agent == "writer":
            # 有研究笔记 作家开始写作
            if state.get("research_notes"):
                return "writer"
            else:
                return "researcher"

        else:
            return "finalize"

    # 入口： 开始 -> 路由
    builder.add_edge(START, "researcher")
    # 条件路由
    builder.add_conditional_edges(
        "researcher",
        router,
        {
            "researcher": "researcher",  # ← 添加这个
            "writer": "writer",
            "finalize": "finalize",
            "end": END
        }
    )
    builder.add_conditional_edges(
        "writer",
        router,
        {
            "researcher": "researcher",
            "finalize": "finalize",
            "end": END
        }
    )
    # 汇总完成 -> 结束
    builder.add_edge("finalize", END)

    # 编译主图 带持久化
    checkpointer = MemorySaver()
    return builder.compile(checkpointer=checkpointer)

# 测试运行
if __name__ == "__main__":
    print("===== 🤝 LangGraph 双智能体协作系统 启动 =====")

    # 初始化 多智能体系统
    multi_agent = build_multi_agent_system()
    # 会话配置
    config = {
        "configurable" : {
            "thread_id": "multi_agent_session_001",
            "user_id": "user_001"
        }
    }
    # 测试写作任务
    test_task = "写一篇关于LangGraph多智能体系统的技术文章"
    test_topic = "LangGraph多智能体系统的原理、架构与实现"
    # 初始化状态
    initial_state = {
        "task": test_task,
        "topic": test_topic,
        "research_notes": "",
        "draft": "",
        "final_article": "",
        "current_agent": "researcher",
        "turn_count": 0,
        "max_turns": settings.MAX_TURNS,
        "messages": [HumanMessage(content=test_task)],
        "error": None
    }
    # 运行多智能体系统
    result = multi_agent.invoke(initial_state, config=config)
    # 输出最终结果
    print("\n" + "="*80)
    print("📄 最终完成的文章：")
    print("="*80)
    print(result["final_article"])
    print("="*80)
    print(f"\n✅ 任务完成，共交互 {result['turn_count']} 轮")

---

四、运行效果演示

✅️ 研究完成研究笔记
研究笔记长度：965字符

-- ✍️ 作家开始工作 --
✅️ 作家完成文章初稿
初稿长度：1211字符

-- 🎉 任务完成，生成最终文章 --

================================================================================
📄 最终完成的文章：
================================================================================
标题：LangGraph多智能体系统的原理、架构与实现

引言

随着信息技术的飞速发展，多智能体系统（Multi-Agent System，MAS）逐渐成为研究热点。LangGraph作为一种基于图论的多智能体系统，因其独特的原理和架构，在多个领域展现出巨大的应用潜力。本文将深入探讨LangGraph的原理、架构与实现，以期为相关研究者提供参考。

正文

一、核心概念

1. LangGraph：LangGraph是一种基于图论的多智能体系统，通过智能体之间的交互和合作，实现复杂任务的解决。

2. 多智能体系统（MAS）：由多个自主的智能体组成，通过通信和协调完成共同目标。

二、发展历程

1. 20世纪80年代：多智能体系统理论初步形成，研究主要集中在智能体的自主性、通信和协调等方面。

2. 90年代：LangGraph概念被提出，并逐渐应用于各个领域。

3. 21世纪初：LangGraph技术得到进一步发展，成为多智能体系统研究的热点。

三、关键技术

1. 智能体模型：包括行为模型、通信模型和知识模型，用于描述智能体的行为、交互和知识。

2. 图论：LangGraph的核心技术之一，用于构建智能体之间的交互网络，实现任务分配和优化。

3. 机器学习：通过机器学习算法，使智能体具备自我学习和适应能力，提高系统性能。

四、应用场景

1. 网络安全：LangGraph可以应用于网络安全领域，通过智能体之间的协同防御，提高网络安全性。

2. 物流管理：在物流管理中，LangGraph可以实现智能调度和优化路径规划，提高物流效率。

3. 供应链管理：LangGraph可以应用于供应链管理，实现智能库存控制和需求预测。

4. 智能交通：在智能交通领域，LangGraph可以用于交通信号控制和优化道路流量。

五、优缺点

1. 优点：

   - 提高系统性能：通过智能体之间的协同合作，实现复杂任务的优化解决。

   - 自主性强：智能体具有自主性，能够根据环境变化做出相应决策。

   - 可扩展性强：LangGraph系统可根据实际需求进行扩展和调整。

2. 缺点：

   - 复杂度高：LangGraph系统的构建和维护较为复杂，需要一定的技术积累。

   - 资源消耗大：多智能体系统在运行过程中，对计算资源的需求较高。

   - 通信开销大：智能体之间的通信需要消耗一定的带宽和计算资源。

六、总结

LangGraph多智能体系统作为一种新兴的技术，具有广泛的应用前景。通过深入研究LangGraph技术，有助于推动多智能体系统在各个领域的应用，为我国科技创新贡献力量。

本文对LangGraph的原理、架构与实现进行了探讨，为相关研究者提供了参考。然而，LangGraph技术仍处于发展阶段，未来需要进一步研究和完善，以应对实际应用中的挑战。
================================================================================

✅ 任务完成，共交互 2 轮

---

五、核心要点总结

1. 多智能体 = 多个子图 + 主图调度：每个智能体是独立的子图，主图负责管理交互流程
2. 角色提示词是关键：明确每个智能体的职责、输出格式和行为准则，是多智能体协作成功的核心
3. 状态是通信的唯一载体：所有智能体通过共享的 State 传递信息，不要在智能体之间直接调用
4. 防死循环机制必不可少：设置最大交互轮次，避免智能体无限对话
5. 持久化支持：主图绑定 Checkpointer 后，支持会话恢复和断点续跑

---

六、进阶扩展方向

1. 增加评审员智能体

新增第三个智能体「评审员」，负责审核文章质量，提出修改意见，形成「研究员→作家→评审员→作家修改」的完整闭环。

2. 给智能体增加工具调用能力

• 给研究员增加网络搜索工具，可以实时获取最新资料
• 给作家增加格式转换工具，可以输出 Markdown、PDF 等格式的文章

3. 更复杂的交互模式

• 循环反馈：作家写完初稿后，研究员可以提出修改意见，作家修改后再提交
• 并行执行：多个研究员同时收集不同方面的资料，提升效率
• 人工介入：在关键步骤暂停，等待用户审核或提供反馈后继续执行

4. 多智能体消息总线

用messages字段作为智能体之间的消息总线，每个智能体都可以读取和写入消息，实现更自然的对话式协作。

这个双智能体系统是多智能体开发的基础，你可以基于这个架构，轻松扩展出更多角色和更复杂的协作流程。