本文基于同样的业务场景和案例，采用LangGraph框架进行对比实践和验证，以便更深入地理解。

🧠 多智能体协作实现逻辑

首先回顾一下业务场景，和上篇文章保持一致，还是支撑用户进行高考信息查询，并且要支持多轮对话，后续会基于以下案例来进行验证：

第一轮提问：2016年考生人数有多少？

第二轮追问年份：2018年呢？

第三轮追问其他：录取人数呢？

针对该业务场景，四个智能体之间的协作逻辑和AgentScope案例也完全保持一致：

1）用户输入问题

2）TemplateAgent判断是否有上下文历史，如果没有则转4，有则进一步匹配历史问题模板，如未匹配到则转3，如匹配到则基于模板SQL，从用户当前问题中提取查询参数植入模板SQL，执行后带查询结果转5

3）有上下文历史，但没有匹配的历史问题，则进入意图识别改写RewriteAgent，利用上下文对话背景对用户当前问题做改写，然后进入SQLAgent，RAG检索元数据、生成SQL、查询数据，最后进入AnalysisAgent分析回答

4）无上下文历史，代表用户是首次提问，则直接进入SQLAgent，RAG检索元数据、生成SQL、查询数据，最后进入AnalysisAgent分析回答

5）根据查询结果直接进入AnalysisAgent分析回答

这个协作逻辑基于 LangGraph框架的自定义状态图StataGraph实现：

# ========================# 🧭 路由函数# ========================def route_to_next_agent(state: State) -> Literal["sql", "analysis", "rewrite", "__end__"]:    if state.get("next_agent") == "sql":        return "sql"    if state.get("next_agent") == "analysis":        return "analysis"    if state.get("next_agent") == "rewrite":        return "rewrite"    return "__end__"    workflow = StateGraph(State)workflow.add_node("template", template_agent)workflow.add_node("rewrite", rewrite_agent)   # 新增改写智能体workflow.add_node("sql", sql_agent)workflow.add_node("analysis", analysis_agent)workflow.add_edge(START, "template")# Template 之后根据条件流向workflow.add_conditional_edges(    "template",    route_to_next_agent,    {        "rewrite": "rewrite",     # 新增 rewrite 分支        "sql": "sql",        "analysis": "analysis",        "__end__": END    })# 改写智能体完成后交给 SQLworkflow.add_edge("rewrite", "sql")# SQL 执行后流向分析workflow.add_edge("sql", "analysis")workflow.add_edge("analysis", END)# ========================# ▶️ 主流程# ========================if __name__ == "__main__":    from uuid import uuid4    print("🚀 智能数据助手（支持追问、多轮上下文）")    print("=" *80)    print("💡 输入问题开始对话，输入 'exit' 退出。")    print("=" *80)    # 每次启动生成唯一 thread_id（上下文记忆关键）    thread_id = str(uuid4())    config = {"configurable": {"thread_id": thread_id}}    while True:        user_input = input("\n👤 你: ").strip()        if not user_input:            continue        if user_input.lower() in {"exit", "quit"}:            print("👋 再见！会话已结束。")            break        # 构造输入        input_data = {            "messages": [HumanMessage(content=user_input)],        }        try:            # 流式执行            for chunk in app.stream(input_data, config=config):                if "analysis" in chunk:                    msg = chunk["analysis"]["messages"][0]                    print(f"💬 助手: {msg.content.strip()}")                    print("-" *60)                elif "sql" in chunk and "messages" in chunk["sql"]:                    # SQL 生成失败的情况                    for m in chunk["sql"]["messages"]:                        if isinstance(m, AIMessage):                            print(f"💬 助手: {m.content.strip()}")                            print("-" *60)                elif "__end__" in chunk:                    print("✅ 对话结束")        except json.JSONDecodeError:            print("⚠️ JSON 解析失败，可能是 LLM 返回了非结构化内容。")        except KeyboardInterrupt:            print("\n⏹️ 中断。输入 exit 可退出。")        except Exception as e:            print(f"💥 出错：{e}")

如以上代码所示，通过“workflow.add_node”方式添加四个Node节点：template、rewrite、sql、analysis分别代表四个智能体。

通过“workflow.add_edge”方式添加对应节点之间的跳转关系，其中“workflow.add_conditional_edges”是添加条件边，通过调用route_to_next_agent()路由函数根据智能体返回的next_agent字段决定流转到哪一个节点。

🔁多轮对话之上下文记忆和状态机

LangGraph上下文记忆的相关机制，主要通过官方提供的 MemorySaver() 实现：

# 启用记忆（支持多轮对话）memory = MemorySaver()app = workflow.compile(checkpointer=memory)

MemorySaver会将每一个节点执行后的 State 存储到检查点（checkpoint）中，使得下一次调用可以从之前的对话状态继续执行。

其特点是原样存 储state状态机，不裁剪、不删减、不合并历史，真正关键的是你在 State 状态机中设计的业务逻辑。

使用“ StateGraph(State)”方式在状态图StateGraph初始化时就将自定义State状态机对象注入并绑定：

# ========================# 🧠 增强状态定义# ========================class State(TypedDict):    messages: Annotated[list[BaseMessage], add_messages]    original_intent: Optional[str]  # 如 "考生人数"    known_table_struct: Optional[List]  # 已确认的表结构    last_sql_template: Optional[str]  # 带占位符的 SQL 模板，如 "SELECT ... WHERE year = {year}"    query_result: Optional[list]  # 最近一次查询结果    next_agent: str

每轮对话都会在全局统一状态机中保存以下信息：

key	含义
messages	历史消息列表
original_intent	用户的核心意图（如“考生人数”）
known_table_struct	已识别的数据表结构
last_sql_template	参数化 SQL 模板，便于复用
query_result	查询结果
next_agent	下一步路由方向

其中当我们使用 Annotated[list[BaseMessage], add_messages] 来定义 State 中的 messages 字段时，新的消息会自动被追加到现有的消息列表之后，这是 LangGraph 为管理对话历史等序列化数据提供的一种便捷且强大的状态更新机制。

LangGraph所有智能体Agent都要遵循统一规范，将Stata对象作为输入输出对象，如以下“模板匹配智能体”templateAgent的代码所示：

# ========================# 🍻模板匹配 智能体（支持追问模板复用）# ========================def template_agent(state: State):    print("=== 🟦 Template Agent ===")    current_query = state["messages"][-1].content    # ───────────────────────────────────────    # 🔄 情况1：无上下文 → SQL Agent    # ───────────────────────────────────────    if (not state.get("known_table_struct")            or not state.get("original_intent")            or not state.get("last_sql_template")):        print("⚠️ 不存在上下文，进入SQL Agent流程")        return {            "messages": state["messages"],            "next_agent": "sql"        }    # ───────────────────────────────────────    # 🔄 情况2：已有上下文 → 尝试参数化追问处理    # ───────────────────────────────────────    print("🔄 检测到历史上下文，尝试参数化追问")    try:        # 具体业务代码跳过
``````plaintext
print(f"✅ 追问查询成功，进入分析流程")                return {                    "query_result": result,                    "next_agent": "analysis"                }
``````plaintext
except Exception as e:        print(f"参数提取失败: {e}")    print("⚠️ 追问处理失败，进入改写流程")    return {        "messages": state["messages"],        "next_agent": "rewrite"    }

案例测试结果

我们测试一下LangGraph实现多智能体协作的效果：

🚀 智能数据助手（支持追问、多轮上下文）================================================================================💡 输入问题开始对话，输入 'exit' 退出。================================================================================👤 你: 2016年考生人数有多少？=== 🟦 Template Agent ===⚠️ 不存在上下文，进入SQL Agent流程=== 🟦 Sql Agent ===🆕 启动完整查询流程调用大模型llama2向量化：2016年考生人数有多少？📚 匹配表结构: ['{"表名": "college_entrance_examination", "表备注": "考生人数与复读人数信息表，包含字段：高考年份(主键)、考生人数(万人)、复读人数(万人)", "字段列表": [{"字段名": "examination_year", "字段类型": "int", "字段备注": "高考年份"}, {"字段名": "candidates_count", "字段类型": "decimal(10,2)", "字段备注": "考生人数(万人)"}, {"字段名": "retake_count", "字段类型": "decimal(10,2)", "字段备注": "复读人数(万人)"}]}']🧠 原始意图: 2016年考生人数有多少？📦 保存 SQL 模板: SELECT candidates_count AS '考生人数' FROM college_entrance_examination WHERE examination_year = {year};=== 🟩 Analysis Agent ===💬 助手: **2016年高考考生人数分析报告**  **事实陈述**  2016年高考考生人数为 **940.00万人**（数据来源：考生人数与复读人数信息表）。**简要洞察**  当前数据仅包含单一时间点的记录，暂无法直接推导趋势变化。若需分析考生人数增减规律，建议结合相邻年份数据（如2015年及2017年）进行对比。**备注**  数据单位为“万人”，数值保留两位小数，统计口径可能包含应届生与复读生。如需进一步解读，可补充复读人数或分省数据辅助分析。---  （报告基于提供的数据生成，内容简洁客观，避免过度推测。）------------------------------------------------------------👤 你: 2018年呢？=== 🟦 Template Agent ===🔄 检测到历史上下文，尝试参数化追问🔍 提取参数: {"year": "2018"}🛠️ 生成新 SQL: SELECT candidates_count AS '考生人数' FROM college_entrance_examination WHERE examination_year = 2018;✅ 追问查询成功，返回 1 行=== 🟩 Analysis Agent ===💬 助手: **2016年高考考生人数分析报告**  **事实陈述**  2016年高考考生人数为 **975.0万人**（数据来源：考生人数与复读人数信息表）。**简要洞察**  1. 当前数据仅包含单一时间点的记录，暂无法分析趋势变化。若需观察考生人数增减趋势，建议补充相邻年份数据（如2015年及2017年数据）。  2. 表中还包含“复读人数”字段，若需进一步分析复读对考生总数的影响，可提供对应数据后补充说明。**备注**  数据单位为“万人”，实际考生人数为9,750,000人。如需更深入的统计分析（如增长率、区域对比等），请提供扩展数据集。---  **报告说明**：本报告基于提供的数据直接呈现核心结论，避免对单一数据点进行推测性解读。------------------------------------------------------------👤 你: 录取人数呢？=== 🟦 Template Agent ===🔄 检测到历史上下文，尝试参数化追问🔍 提取参数: {}⚠️ 年份参数提取失败，进入改写流程=== 🟨 Rewrite Agent ===✍️ 改写后问题: 2018年全国高考录取人数是多少？=== 🟦 Sql Agent ===🆕 启动完整查询流程调用大模型llama2向量化：2018年全国高考录取人数是多少？📚 匹配表结构: ['{"表名": "college_entrance_admission", "表备注": "录取人数与普通高校数信息表，包含字段：录取年份(主键)、录取人数(万人)、招生高校数、本科录取人数(万人)、专科录取人数(万人)", "字段列表": [{"字段名": "admission_year", "字段类型": "int", "字段备注": "录取年份"}, {"字段名": "admission_count", "字段类型": "decimal(10,2)", "字段备注": "录取人数(万人)"}, {"字段名": "university_count", "字段类型": "int", "字段备注": "招生高校数"}, {"字段名": "undergraduate_admission_count", "字段类型": "decimal(10,2)", "字段备注": "本科录取人数(万人)"}, {"字段名": "specialty_admission_count", "字段类型": "decimal(10,2)", "字段备注": "专科录取人数(万人)"}]}']🧠 原始意图: 2018年全国高考录取人数是多少？📦 保存 SQL 模板: SELECT admission_count AS '考生人数' FROM college_entrance_admission WHERE admission_year = {year};=== 🟩 Analysis Agent ===💬 助手: **2018年全国高考录取人数分析报告**  **事实陈述**  根据数据表 `college_entrance_admission` 记录，**2018年全国高考录取人数为 790.99 万人**。**简要洞察**  1. **数据背景**：该数据反映当年普通高校（含本科及专科）录取总规模，但未提供与考生人数、录取率的对比信息。  2. **趋势参考**：若结合近年数据（需补充），可进一步分析录取人数变化趋势；当前单一数据点建议谨慎用于预测。  3. **结构细化**：表中包含本科与专科分层数据字段（未在本次结果中体现），后续可针对学历层次结构进行深入分析。**注意事项**  - 当前数据为单一数值，分析需避免过度推论。  - 如需评估录取竞争程度，需补充同年高考报名人数或招生计划数据。---  **报告说明**：本分析基于提供的数据表结构及查询结果，确保表述与原始数据一致。------------------------------------------------------------👤 你:

第一轮问题：2016年考生人数有多少？

第一轮没有上下文，也没有全局变量模板，所以直接走SQL Agent + Analysis AgentRAG检索、生成SQL、查数、分析。同时SQL Agent生成成功后记录SQL模板到全局state。

第二轮问题：2018年呢？

第二轮我们追问2018年，这时存在全局变量模板，所以走模板匹配提取年份参数，直接查数然后通过业务判断存在数据，走Analysis Agent分析。

第三轮问题：录取人数呢？

第三轮我们追问录取人数，但是匹配不到模板，所以上层业务判断走Rewrite Agent + SQL Agent + Analysis Agent改写问题完善条件。改写时，参考Memory中上下文2016和2018年回答，改写为2018年全国高考录取人数是多少？然后走RAG检索、生成SQL、查数、分析。

1. 🧩LangGraph实现小结

本文使用LangGraph的Custom图模式实现我们的多智能体协作处理业务：

• 通过StataGraph状态图来管理Node节点（对应智能体Agent）+edge边（对应执行路径）；
• 通过Checkpoint Memory实现长时记忆，管理上下文和结构化参数（消息、SQL 模板、表结构、查询结果等），所有agent都要遵循一定规范，将状态机作为输入和输出。
• 相对来说，上一篇文章中介绍AgentScope框架更松散，其多Agent协作机制是通过自定义控制逻辑+管道pipeline来串联，通过框架提供的memory只能管理多轮对话的上下文，多智能体多轮次之间的参数传递，需要通过自定义全局变量来实现。

1. 附录：LangGraph几种典型协作模式

LangGraph 是基于LangChain生态构建，它强调「状态图（StateGraph）」和「可追溯记忆（Checkpoint Memory）」，它提出了几种Multi-Agent模式，适用于不同的调用逻辑和控制流需求：

网络（Network）模式

• 各Agent节点彼此多对多通信：每个Agent有权限决定接下来调用哪个其他Agent。
• 适用于“没有固定顺序”“较灵活的调用关系”的场景。
• 代码示例中，Agent agent_1、agent_2、agent_3 均可跳转至其他两个或结束。

主管（Supervisor）模式

• 一个“主管Agent”负责决定下一步调用哪个子Agent。子Agent执行后再回到主管。
• 适用于“中央调度控制”“明确谁来决定流程”的场景。
• 示例：supervisor 决定调用 agent_1 或 agent_2。

主管（Supervisor as Tools 工具调用）模式

• 是“主管”模式的一个特例：子Agent被暴露为“工具”（tool），主管Agent像工具调用系统一样，决定调用哪个工具／子Agent。
• 典型流程：主管以 ReAct （思考→行动→思考）方式循环调用工具直到决策结束。
• 提供了 create_react_agent 等预置组件。

分层（Hierarchical）模式

• 当Agent数量增多、流程更复杂时，用“主管的主管”来管理团队／子系统，形成层级结构。
• 每个团队内有自己的主管+子Agent，再由顶层主管决定调用哪个团队。
• 适合更大规模、跨领域、多团队协作的系统。

自定义（Custom Workflow）模式（对应本文案例）

• 你也可以手动定义Agent节点及其顺序（显式控制流）或部分由 LLM 决定（动态控制流）。
• 显式控制流：在图中预先定义边 agent_1 → agent_2 → agent_3。
• 适合需要「强控制流」「复杂状态管理」「上下文复用」的场景。

如何学习AI大模型？

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

这份完整版的大模型 AI 学习和面试资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】

第一阶段： 从大模型系统设计入手，讲解大模型的主要方法；

第二阶段： 在通过大模型提示词工程从Prompts角度入手更好发挥模型的作用；

第三阶段： 大模型平台应用开发借助阿里云PAI平台构建电商领域虚拟试衣系统；

第四阶段： 大模型知识库应用开发以LangChain框架为例，构建物流行业咨询智能问答系统；

第五阶段： 大模型微调开发借助以大健康、新零售、新媒体领域构建适合当前领域大模型；

第六阶段： 以SD多模态大模型为主，搭建了文生图小程序案例；

第七阶段： 以大模型平台应用与开发为主，通过星火大模型，文心大模型等成熟大模型构建大模型行业应用。

👉学会后的收获：👈

• 基于大模型全栈工程实现（前端、后端、产品经理、设计、数据分析等），通过这门课可获得不同能力；

• 能够利用大模型解决相关实际项目需求： 大数据时代，越来越多的企业和机构需要处理海量数据，利用大模型技术可以更好地处理这些数据，提高数据分析和决策的准确性。因此，掌握大模型应用开发技能，可以让程序员更好地应对实际项目需求；

• 基于大模型和企业数据AI应用开发，实现大模型理论、掌握GPU算力、硬件、LangChain开发框架和项目实战技能， 学会Fine-tuning垂直训练大模型（数据准备、数据蒸馏、大模型部署）一站式掌握；

• 能够完成时下热门大模型垂直领域模型训练能力，提高程序员的编码能力： 大模型应用开发需要掌握机器学习算法、深度学习框架等技术，这些技术的掌握可以提高程序员的编码能力和分析能力，让程序员更加熟练地编写高质量的代码。

1.AI大模型学习路线图
2.100套AI大模型商业化落地方案
3.100集大模型视频教程
4.200本大模型PDF书籍
5.LLM面试题合集
6.AI产品经理资源合集

👉获取方式：
😝有需要的小伙伴，可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓