LangGraph 多步推理：State + Node + 条件路由，手写 StateGraph
系列导读：本系列共 6 篇，带你从零到一构建完整的 RAG + LangGraph + MCP 项目。
- 第 1 篇：最小 RAG 实现，纯 numpy，无任何 AI 框架
- 第 2 篇：接入 Ollama 本地大模型，实现真实语义检索
- 第 3 篇：接入 ChromaDB 持久化向量数据库
- 第 4 篇：用 LangChain 重构 + 多轮对话
- 第 5 篇（本文）：LangGraph 多步推理工作流
- 第 6 篇：MCP 工具调用协议集成
一、第 4 篇的局限：线性链条
前 4 篇的 RAG 都是线性流程：
问题 → 检索 → 生成 → 回答但实际场景更复杂：
- 用户说"你好"，还需要去检索知识库吗？（不需要）
- 检索到的内容相关性很低，直接生成回答质量会很差，怎么办？（重新检索）
- 问题太模糊，直接检索效果不好，应该先改写问题？（Query Rewriting）
LangGraph 的思路：把 Agent 的执行过程建模成有向图，支持条件分支、循环、回退。
二、LangGraph 核心概念
State（状态）
贯穿所有节点的共享数据，类似函数参数，每个节点读取并更新它：
fromtypingimportTypedDict,OptionalclassRAGState(TypedDict):question:str# 用户原始问题query:str# 实际检索用的查询（可能被改写）retrieved_docs:list# 检索到的文档retrieval_score:float# 检索质量评分answer:Optional[str]# 最终回答needs_retrieval:bool# 是否需要检索retry_count:int# 重试次数（防死循环）Node（节点）
每个节点是一个纯函数：接收当前 State，返回要更新的字段：
defnode_retrieve(state:RAGState)->dict:# 读取 statequery=state["query"]# 执行操作results=collection.query(query_texts=[query],n_results=3)# 返回要更新的字段return{"retrieved_docs":results,"retrieval_score":0.85,}Edge（边）
节点之间的连接，分两种：
# 固定边：A 执行完一定去 Bgraph.add_edge("retrieve","evaluate")# 条件边：根据 state 决定走哪条路graph.add_conditional_edges("evaluate",lambdastate:"generate"ifstate["retrieval_score"]>=0.5else"expand",{"generate":"generate","expand":"expand"})三、本文实现的工作流
[用户输入] ↓ [analyze] 分析问题 ↙ ↘ 需要检索 不需要检索（闲聊） ↓ ↓ [retrieve] [direct_answer] ↓ [evaluate] 评估检索质量 ↙ ↘ 质量好 质量差（首次） ↓ ↓ [generate] [expand] 扩展检索 ↑ ↓ └─────────────────┘四、关键节点实现
节点1：分析问题（含 Query Rewriting）
defnode_analyze_question(state:RAGState)->dict:""" 两个功能： 1. 判断是否需要检索（闲聊不需要） 2. 改写问题提升检索精度（Query Rewriting） """question=state["question"]# 让模型判断是否需要检索judge_prompt=f"""判断以下问题是否需要查询企业知识库来回答。 闲聊/打招呼/常识问题回答 NO，公司制度/流程/政策问题回答 YES。 只回答 YES 或 NO。 问题：{question}"""judgment=ollama_chat(judge_prompt).strip().upper()needs_retrieval="YES"injudgmentifnotneeds_retrieval:return{"needs_retrieval":False,"query":question}# Query Rewriting：把口语化问题改成检索关键词rewrite_prompt=f"""将用户问题改写为文档检索关键词，去掉语气词，保留核心词（不超过20字）。 原问题：{question}搜索关键词："""rewritten=ollama_chat(rewrite_prompt).strip()print(f" 原问题：{question}→ 改写后：{rewritten}")return{"needs_retrieval":True,"query":rewritten}Query Rewriting（查询改写）是 RAG 优化的重要技巧：
用户："帮我看看请假需要走什么流程啊" 改写后："请假流程 OA系统 申请步骤" ← 更接近文档中的表达方式节点3：评估检索质量
defnode_evaluate_retrieval(state:RAGState)->dict:score=state["retrieval_score"]threshold=0.5ifscore<threshold:print(f" 质量不足（{score:.3f}<{threshold}），需要扩展检索")else:print(f" 质量良好（{score:.3f}），继续生成")return{}# 不修改 state，路由决策在条件边里节点4：扩展检索（回退策略）
defnode_expand_retrieval(state:RAGState)->dict:""" 当检索质量不足时： 1. 用原始问题（不用改写后的）重新检索 2. 扩大 top_k 从 3 到 5 """results=collection.query(query_texts=[state["question"]],# 原始问题n_results=5,# 扩大范围)scores=[1-dfordinresults["distances"][0]]return{"retrieved_docs":list(zip(results["documents"][0],scores)),"retrieval_score":max(scores),"retry_count":state["retry_count"]+1,}五、手写 StateGraph（兼容 Python 3.8）
langgraph包要求 Python ≥ 3.9，这里手写等价实现，原理完全一致：
classStateGraph:def__init__(self,state_class):self.nodes={}self.edges={}self.conditional_edges={}self.entry=Nonedefset_entry_point(self,name):self.entry=namedefadd_node(self,name,fn):self.nodes[name]=fndefadd_edge(self,from_,to_):self.edges[from_]=to_defadd_conditional_edges(self,from_node,condition_fn,routing):"""condition_fn(state) 返回字符串，routing 映射到下一节点"""self.conditional_edges[from_node]=(condition_fn,routing)definvoke(self,initial_state)->dict:state=dict(initial_state)current=self.entry visited=[]whilecurrentandcurrent!="END":visited.append(current)updates=self.nodes[current](state)# 执行节点state.update(updates)# 更新 state# 决定下一个节点ifcurrentinself.conditional_edges:condition_fn,routing=self.conditional_edges[current]result=condition_fn(state)current=routing.get(result,"END")elifcurrentinself.edges:current=self.edges[current]else:current="END"print(f"执行路径：{' → '.join(visited)}")returnstate六、组装工作流
defbuild_workflow():# 条件路由函数defroute_after_analyze(state):return"retrieve"ifstate["needs_retrieval"]else"direct_answer"defroute_after_evaluate(state):# 质量好 或 已重试过 → 生成；否则 → 扩展检索ifstate["retrieval_score"]>=0.5orstate["retry_count"]>=1:return"generate"return"expand"graph=StateGraph(RAGState)# 注册节点graph.add_node("analyze",node_analyze_question)graph.add_node("retrieve",node_retrieve)graph.add_node("evaluate",node_evaluate_retrieval)graph.add_node("expand",node_expand_retrieval)graph.add_node("generate",node_generate_answer)graph.add_node("direct_answer",node_direct_answer)graph.set_entry_point("analyze")# 固定边graph.add_edge("retrieve","evaluate")graph.add_edge("expand","generate")graph.add_edge("generate","END")graph.add_edge("direct_answer","END")# 条件边graph.add_conditional_edges("analyze",route_after_analyze,{...})graph.add_conditional_edges("evaluate",route_after_evaluate,{...})returngraph七、运行效果
闲聊问题：
>>> 执行节点：analyze 判断：不需要检索（闲聊类问题） >>> 执行节点：direct_answer 🤖 你好！我是企业知识库助手... 执行路径：analyze → direct_answer知识库问题：
>>> 执行节点：analyze 原问题：我想请假怎么申请？ → 改写后：请假申请流程 OA系统 >>> 执行节点：retrieve 检索到 3 个片段，最高相似度：0.872 >>> 执行节点：evaluate 质量良好（0.872 >= 0.5），继续生成 >>> 执行节点：generate 🤖 根据HR手册，请假流程如下... 执行路径：analyze → retrieve → evaluate → generate检索质量差时自动重试：
>>> 执行节点：evaluate 质量不足（0.38 < 0.5），需要扩展检索 >>> 执行节点：expand 扩展检索到 5 个片段 >>> 执行节点：generate 执行路径：analyze → retrieve → evaluate → expand → generate八、与真实 LangGraph 的对比
使用真实langgraph包（Python ≥ 3.9）的等价写法：
fromlanggraph.graphimportStateGraph,END# 构建方式完全一样graph=StateGraph(RAGState)graph.add_node("analyze",node_analyze_question)graph.add_edge("retrieve","evaluate")graph.add_conditional_edges("analyze",route_after_analyze,{...})graph.set_entry_point("analyze")# 编译（手写版不需要这步）app=graph.compile()# 调用result=app.invoke(initial_state)两者的 API 几乎相同，手写版帮你理解底层逻辑，真实版提供更多高级功能（并行节点、流式状态更新、持久化 checkpoint 等）。
总结
LangGraph 的核心思想：
| 概念 | 含义 | 对应代码 |
|---|---|---|
| State | 贯穿所有节点的共享数据 | RAGStateTypedDict |
| Node | 处理步骤（纯函数） | node_xxx(state) → dict |
| Edge | 节点连接 | add_edge/add_conditional_edges |
| 条件路由 | 根据 state 决定走哪条路 | 条件函数返回路由键 |
本文实现了 6 个节点的工作流，包含：
- 问题分类：区分闲聊和知识库问题
- Query Rewriting：改写问题提升检索精度
- 质量评估：评分低时触发重试
- 防死循环：通过
retry_count限制重试次数
下一篇加入 MCP（Model Context Protocol）工具调用，让 Agent 能调用外部工具（查询知识库、获取日期、计算工作日），完成整个项目。