一、前言LangGraph 是 LangChain 生态中用于构建有状态 Agent 工作流的框架。和普通链式调用不同LangGraph 更强调“图”的概念我们可以把一个复杂任务拆成多个节点然后通过边、条件边、状态更新规则把这些节点组织成一个可控、可追踪、可扩展的执行流程。本文基于 LangChain官方文档 做出几个demo 代码进行知识点的相关总结与实践主要涉及以下内容StateGraph 的基础用法状态字段与 reducer 合并规则条件路由与分支流程LLM 工具调用 AgentRAG 检索增强生成流程ToolNode 与 tools_conditionOverwrite 覆盖状态输入输出 schema 隔离节点间私有状态传递这些 demo 虽然代码量不大但基本覆盖了 LangGraph 入门到 Agent 编排的核心知识。二、LangGraph 的核心思想LangGraph 的核心可以概括为一句话用图来组织任务流程用状态来承载上下文用节点来处理业务逻辑用边来控制执行方向。一个最基本的 LangGraph 程序通常包含四步fromlanggraph.graphimportStateGraphfromlanggraph.constantsimportSTART,END builderStateGraph(State)builder.add_node(node,node_func)builder.add_edge(START,node)builder.add_edge(node,END)graphbuilder.compile()其中StateGraph(State)定义图的状态结构。add_node添加处理节点。add_edge添加固定执行边。add_conditional_edges添加条件路由边。compile()编译成可执行图。invoke() / stream()执行图。LangGraph 并不是简单地把函数串起来而是每个节点都接收当前状态并返回状态更新。三、Demo1物流配送状态机Demo1用一个“包裹配送流程”演示了 LangGraph 最基础也最重要的能力状态流转和条件分支。定义状态代码中定义了一个 PackageStateclassPackageState(TypedDict):package_id:strorigin:strdestination:strnext:strstatus:strhistory:Annotated[list[str],operator.add]total_distance:Annotated[int,operator.add]priority:str这里最值得注意的是history:Annotated[list[str], operator.add]total_distance:Annotated[int, operator.add]这表示当节点返回新的 history 或 total_distance 时LangGraph 不会简单覆盖旧值而是使用 operator.add 进行合并。所以多个节点都可以追加物流轨迹return {“status”: “已揽收”,“history”: [f在{origin}进行揽收]}最终 history 会变成一条完整的配送链路。2. 定义节点这个 demo 中有几个典型节点receive_package揽收站sort_package分拣中心standard_delivery标准配送express_delivery加急配送final_delivery配送站每个节点都是一个普通函数defreceive_package(state:PackageState):originstate[origin]return{status:已揽收,history:[f在{origin}进行揽收]}节点只需要关心“读什么状态返回什么更新”不需要关心整个流程怎么走。3. 条件路由物流流程里最关键的是配送方式选择defselect_delivery(state:PackageState):prioritystate[priority]ifpriority加急:return备注加急else:return无备注然后用add_conditional_edges绑定路由结果delivery.add_conditional_edges(分拣中心,select_delivery,{备注加急:加急配送,无备注:标准配送})这就是 LangGraph 中非常典型的条件分支模式分拣中心 - 判断 priority - 加急配送 / 标准配送这个 demo 适合用来理解 LangGraph 的基本执行模型。四、Demo2带搜索工具的 AgentDemo2 演示了一个带 Tavily 搜索工具的 Agent。整体流程如下START↓llm_call↓是否需要工具调用├─ 是tool_node - llm_call└─ 否END定义消息状态classMessageState(TypedDict):messages:Annotated[List[AnyMessage],operator.add]llm_calls:Annotated[int,operator.add]这里 messages 保存对话消息llm_calls 记录模型调用次数。operator.add 表示每个节点返回的新消息会追加到历史消息中。2. 绑定工具代码使用 Tavily 搜索searchTavilySearch(max_results4,api_keyos.environ.get(TAVILY_API_KEY))tools[search]model_with_toolsmodel.bind_tools(tools)通过 bind_tools 后模型就可以在需要的时候返回工具调用请求。3. LLM 节点defllm_call(state:MessageState):messagesstate[messages]resultmodel_with_tools.invoke([SystemMessage(content你是一个乐于助人的助手支持工具调用进行搜索)]messages)return{messages:[result],llm_calls:state.get(llm_calls,0)1}该节点的作用是调用模型并把模型输出追加到 messages。如果模型认为需要搜索它会在返回消息中包含 tool_calls。4. 工具节点deftool_node(state:MessageState):result[]fortool_callinstate[messages][-1].tool_calls:tooltools_by_name[tool_call[name]]restool.invoke(tool_call[args])result.append(ToolMessage(contentres,tool_call_idtool_call[id]))return{messages:result}工具节点负责读取上一条 AI 消息中的 tool_calls。找到对应工具。执行工具。把工具结果包装成 ToolMessage。5. 条件判断defshould_continue(state:MessageState):last_msgstate[messages][-1]ifhasattr(last_msg,tool_calls)andlast_msg.tool_calls:returntool_nodereturnEND如果模型请求了工具调用就进入 tool_node否则流程结束。这就是最基础的 ReAct Agent 图结构。五、Demo3RAG 检索增强生成Demo3 实现了一个基于本地 Markdown 文档的 RAG Agent。它的流程可以总结为用户问题↓模型判断是否需要检索↓调用 retriever_tool↓判断检索文档是否相关├─ 相关生成答案└─ 不相关重写问题再次检索文档加载代码加载了本地 Markdown 文档paths[../LangGraph/docs/markdown/Linux.md]docs[UnstructuredMarkdownLoader(path).load()forpathinpaths]docs_list[itemforsublistindocsforiteminsublist]然后使用文本分割器切块text_splitterRecursiveCharacterTextSplitter.from_tiktoken_encoder(encoding_namecl100k_base,chunk_size1000,chunk_overlap50)doc_splitstext_splitter.split_documents(docs_list)这里的参数含义chunk_size1000每个文档块约 1000 token。chunk_overlap50相邻文档块保留 50 token 重叠减少语义断裂。2. 构建向量库embeddingsZhipuAIEmbeddings(modelembedding-3,api_keyos.getenv(ZHIPU_API_KEY))vectorstoreInMemoryVectorStore.from_documents(documentsdoc_splits,embeddingembeddings)retrievervectorstore.as_retriever(search_kwargs{k:2})这里使用的是内存向量库适合 demo 和本地测试。生产环境中可以替换成 Milvus、Chroma、pgvector、Elasticsearch 等持久化向量数据库。3. 创建检索工具retriever_toolcreate_retriever_tool(retriever,retrieve_bit,搜索并返回有关Linux的信息。,)创建工具后模型就可以通过工具调用的方式执行文档检索。4. 生成查询或直接回答defgenerate_query_or_respond(state:MessageState):ai_messagemodel.bind_tools([retriever_tool]).invoke(state[messages])return{messages:[ai_message]}这个节点让模型自己判断如果能直接回答就直接返回答案。如果需要查资料就调用 retrieve_bit 工具。5. 使用 ToolNode 执行工具retriever_nodeToolNode([retriever_tool])这里使用了 LangGraph 预置的 ToolNode。ToolNode 会自动读取 AI 消息中的工具调用并执行对应工具。6. 使用 tools_condition 判断是否调用工具workflow.add_conditional_edges(generate_query_or_respond,tools_condition,{tools:retrieve,__end__:END,})tools_condition是 LangGraph 预置的条件函数。它会判断最后一条 AI 消息中是否包含工具调用有工具调用进入 retrieve没有工具调用结束7. 文档相关性评分RAG 中常见问题是检索出来的内容不一定和用户问题相关。这个 demo 使用结构化输出对文档进行评分classGradeDocuments(BaseModel):score:strField(description相关性评分相关为yes,不相关为no)评分节点defgrade_documents(state:MessageState)-Literal[generate_answer,rewrite_question]:questionstate[messages][0].content contextstate[messages][-1].content promptGRADE_PROMPT.format(contextcontext,questionquestion)resultmodel.with_structured_output(GradeDocuments).invoke([HumanMessage(contentprompt)])ifresult.scoreyes:returngenerate_answerelse:returnrewrite_question如果文档相关就进入答案生成节点如果不相关就重写问题后重新检索。8. 问题重写defrewrite_question(state:MessageState):questionstate[messages][0].content promptREWRITE_PROMPT.format(questionquestion)resultmodel.invoke([HumanMessage(contentprompt)])return{messages:[HumanMessage(contentresult.content)]}问题重写是 RAG 中非常实用的策略尤其适合用户问题比较模糊、关键词不准确、语义不完整的情况。9. 基于上下文生成答案defgenerate_answer(state:MessageState):questionstate[messages][0].content contextstate[messages][-1].content promptGENERATE_PROMPT.format(questionquestion,contextcontext)return{messages:[model.invoke([HumanMessage(contentprompt)])]}该节点只基于检索到的上下文回答问题并要求答案简洁。这是一个经典 RAG 生成节点。六、Demo4使用 Overwrite 覆盖状态Demo4 演示了 LangGraph 中的 Overwrite。状态定义如下classState(TypedDict):messages:Annotated[list[str],operator.add]默认情况下messages 会通过 operator.add 追加defadd_message(state:State):return{messages:[first message]}但如果希望覆盖旧值可以使用fromlanggraph.typesimportOverwritedefreplace_message(state:State):return{messages:Overwrite([second message])}最终结果不是[“”, “first message”, “second message”]而是[“second message”]这说明 Overwrite 可以绕过 reducer 的默认合并逻辑强制覆盖字段。它适合以下场景清空历史消息重置中间状态用最终结果替换草稿结果避免列表无限增长七、Demo5input_schema 与 output_schemaDemo5 演示了 LangGraph 的输入输出 schema 隔离。代码中定义了三个状态classInputState(TypedDict):question:strclassOutputState(TypedDict):answer:strclassState(InputState,OutputState):pass构建图时指定builderStateGraph(State,input_schemaInputState,output_schemaOutputState)这样外部调用图时只需要传入 questionresultgraph.invoke({question:i am a question})最终返回结果只暴露answer{answer:Answer to : i am a question}这个能力适合在工程中封装复杂工作流。内部状态可以很复杂但对外只暴露必要字段接口更清晰。八、Demo6节点之间传递私有状态Demo6 演示了一个很实用的技巧节点之间可以传递不暴露给最终输出的中间状态。代码定义了多个状态类型classState(TypedDict):result:strclassNode1OutputState(TypedDict):sensitive_data:strclassNode2InputState(TypedDict):sensitive_data:str第一个节点返回敏感数据defnode_1(state:State)-Node1OutputState:return{sensitive_data:我是敏感数据}第二个节点读取敏感数据并处理defnode_2(state:Node2InputState)-State:return{result:处理后的数据}第三个节点构造最终结果defnode_3(state:State):return{result:state[result]完成}然后通过builder.add_sequence([node_1, node_2, node_3])builder.add_edge(START, “node_1”)将三个节点串成顺序流程。这个 demo 可以用来理解节点输入类型可以不同。节点输出可以包含中间字段。中间字段可以只在图内部流转。对外最终输出可以保持简洁。在真实业务里这种模式适合鉴权流程敏感信息处理数据清洗链路多阶段任务编排九、图可视化test2Serach.py 和 test3RAG.py 都使用了图可视化img_bytes graph.get_graph(xrayTrue).draw_mermaid_png()with open(“graph.jpg”, “wb”) as f:f.write(img_bytes)这段代码会把 LangGraph 的执行流程画成图片。xrayTrue 可以展示更详细的图结构对于调试复杂 Agent 很有帮助。建议在开发 LangGraph Agent 时养成一个习惯每次增加节点和条件边后都生成一次图确认流程是不是自己预期的样子。因为 Agent 流程一旦包含工具调用、重写、评分、循环单靠读代码很容易看漏路径。十、开发中的注意点结合这些 demo有几个工程实践建议API Key 和数据库连接不要写死在代码里建议统一放到 .env。InMemoryVectorStore适合 demo不适合长期生产存储。工具调用可以手写节点也可以优先使用 ToolNode。判断是否调用工具时简单场景可以手写should_continue标准工具调用场景推荐tools_condition。长流程一定要画图尤其是包含条件边和循环时。状态字段要提前设计好哪些字段追加、哪些字段覆盖需要明确。对外接口建议使用input_schema和output_schema避免泄露内部中间状态。RAG 流程中最好加入文档相关性判断否则容易基于无关上下文生成答案。问题重写可以显著提升召回质量但要避免无限循环生产环境建议增加最大重试次数。十一、总结demo1 用物流配送解释了 LangGraph 的基本图结构状态、节点、边、条件边。demo2 把普通状态图升级成了带搜索工具的 Agent展示了 LLM 如何通过 tool_calls 驱动工具节点。demo3 进一步实现了完整 RAG 流程包括文档加载、切分、向量检索、检索工具、文档评分、问题重写和最终回答。demo4,demo5,demo6 则补充了工程中非常重要的状态控制能力覆盖状态、输入输出隔离、节点间私有状态传递。学完这几个 demo 后可以对 LangGraph 形成一个清晰认识LangGraph 不只是用来写 Agent 的库它更像是一个面向 LLM 应用的有状态工作流引擎。当任务流程变得复杂比如需要工具调用、检索、判断、重试、分支、循环、记忆和人工干预时LangGraph 的图结构会比普通链式调用更清晰、更可控也更适合工程化落地。参考资料LangChain官方文档代码实践仓库