系列导读本系列共 6 篇带你从零到一构建完整的 RAG LangGraph MCP 项目。第 1 篇最小 RAG 实现纯 numpy无任何 AI 框架第 2 篇接入 Ollama 本地大模型实现真实语义检索第 3 篇接入 ChromaDB 持久化向量数据库第 4 篇用 LangChain 重构 多轮对话第 5 篇本文LangGraph 多步推理工作流第 6 篇MCP 工具调用协议集成一、第 4 篇的局限线性链条前 4 篇的 RAG 都是线性流程问题 → 检索 → 生成 → 回答但实际场景更复杂用户说你好还需要去检索知识库吗不需要检索到的内容相关性很低直接生成回答质量会很差怎么办重新检索问题太模糊直接检索效果不好应该先改写问题Query RewritingLangGraph 的思路把 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:# 读取 statequerystate[query]# 执行操作resultscollection.query(query_texts[query],n_results3)# 返回要更新的字段return{retrieved_docs:results,retrieval_score:0.85,}Edge边节点之间的连接分两种# 固定边A 执行完一定去 Bgraph.add_edge(retrieve,evaluate)# 条件边根据 state 决定走哪条路graph.add_conditional_edges(evaluate,lambdastate:generateifstate[retrieval_score]0.5elseexpand,{generate:generate,expand:expand})三、本文实现的工作流[用户输入] ↓ [analyze] 分析问题 ↙ ↘ 需要检索 不需要检索闲聊 ↓ ↓ [retrieve] [direct_answer] ↓ [evaluate] 评估检索质量 ↙ ↘ 质量好 质量差首次 ↓ ↓ [generate] [expand] 扩展检索 ↑ ↓ └─────────────────┘四、关键节点实现节点1分析问题含 Query Rewritingdefnode_analyze_question(state:RAGState)-dict: 两个功能 1. 判断是否需要检索闲聊不需要 2. 改写问题提升检索精度Query Rewriting questionstate[question]# 让模型判断是否需要检索judge_promptf判断以下问题是否需要查询企业知识库来回答。 闲聊/打招呼/常识问题回答 NO公司制度/流程/政策问题回答 YES。 只回答 YES 或 NO。 问题{question}judgmentollama_chat(judge_prompt).strip().upper()needs_retrievalYESinjudgmentifnotneeds_retrieval:return{needs_retrieval:False,query:question}# Query Rewriting把口语化问题改成检索关键词rewrite_promptf将用户问题改写为文档检索关键词去掉语气词保留核心词不超过20字。 原问题{question}搜索关键词rewrittenollama_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:scorestate[retrieval_score]threshold0.5ifscorethreshold: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 resultscollection.query(query_texts[state[question]],# 原始问题n_results5,# 扩大范围)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.8langgraph包要求 Python ≥ 3.9这里手写等价实现原理完全一致classStateGraph:def__init__(self,state_class):self.nodes{}self.edges{}self.conditional_edges{}self.entryNonedefset_entry_point(self,name):self.entrynamedefadd_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:statedict(initial_state)currentself.entry visited[]whilecurrentandcurrent!END:visited.append(current)updatesself.nodes[current](state)# 执行节点state.update(updates)# 更新 state# 决定下一个节点ifcurrentinself.conditional_edges:condition_fn,routingself.conditional_edges[current]resultcondition_fn(state)currentrouting.get(result,END)elifcurrentinself.edges:currentself.edges[current]else:currentENDprint(f执行路径{ → .join(visited)})returnstate六、组装工作流defbuild_workflow():# 条件路由函数defroute_after_analyze(state):returnretrieveifstate[needs_retrieval]elsedirect_answerdefroute_after_evaluate(state):# 质量好 或 已重试过 → 生成否则 → 扩展检索ifstate[retrieval_score]0.5orstate[retry_count]1:returngeneratereturnexpandgraphStateGraph(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# 构建方式完全一样graphStateGraph(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)# 编译手写版不需要这步appgraph.compile()# 调用resultapp.invoke(initial_state)两者的 API 几乎相同手写版帮你理解底层逻辑真实版提供更多高级功能并行节点、流式状态更新、持久化 checkpoint 等。总结LangGraph 的核心思想概念含义对应代码State贯穿所有节点的共享数据RAGStateTypedDictNode处理步骤纯函数node_xxx(state) → dictEdge节点连接add_edge/add_conditional_edges条件路由根据 state 决定走哪条路条件函数返回路由键本文实现了 6 个节点的工作流包含问题分类区分闲聊和知识库问题Query Rewriting改写问题提升检索精度质量评估评分低时触发重试防死循环通过retry_count限制重试次数下一篇加入 MCPModel Context Protocol工具调用让 Agent 能调用外部工具查询知识库、获取日期、计算工作日完成整个项目。