在实际构建基于大语言模型的智能应用时很多开发者会遇到一个瓶颈简单的链式调用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。整个工作流的执行过程就是状态在不同节点间传递和演化的过程。状态是一个字典TypedDict你需要预先定义一个 Python 类型通常使用TypedDict来声明你的工作流状态包含哪些字段。例如一个问答工作流的状态可能包含question用户问题、context检索到的背景知识、answer生成的答案、next_step决定下一步做什么等。节点是状态的处理器每个节点都是一个函数它接收当前完整的状态字典对其进行读取、修改或添加新字段然后返回更新后的状态或状态的部分字段。边决定了状态的流向边可以是有条件的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 环境与工具要求确保你的开发环境满足以下要求组件要求说明Python3.10 或更高版本LangChain/LangGraph 对 Python 版本有要求。包管理pip 或 poetry推荐使用虚拟环境。向量数据库Milvus 或 Chroma本文示例使用轻量级的 Chroma内存模式。大模型 APIOpenAI 或 智谱AI 等需要相应的 API Key。代码编辑器VS Code, PyCharm 等任意你熟悉的 IDE。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文件并填入以下核心依赖langchain0.1.0 langchain-community0.0.10 langgraph0.0.50 chromadb0.4.22 tiktoken0.5.0 pydantic2.0.0 python-dotenv1.0.0 # 根据你使用的模型选择 openai1.6.0 langchain-openai0.0.5 # 或者使用智谱AI zhipuai2.0.0 # 或者使用DeepSeek httpx0.25.0执行安装pip install -r requirements.txt2.3 配置文件与密钥管理在项目根目录创建.env文件用于存储敏感信息切勿提交到版本控制系统。# .env # OpenAI 配置 OPENAI_API_KEYyour_openai_api_key_here OPENAI_BASE_URLhttps://api.openai.com/v1 # 如果使用官方接口 # 智谱AI 配置 (可选) ZHIPUAI_API_KEYyour_zhipuai_api_key_here # DeepSeek 配置 (可选) DEEPSEEK_API_KEYyour_deepseek_api_key_here DEEPSEEK_BASE_URLhttps://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( modelconfig.OPENAI_MODEL, api_keyconfig.OPENAI_API_KEY, base_urlconfig.OPENAI_BASE_URL, temperaturetemperature ) elif provider zhipu: if not config.ZHIPUAI_API_KEY: raise ValueError(ZHIPUAI_API_KEY 未配置) # LangChain 社区版可能更新请参考最新文档 return ChatZhipuAI( modelconfig.ZHIPUAI_MODEL, api_keyconfig.ZHIPUAI_API_KEY, temperaturetemperature ) elif provider deepseek: # 示例通过 OpenAI 兼容接口调用 DeepSeek if not config.DEEPSEEK_API_KEY: raise ValueError(DEEPSEEK_API_KEY 未配置) return ChatOpenAI( modelconfig.DEEPSEEK_MODEL, api_keyconfig.DEEPSEEK_API_KEY, base_urlconfig.DEEPSEEK_BASE_URL, temperaturetemperature ) 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( modelconfig.EMBEDDING_MODEL, api_keyconfig.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_okTrue) vector_store Chroma( collection_namecollection_name, embedding_functionembedding_model, persist_directorypersist_dir ) return vector_store def add_documents_to_store(texts: list, collection_name: str rag_docs): 将文本分割并存入向量数据库初始化知识库用 text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, 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(temperature0.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 state4. 组装 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_pathworkflow_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_nametech_kb) print(知识库初始化完成。)运行它python init_kb.py5.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 提供了强大的调试支持。你可以通过设置debugTrue或使用 LangGraph Studio一个Web UI来查看每一步的状态变化。# 在 app.invoke 时开启简单调试 final_state app.invoke(initial_state, config{configurable: {thread_id: test-1}, recursion_limit: 50}) # 查看执行后的状态流需要更复杂的设置 # 或者将中间状态打印出来在生产环境中你应该将关键的状态转换和节点输入输出记录到日志系统中便于问题排查。6. 常见问题排查与优化实践在实际部署中你会遇到各种问题。以下是一些典型场景的排查路径和优化建议。6.1 问题排查清单问题现象可能原因检查步骤解决方案图编译失败提示状态字段错误1.TypedDict定义与节点返回值不匹配。2. 字段合并策略Annotated使用错误。1. 检查GraphState每个字段的类型。2. 检查节点函数返回的字典键是否与GraphState一致。3. 确认operator.add是否用于数值或列表累加字符串通常直接覆盖。修正TypedDict定义或节点返回值。对于简单覆盖Annotated[str, operator.add]也可工作但理解其语义。检索节点返回空上下文1. 向量数据库为空或未正确持久化。2. 嵌入模型与建库时不同。3. 问题与知识库内容语义不匹配。4.top_k参数太小。1. 运行init_kb.py确认数据已入库。2. 检查config.EMBEDDING_MODEL是否一致。3. 直接运行vector_store.similarity_search(question)测试。4. 调整RETRIEVE_TOP_K。1. 重新初始化知识库。2. 确保嵌入模型配置一致。3. 优化知识库文档或尝试重写用户问题。4. 增大top_k但需权衡性能与精度。答案生成质量差胡言乱语1. 提示词Prompt设计不佳。2. 上下文未正确传递给 LLM。3. LLM 的temperature参数过高。4. 上下文过多导致模型注意力分散。1. 打印出最终发送给 LLM 的完整提示词。2. 检查formatted_context是否正确拼接。3. 降低temperature到 0.1 或 0.2。4. 检查检索到的上下文是否相关。1. 优化系统提示和上下文格式。2. 确保上下文被正确格式化并放入提示词占位符。3. 使用更低的temperature。4. 使用LLMChainFilter或ContextualCompressionRetriever对上下文进行压缩和过滤。工作流陷入死循环1. 条件边conditional_edges的路由逻辑有误导致在两个节点间来回跳转。2. 节点未正确设置next_step。1. 在节点函数中打印state[next_step]。2. 检查所有条件边的路由函数返回值是否都在目标映射中。3. 设置recursion_limit参数。1. 仔细检查每个节点修改next_step的逻辑。2. 确保路由函数返回的字符串与add_conditional_edges中定义的键完全匹配。3. 在app.invoke中设置recursion_limit。调用 LLM API 超时或报错1. 网络问题。2. API Key 无效或余额不足。3. 模型名称错误。4. 请求频率超限。1. 检查网络连接。2. 在.env中确认 API Key 正确。3. 直接使用get_llm()实例调用一个简单对话测试。4. 查看 LLM 提供商的控制台或日志。1. 配置网络代理注意合规性。2. 更换或充值 API Key。3. 核对官方文档使用正确的模型名。4. 增加重试机制和退避策略使用tenacity库。6.2 生产环境最佳实践状态持久化上述示例的状态存在于单次内存调用中。对于长对话或需要中断恢复的场景需要将State持久化到数据库如 Redis、PostgreSQL。LangGraph 支持Checkpointer机制。异步支持LangGraph 天然支持异步。将节点函数定义为async def并使用ainvoke可以大幅提升 I/O 密集型操作如网络请求、数据库查询的并发性能。更复杂的路由可以引入一个专门的“路由节点”Router Node使用一个小型 LLM 或规则引擎来分析question并设置next_step为retrieve,direct_answer,call_tool等实现更智能的多路径工作流。加入人工审核节点在生成答案后可以增加一个human_review节点将答案挂起等待人工审核通过后再流向END。这在高风险领域非常有用。监控与可观测性为每个节点的执行时间、Token 消耗、API 调用状态添加监控和日志。使用langsmithLangChain 官方平台可以很好地追踪和评估链/图的运行情况。配置管理将模型参数、提示词模板、检索参数等外部化到配置文件如 YAML或配置中心避免硬编码。7. 扩展方向构建更复杂的多智能体系统本文展示的是一个双节点检索生成的简单 RAG 工作流。基于此架构你可以轻松扩展查询理解与重写节点在检索前先对用户问题进行澄清、扩展或重写提升检索命中率。多路检索与融合节点同时从不同知识库或使用不同检索策略如关键词向量获取信息然后进行去重和排序。答案验证与溯源节点生成答案后让另一个智能体检查答案中的事实是否与提供的上下文一致并标注出处。工具调用节点集成 LangChain Tools让智能体在需要时调用计算器、搜索引擎、数据库查询等外部工具。长期记忆节点引入ChatMessageHistory将历史对话摘要或关键信息存入状态实现跨轮次记忆。每个扩展都可以通过增加新的节点和定义它们与现有节点之间的边来实现。StateGraph 的这种模块化设计使得维护和迭代复杂系统变得清晰可控。通过本次实战你应该已经掌握了使用 LangGraph 和 LangChain 构建工业级多智能体应用的核心方法从状态设计、节点实现、图组装到运行调试。接下来你可以尝试将更多的业务逻辑封装成节点设计更精细的状态流转规则从而打造出真正强大、可靠的 AI 应用工作流。