LangGraph实战：构建多智能体RAG问答系统的状态驱动工作流

在实际构建基于大语言模型的智能应用时，很多开发者会遇到一个瓶颈：简单的链式调用（Chain）难以处理复杂的、有状态的、需要多角色协作的业务流程。例如，一个智能客服系统，可能需要先理解用户意图，然后查询知识库，再根据历史对话判断是否需要转接给人工，整个过程涉及状态流转、条件判断和多个“智能体”（Agent）的协同。这正是 LangGraph 与 LangChain 结合所要解决的核心问题。

LangChain 提供了丰富的组件（如模型封装、提示词模板、记忆、检索器等）来连接大模型与外部数据，但它本质上是一个编排框架。而 LangGraph 是 LangChain 生态系统中的一个库，它引入了有向图的概念，特别擅长描述和管理有状态、多步骤、带循环和条件分支的工作流。StateGraph 是其核心，它允许你明确定义工作流中每个节点的行为以及节点间状态如何流转。

本文将以一个工业级的“多智能体 RAG 问答系统”为蓝本，带你从零开始，深度拆解如何利用 LangGraph 的 StateGraph 进行状态管控，整合多个具备不同能力的智能体，并接入 RAG 与多模型，最终实现一个可落地、可维护的复杂应用。你将理解从架构设计、状态定义、节点实现到运行调试的完整闭环。

1. 理解 LangGraph 与 StateGraph 的核心机制

在开始编码之前，必须厘清几个核心概念，这决定了你能否正确设计工作流。

1.1 LangGraph 是什么？与 LangChain 有何关系？

LangGraph 不是一个独立的框架，它是 LangChain 框架的一个扩展库。你可以把它想象成 LangChain 的“工作流引擎”。

• LangChain的核心价值在于“链”（Chain），它将多个组件（LLM、提示词、工具、记忆等）线性地组合起来执行一个任务。链适合顺序明确的流程，但对于需要循环（如反复追问用户）、条件分支（如根据结果选择不同处理路径）或并行执行的任务，原生的链就显得力不从心。
• LangGraph则引入了“图”（Graph）的概念。在图里，每个节点（Node）是一个可执行单元（可以是一个简单的函数，也可以是一个复杂的 LangChain Chain 或 Agent），边（Edge）定义了节点之间的流转路径。StateGraph 是 LangGraph 中用于管理图状态的核心类。

简单类比：LangChain 像是一条装配线，而 LangGraph 像是一个由多条装配线和调度中心组成的智能工厂，能够根据产品（输入）的不同，动态决定走哪条线，并且可以来回加工。

1.2 StateGraph 与状态（State）的设计哲学

StateGraph 的关键在于“状态”（State）。整个工作流的执行过程，就是状态在不同节点间传递和演化的过程。

1. 状态是一个字典（TypedDict）：你需要预先定义一个 Python 类型（通常使用TypedDict），来声明你的工作流状态包含哪些字段。例如，一个问答工作流的状态可能包含：question（用户问题）、context（检索到的背景知识）、answer（生成的答案）、next_step（决定下一步做什么）等。
2. 节点是状态的处理器：每个节点都是一个函数，它接收当前完整的状态字典，对其进行读取、修改或添加新字段，然后返回更新后的状态（或状态的部分字段）。
3. 边决定了状态的流向：边可以是有条件的（conditional edge）或无条件的（normal edge）。无条件边总是将状态传递给下一个节点。有条件边则根据状态中的某个字段（如next_step）的值，决定下一步跳转到哪个节点。

这种设计将工作流的控制逻辑（流程图）与业务逻辑（每个节点的处理函数）清晰地分离开，使得复杂的流程变得可描述、可维护和可调试。

1.3 多智能体（Multi-Agent）在 LangGraph 中的体现

在 LangGraph 的语境下，“智能体”可以理解为具备特定职责的节点。一个节点可以封装一个完整的 LangChain Agent（拥有工具和推理能力），也可以是一个更简单的函数。

多智能体协作就是通过设计不同的节点，并规划好它们之间的流转路径来实现的。例如：

• 路由智能体（Router Agent）：分析用户问题，决定是进行普通对话、知识库查询还是需要专项工具处理。
• 检索智能体（Retrieval Agent）：专门负责从向量数据库（如 Milvus, Chroma）中检索相关文档。
• 解答智能体（Answering Agent）：根据问题和检索到的上下文，生成最终答案。
• 验证智能体（Validation Agent）：对生成的答案进行事实性或安全性检查。

这些“智能体”作为图中的节点，通过共享和修改同一个 State 对象来协同工作。

2. 环境准备与项目依赖配置

我们将构建一个项目，它使用 LangGraph 协调一个检索智能体和一个解答智能体，并可以灵活接入 OpenAI GPT 或国内大模型（如智谱 AI、DeepSeek）。

2.1 环境与工具要求

确保你的开发环境满足以下要求：

2.2 创建项目与安装依赖

创建一个新的项目目录，并初始化虚拟环境。

mkdir langgraph-multiagent-rag cd langgraph-multiagent-rag python -m venv venv # 激活虚拟环境 (Windows) venv\Scripts\activate # 激活虚拟环境 (MacOS/Linux) source venv/bin/activate

创建requirements.txt文件，并填入以下核心依赖：

langchain>=0.1.0 langchain-community>=0.0.10 langgraph>=0.0.50 chromadb>=0.4.22 tiktoken>=0.5.0 pydantic>=2.0.0 python-dotenv>=1.0.0 # 根据你使用的模型选择 openai>=1.6.0 langchain-openai>=0.0.5 # 或者使用智谱AI zhipuai>=2.0.0 # 或者使用DeepSeek httpx>=0.25.0

执行安装：

pip install -r requirements.txt

2.3 配置文件与密钥管理

在项目根目录创建.env文件，用于存储敏感信息，切勿提交到版本控制系统。

# .env # OpenAI 配置 OPENAI_API_KEY=your_openai_api_key_here OPENAI_BASE_URL=https://api.openai.com/v1 # 如果使用官方接口 # 智谱AI 配置 (可选) ZHIPUAI_API_KEY=your_zhipuai_api_key_here # DeepSeek 配置 (可选) DEEPSEEK_API_KEY=your_deepseek_api_key_here DEEPSEEK_BASE_URL=https://api.deepseek.com

同时，创建一个config.py文件来集中管理配置。

# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 class Config: # 模型选择： openai, zhipu, deepseek LLM_PROVIDER = os.getenv("LLM_PROVIDER", "openai") # OpenAI 配置 OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-3.5-turbo") OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL") # 智谱AI 配置 ZHIPUAI_API_KEY = os.getenv("ZHIPUAI_API_KEY") ZHIPUAI_MODEL = os.getenv("ZHIPUAI_MODEL", "glm-4") # DeepSeek 配置 DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY") DEEPSEEK_MODEL = os.getenv("DEEPSEEK_MODEL", "deepseek-chat") DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL") # 向量数据库配置 VECTOR_DB_TYPE = os.getenv("VECTOR_DB_TYPE", "chroma") # chroma 或 milvus CHROMA_PERSIST_DIR = os.getenv("CHROMA_PERSIST_DIR", "./chroma_db") # RAG 相关 EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "text-embedding-3-small") # OpenAI 嵌入模型 RETRIEVE_TOP_K = int(os.getenv("RETRIEVE_TOP_K", 3)) config = Config()

3. 定义工作流状态与构建智能体节点

这是 LangGraph 应用的核心设计阶段。我们将定义一个状态，并创建两个核心智能体节点：检索节点和解答节点。

3.1 定义状态类型（State Schema）

在state.py中，我们使用TypedDict来定义工作流的“数据容器”。

# state.py from typing import TypedDict, List, Optional, Annotated import operator from typing_extensions import TypedDict # 定义整个图的状态结构 class GraphState(TypedDict): """ 图的状态，在所有节点间共享和传递。 """ # 输入 question: str # 用户原始问题 # 处理过程 context: Optional[List[str]] # 检索到的文档片段列表 answer: Optional[str] # 生成的最终答案 # 控制流 next_step: Annotated[str, operator.add] # 决定下一步做什么。operator.add 表示该字段可被多个节点追加（本例中简单覆盖即可） # 错误与元数据 error: Optional[str] # 错误信息 retrieval_attempts: Annotated[int, operator.add] # 检索尝试次数，用于防止死循环

关键解释：

• question,context,answer是业务数据字段。
• next_step是一个关键的控制字段，节点可以通过修改它来影响工作流走向。Annotated和operator.add是 LangGraph 的语法，用于声明字段的合并策略（这里是字符串覆盖，但使用add是常见写法）。
• error和retrieval_attempts用于错误处理和流程控制。

3.2 构建模型工厂与工具函数

为了支持多模型接入，我们创建一个模型工厂。同时，创建一些工具函数，如文本分割和向量数据库初始化。

# llm_factory.py from langchain_openai import ChatOpenAI from langchain_community.chat_models import ChatZhipuAI # 注意：DeepSeek 可能需要自定义封装或使用社区集成 from config import config import warnings def get_llm(model_type: str = None, temperature: float = 0.1): """获取大语言模型实例""" provider = model_type or config.LLM_PROVIDER if provider == "openai": if not config.OPENAI_API_KEY: raise ValueError("OPENAI_API_KEY 未配置") return ChatOpenAI( model=config.OPENAI_MODEL, api_key=config.OPENAI_API_KEY, base_url=config.OPENAI_BASE_URL, temperature=temperature ) elif provider == "zhipu": if not config.ZHIPUAI_API_KEY: raise ValueError("ZHIPUAI_API_KEY 未配置") # LangChain 社区版可能更新，请参考最新文档 return ChatZhipuAI( model=config.ZHIPUAI_MODEL, api_key=config.ZHIPUAI_API_KEY, temperature=temperature ) elif provider == "deepseek": # 示例：通过 OpenAI 兼容接口调用 DeepSeek if not config.DEEPSEEK_API_KEY: raise ValueError("DEEPSEEK_API_KEY 未配置") return ChatOpenAI( model=config.DEEPSEEK_MODEL, api_key=config.DEEPSEEK_API_KEY, base_url=config.DEEPSEEK_BASE_URL, temperature=temperature ) else: raise ValueError(f"不支持的模型提供商: {provider}") def get_embedding_model(): """获取文本嵌入模型实例（用于向量化）""" # 这里以 OpenAI 为例 from langchain_openai import OpenAIEmbeddings if not config.OPENAI_API_KEY: raise ValueError("OPENAI_API_KEY 未配置，无法初始化嵌入模型") return OpenAIEmbeddings( model=config.EMBEDDING_MODEL, api_key=config.OPENAI_API_KEY )

# vector_store.py from langchain_community.vectorstores import Chroma from langchain.text_splitter import RecursiveCharacterTextSplitter from llm_factory import get_embedding_model from config import config import os def get_vector_store(collection_name: str = "rag_docs"): """获取或创建向量数据库连接""" embedding_model = get_embedding_model() persist_dir = config.CHROMA_PERSIST_DIR # 确保目录存在 os.makedirs(persist_dir, exist_ok=True) vector_store = Chroma( collection_name=collection_name, embedding_function=embedding_model, persist_directory=persist_dir ) return vector_store def add_documents_to_store(texts: list, collection_name: str = "rag_docs"): """将文本分割并存入向量数据库（初始化知识库用）""" text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, separators=["\n\n", "\n", "。", "！", "？", "，", " ", ""] ) splits = text_splitter.split_text("\n".join(texts)) if isinstance(texts[0], str) else text_splitter.split_documents(texts) vector_store = get_vector_store(collection_name) vector_store.add_texts(splits) # 或 add_documents vector_store.persist() print(f"已添加 {len(splits)} 个文本块到向量库。")

3.3 实现智能体节点函数

现在，我们实现两个核心节点函数：retrieve和generate_answer。每个函数都接收GraphState并返回更新后的状态。

# nodes.py from state import GraphState from vector_store import get_vector_store from llm_factory import get_llm from langchain_core.prompts import ChatPromptTemplate from config import config from typing import List def retrieve(state: GraphState) -> GraphState: """检索节点：从向量数据库中获取与问题相关的上下文""" print(f"--- 执行检索 ---") question = state["question"] # 防止无限检索循环 if state.get("retrieval_attempts", 0) > 2: state["error"] = "检索尝试次数过多，可能知识库中无相关答案。" state["next_step"] = "error" return state try: vector_store = get_vector_store() # 相似度检索 retriever = vector_store.as_retriever(search_kwargs={"k": config.RETRIEVE_TOP_K}) docs = retriever.invoke(question) # 或 get_relevant_documents(question) context_list = [doc.page_content for doc in docs] if docs else [] if not context_list: state["context"] = ["未找到相关背景知识。"] state["next_step"] = "generate" # 即使没找到，也尝试生成 else: state["context"] = context_list state["next_step"] = "generate" # 更新尝试次数 state["retrieval_attempts"] = state.get("retrieval_attempts", 0) + 1 except Exception as e: state["error"] = f"检索过程发生错误: {str(e)}" state["next_step"] = "error" return state def generate_answer(state: GraphState) -> GraphState: """解答节点：基于问题和上下文生成答案""" print(f"--- 生成答案 ---") question = state["question"] context = state.get("context", []) # 构建提示词 prompt_template = ChatPromptTemplate.from_messages([ ("system", "你是一个专业的助手，请严格根据提供的上下文来回答问题。如果上下文不包含答案，请如实说明你不知道。"), ("human", "上下文：\n{context}\n\n问题：{question}") ]) # 格式化上下文 formatted_context = "\n\n---\n\n".join(context) if context else "未提供上下文。" # 创建链 llm = get_llm(temperature=0.1) # 低 temperature 使输出更确定 chain = prompt_template | llm try: response = chain.invoke({"context": formatted_context, "question": question}) answer = response.content state["answer"] = answer state["next_step"] = "end" # 正常结束 except Exception as e: state["error"] = f"答案生成过程发生错误: {str(e)}" state["next_step"] = "error" return state def error_handler(state: GraphState) -> GraphState: """错误处理节点""" print(f"--- 进入错误处理 ---") error_msg = state.get("error", "未知错误") state["answer"] = f"抱歉，处理您的问题时遇到了错误：{error_msg}。请稍后重试或简化您的问题。" state["next_step"] = "end" # 强制结束 return state

4. 组装 StateGraph 并编译为可执行应用

有了节点和状态定义，现在可以将它们组装成一个完整的工作流图。

4.1 创建图并添加节点与边

在graph_builder.py中，我们构建并编译这个图。

# graph_builder.py from langgraph.graph import StateGraph, END from state import GraphState from nodes import retrieve, generate_answer, error_handler def create_workflow_graph(): """创建并返回编译好的工作流图""" # 1. 初始化一个 StateGraph，指定状态类型 workflow = StateGraph(GraphState) # 2. 添加节点 workflow.add_node("retrieve", retrieve) # 检索节点 workflow.add_node("generate", generate_answer) # 生成节点 workflow.add_node("error", error_handler) # 错误处理节点 # 3. 设置入口点 workflow.set_entry_point("retrieve") # 4. 添加边（定义节点间的流转逻辑） # 从 retrieve 节点出来，根据状态中的 next_step 决定下一步 workflow.add_conditional_edges( "retrieve", # 这是一个路由函数，根据状态返回下一个节点的名称 lambda state: state.get("next_step", "generate"), { "generate": "generate", # 如果 next_step 是 "generate"，去 generate 节点 "error": "error", # 如果 next_step 是 "error"，去 error 节点 } ) # 从 generate 节点出来，通常直接结束 workflow.add_edge("generate", END) # 从 error 节点出来，也直接结束 workflow.add_edge("error", END) # 5. 编译图 app = workflow.compile() return app # 创建全局图应用实例 app = create_workflow_graph()

4.2 图的结构可视化

LangGraph 提供了可视化方法，可以帮助你理解工作流。在 Python 交互环境或脚本中添加：

# 可视化图（需要安装 graphviz） try: from IPython.display import Image, display # 如果你在 Jupyter Notebook 中 display(Image(app.get_graph().draw_mermaid_png())) except: # 或者生成文件 app.get_graph().draw_mermaid_png(output_file_path="workflow_graph.png") print("流程图已保存为 workflow_graph.png")

这将生成一个清晰的流程图，显示retrieve-> (条件判断) ->generate/error->END的路径。

5. 运行、验证与调试工作流

现在，我们可以初始化知识库，并运行这个多智能体 RAG 系统了。

5.1 初始化知识库（一次性操作）

创建一个脚本init_kb.py来向向量数据库添加一些示例文档。

# init_kb.py from vector_store import add_documents_to_store # 示例知识文档 sample_docs = [ "LangGraph 是 LangChain 的一个扩展库，用于构建有状态、多步骤的工作流。", "StateGraph 是 LangGraph 的核心类，它通过一个共享的状态字典来管理节点间的数据流。", "RAG (Retrieval-Augmented Generation) 是一种通过检索外部知识来增强大模型生成能力的技术。", "多智能体系统由多个具备特定功能的 Agent 组成，它们通过协作完成复杂任务。", "向量数据库，如 Chroma 或 Milvus，用于高效存储和检索文本的向量表示。" ] if __name__ == "__main__": add_documents_to_store(sample_docs, collection_name="tech_kb") print("知识库初始化完成。")

运行它：

python init_kb.py

5.2 创建主运行脚本并测试

创建main.py作为应用的入口点。

# main.py from graph_builder import app from state import GraphState def run_agent(question: str): """运行智能体工作流""" print(f"用户问题: {question}") print("-" * 40) # 初始化状态 initial_state: GraphState = { "question": question, "context": None, "answer": None, "next_step": "retrieve", # 从检索开始 "error": None, "retrieval_attempts": 0 } # 执行图 final_state = app.invoke(initial_state) print("-" * 40) print(f"最终答案: {final_state.get('answer')}") if final_state.get("error"): print(f"错误信息: {final_state.get('error')}") print(f"检索到的上下文: {final_state.get('context')}") return final_state if __name__ == "__main__": # 测试问题 test_questions = [ "LangGraph 是什么？", "RAG 技术有什么作用？", "告诉我一个不存在的知识，比如‘虚空引擎’的原理。" ] for q in test_questions: result = run_agent(q) print("\n" + "="*60 + "\n")

运行主程序：

python main.py

预期输出示例：

用户问题: LangGraph 是什么？ ---------------------------------------- --- 执行检索 --- --- 生成答案 --- ---------------------------------------- 最终答案: LangGraph 是 LangChain 的一个扩展库，用于构建有状态、多步骤的工作流。 检索到的上下文: ['LangGraph 是 LangChain 的一个扩展库，用于构建有状态、多步骤的工作流。', 'StateGraph 是 LangGraph 的核心类，它通过一个共享的状态字典来管理节点间的数据流。'] ============================================================ 用户问题: 告诉我一个不存在的知识，比如‘虚空引擎’的原理。 ---------------------------------------- --- 执行检索 --- --- 生成答案 --- ---------------------------------------- 最终答案: 根据提供的上下文，其中不包含关于“虚空引擎”的原理信息。我无法回答这个问题。 检索到的上下文: ['未找到相关背景知识。'] ============================================================

5.3 调试与状态追踪

LangGraph 提供了强大的调试支持。你可以通过设置debug=True或使用 LangGraph Studio（一个Web UI）来查看每一步的状态变化。

# 在 app.invoke 时开启简单调试 final_state = app.invoke(initial_state, config={"configurable": {"thread_id": "test-1"}, "recursion_limit": 50}) # 查看执行后的状态流（需要更复杂的设置） # 或者，将中间状态打印出来

在生产环境中，你应该将关键的状态转换和节点输入输出记录到日志系统中，便于问题排查。

6. 常见问题排查与优化实践

在实际部署中，你会遇到各种问题。以下是一些典型场景的排查路径和优化建议。

6.1 问题排查清单

6.2 生产环境最佳实践

1. 状态持久化：上述示例的状态存在于单次内存调用中。对于长对话或需要中断恢复的场景，需要将State持久化到数据库（如 Redis、PostgreSQL）。LangGraph 支持Checkpointer机制。
2. 异步支持：LangGraph 天然支持异步。将节点函数定义为async def，并使用ainvoke可以大幅提升 I/O 密集型操作（如网络请求、数据库查询）的并发性能。
3. 更复杂的路由：可以引入一个专门的“路由节点”（Router Node），使用一个小型 LLM 或规则引擎来分析question，并设置next_step为"retrieve","direct_answer","call_tool"等，实现更智能的多路径工作流。
4. 加入人工审核节点：在生成答案后，可以增加一个human_review节点，将答案挂起，等待人工审核通过后再流向END。这在高风险领域非常有用。
5. 监控与可观测性：为每个节点的执行时间、Token 消耗、API 调用状态添加监控和日志。使用langsmith（LangChain 官方平台）可以很好地追踪和评估链/图的运行情况。
6. 配置管理：将模型参数、提示词模板、检索参数等外部化到配置文件（如 YAML）或配置中心，避免硬编码。

7. 扩展方向：构建更复杂的多智能体系统

本文展示的是一个双节点（检索+生成）的简单 RAG 工作流。基于此架构，你可以轻松扩展：

• 查询理解与重写节点：在检索前，先对用户问题进行澄清、扩展或重写，提升检索命中率。
• 多路检索与融合节点：同时从不同知识库或使用不同检索策略（如关键词+向量）获取信息，然后进行去重和排序。
• 答案验证与溯源节点：生成答案后，让另一个智能体检查答案中的事实是否与提供的上下文一致，并标注出处。
• 工具调用节点：集成 LangChain Tools，让智能体在需要时调用计算器、搜索引擎、数据库查询等外部工具。
• 长期记忆节点：引入ChatMessageHistory，将历史对话摘要或关键信息存入状态，实现跨轮次记忆。

每个扩展都可以通过增加新的节点和定义它们与现有节点之间的边来实现。StateGraph 的这种模块化设计，使得维护和迭代复杂系统变得清晰可控。

通过本次实战，你应该已经掌握了使用 LangGraph 和 LangChain 构建工业级多智能体应用的核心方法：从状态设计、节点实现、图组装到运行调试。接下来，你可以尝试将更多的业务逻辑封装成节点，设计更精细的状态流转规则，从而打造出真正强大、可靠的 AI 应用工作流。