一、前言

LangGraph 是 LangChain 生态中用于构建有状态 Agent 工作流的框架。和普通链式调用不同，LangGraph 更强调“图”的概念：我们可以把一个复杂任务拆成多个节点，然后通过边、条件边、状态更新规则，把这些节点组织成一个可控、可追踪、可扩展的执行流程。
本文基于 LangChain官方文档 做出几个demo 代码进行知识点的相关总结与实践，主要涉及以下内容：
StateGraph 的基础用法
状态字段与 reducer 合并规则
条件路由与分支流程
LLM 工具调用 Agent
RAG 检索增强生成流程
ToolNode 与 tools_condition
Overwrite 覆盖状态
输入输出 schema 隔离
节点间私有状态传递
这些 demo 虽然代码量不大，但基本覆盖了 LangGraph 入门到 Agent 编排的核心知识。

二、LangGraph 的核心思想

LangGraph 的核心可以概括为一句话：
用图来组织任务流程，用状态来承载上下文，用节点来处理业务逻辑，用边来控制执行方向。

一个最基本的 LangGraph 程序通常包含四步：

fromlanggraph.graphimportStateGraphfromlanggraph.constantsimportSTART,END builder=StateGraph(State)builder.add_node("node",node_func)builder.add_edge(START,"node")builder.add_edge("node",END)graph=builder.compile()

其中：
StateGraph(State)：定义图的状态结构。
add_node：添加处理节点。
add_edge：添加固定执行边。
add_conditional_edges：添加条件路由边。
compile()：编译成可执行图。
invoke() / stream()：执行图。
LangGraph 并不是简单地把函数串起来，而是每个节点都接收当前状态，并返回状态更新。

三、Demo1：物流配送状态机

Demo1用一个“包裹配送流程”演示了 LangGraph 最基础也最重要的能力：状态流转和条件分支。

1. 定义状态
   代码中定义了一个 PackageState：

classPackageState(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):origin=state["origin"]return{"status":"已揽收","history":[f"在{origin}进行揽收"]}

节点只需要关心“读什么状态，返回什么更新”，不需要关心整个流程怎么走。
3. 条件路由
物流流程里最关键的是配送方式选择：

defselect_delivery(state:PackageState):priority=state["priority"]ifpriority=="加急":return"备注加急"else:return"无备注"

然后用add_conditional_edges绑定路由结果：

delivery.add_conditional_edges("分拣中心",select_delivery,{"备注加急":"加急配送","无备注":"标准配送"})

这就是 LangGraph 中非常典型的条件分支模式：
分拣中心 -> 判断 priority -> 加急配送 / 标准配送
这个 demo 适合用来理解 LangGraph 的基本执行模型。

四、Demo2：带搜索工具的 Agent

Demo2 演示了一个带 Tavily 搜索工具的 Agent。
整体流程如下：
START
↓
llm_call
↓
是否需要工具调用？
├─ 是：tool_node -> llm_call
└─ 否：END

1. 定义消息状态

classMessageState(TypedDict):messages:Annotated[List[AnyMessage],operator.add]llm_calls:Annotated[int,operator.add]

这里 messages 保存对话消息，llm_calls 记录模型调用次数。
operator.add 表示每个节点返回的新消息会追加到历史消息中。
2. 绑定工具
代码使用 Tavily 搜索：

search=TavilySearch(max_results=4,api_key=os.environ.get("TAVILY_API_KEY"))tools=[search]model_with_tools=model.bind_tools(tools)

通过 bind_tools 后，模型就可以在需要的时候返回工具调用请求。
3. LLM 节点

defllm_call(state:MessageState):messages=state["messages"]result=model_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:tool=tools_by_name[tool_call["name"]]res=tool.invoke(tool_call["args"])result.append(ToolMessage(content=res,tool_call_id=tool_call["id"]))return{"messages":result}

工具节点负责：
读取上一条 AI 消息中的 tool_calls。
找到对应工具。
执行工具。
把工具结果包装成 ToolMessage。
5. 条件判断

defshould_continue(state:MessageState):last_msg=state["messages"][-1]ifhasattr(last_msg,"tool_calls")andlast_msg.tool_calls:return"tool_node"returnEND

如果模型请求了工具调用，就进入 tool_node；否则流程结束。
这就是最基础的 ReAct Agent 图结构。

五、Demo3：RAG 检索增强生成

Demo3 实现了一个基于本地 Markdown 文档的 RAG Agent。
它的流程可以总结为：
用户问题
↓
模型判断是否需要检索
↓
调用 retriever_tool
↓
判断检索文档是否相关
├─ 相关：生成答案
└─ 不相关：重写问题，再次检索

1. 文档加载
   代码加载了本地 Markdown 文档：

paths=["../LangGraph/docs/markdown/Linux.md"]docs=[UnstructuredMarkdownLoader(path).load()forpathinpaths]docs_list=[itemforsublistindocsforiteminsublist]

然后使用文本分割器切块：

text_splitter=RecursiveCharacterTextSplitter.from_tiktoken_encoder(encoding_name="cl100k_base",chunk_size=1000,chunk_overlap=50)doc_splits=text_splitter.split_documents(docs_list)

这里的参数含义：
chunk_size=1000：每个文档块约 1000 token。
chunk_overlap=50：相邻文档块保留 50 token 重叠，减少语义断裂。
2. 构建向量库

embeddings=ZhipuAIEmbeddings(model="embedding-3",api_key=os.getenv("ZHIPU_API_KEY"))vectorstore=InMemoryVectorStore.from_documents(documents=doc_splits,embedding=embeddings)retriever=vectorstore.as_retriever(search_kwargs={"k":2})

这里使用的是内存向量库，适合 demo 和本地测试。
生产环境中可以替换成 Milvus、Chroma、pgvector、Elasticsearch 等持久化向量数据库。
3. 创建检索工具

retriever_tool=create_retriever_tool(retriever,"retrieve_bit","搜索并返回有关Linux的信息。",)

创建工具后，模型就可以通过工具调用的方式执行文档检索。
4. 生成查询或直接回答

defgenerate_query_or_respond(state:MessageState):ai_message=model.bind_tools([retriever_tool]).invoke(state["messages"])return{"messages":[ai_message]}

这个节点让模型自己判断：
如果能直接回答，就直接返回答案。
如果需要查资料，就调用 retrieve_bit 工具。
5. 使用 ToolNode 执行工具

retriever_node=ToolNode([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:str=Field(description="相关性评分：相关为yes,不相关为no")

评分节点：

defgrade_documents(state:MessageState)->Literal["generate_answer","rewrite_question"]:question=state["messages"][0].content context=state["messages"][-1].content prompt=GRADE_PROMPT.format(context=context,question=question)result=model.with_structured_output(GradeDocuments).invoke([HumanMessage(content=prompt)])ifresult.score=="yes":return"generate_answer"else:return"rewrite_question"

如果文档相关，就进入答案生成节点；如果不相关，就重写问题后重新检索。
8. 问题重写

defrewrite_question(state:MessageState):question=state["messages"][0].content prompt=REWRITE_PROMPT.format(question=question)result=model.invoke([HumanMessage(content=prompt)])return{"messages":[HumanMessage(content=result.content)]}

问题重写是 RAG 中非常实用的策略，尤其适合用户问题比较模糊、关键词不准确、语义不完整的情况。
9. 基于上下文生成答案

defgenerate_answer(state:MessageState):question=state["messages"][0].content context=state["messages"][-1].content prompt=GENERATE_PROMPT.format(question=question,context=context)return{"messages":[model.invoke([HumanMessage(content=prompt)])]}

该节点只基于检索到的上下文回答问题，并要求答案简洁。
这是一个经典 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 的默认合并逻辑，强制覆盖字段。
它适合以下场景：
清空历史消息
重置中间状态
用最终结果替换草稿结果
避免列表无限增长

七、Demo5：input_schema 与 output_schema

Demo5 演示了 LangGraph 的输入输出 schema 隔离。
代码中定义了三个状态：

classInputState(TypedDict):question:strclassOutputState(TypedDict):answer:strclassState(InputState,OutputState):pass

构建图时指定：

builder=StateGraph(State,input_schema=InputState,output_schema=OutputState)

这样外部调用图时只需要传入 question：

result=graph.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(xray=True).draw_mermaid_png()
with open(“graph.jpg”, “wb”) as f:
f.write(img_bytes)
这段代码会把 LangGraph 的执行流程画成图片。
xray=True 可以展示更详细的图结构，对于调试复杂 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官方文档
代码实践仓库