LangChain 与 LangGraph 实战：构建可编排的大模型智能应用

一、为什么需要 LangChain 和 LangGraph直接调用大模型 API 很容易但当应用变得复杂时开发者很快会遇到以下问题Prompt 散落在各个业务函数中多轮对话上下文难以管理工具调用逻辑与模型调用逻辑混杂RAG 检索流程难以复用任务失败后无法恢复多个 Agent 之间缺乏清晰的协作关系复杂流程只能通过大量if else维护。LangChain 主要解决的是“大模型应用组件化”问题。它提供了模型封装、Prompt 模板、输出解析器、工具调用、检索器和 Agent 等组件。LangGraph 则进一步解决“复杂流程编排”问题。它把大模型应用表示为一个有状态的图输入 | v 意图识别 | -- 知识库检索 | -- 工具调用 | v 答案生成 | v 输出LangChain 更像是组件库LangGraph 更像是工作流引擎。两者可以结合使用。二、安装依赖本文使用 Ollama 作为本地模型服务也可以替换为 OpenAI、通义千问或其他兼容模型。安装依赖pipinstall-Ulangchain langchain-ollama langgraph pydantic如果需要向量检索可以额外安装pipinstallchromadb langchain-chroma下载 Ollama 模型ollama pull qwen2.5:7b三、LangChain 调用本地模型最简单的模型调用代码如下fromlangchain_ollamaimportChatOllama llmChatOllama(modelqwen2.5:7b,temperature0.2)responsellm.invoke(请解释什么是LangChain)print(response.content)ChatOllama返回的是一个消息对象而不是普通字符串。生成的文本通常位于response.content3.1 使用消息角色fromlangchain_core.messagesimportSystemMessage,HumanMessage messages[SystemMessage(content你是一名资深软件架构师。),HumanMessage(content如何设计一个高可用的消息队列系统)]responsellm.invoke(messages)print(response.content)消息角色包括SystemMessage定义模型角色和行为HumanMessage用户输入AIMessage模型输出ToolMessage工具执行结果。通过消息类型组织上下文比手动拼接字符串更加清晰。四、使用 Prompt 模板将 Prompt 直接写在业务代码里会导致后续维护困难。LangChain 提供了模板机制。fromlangchain_core.promptsimportChatPromptTemplate promptChatPromptTemplate.from_messages([(system,你是一名专业技术作者请使用清晰、准确的语言回答问题。),(human,请围绕以下主题写一段技术说明{topic})])chainprompt|llm resultchain.invoke({topic:向量数据库的索引原理})print(result.content)这里的|表示 Runnable 组合。执行过程如下输入变量 | Prompt模板 | ChatOllama | AIMessage这比手动执行多个函数更加符合声明式编程思想。五、结构化输出大模型输出自然语言程序很难直接解析。实际业务中通常需要模型返回固定结构。定义 Pydantic 模型frompydanticimportBaseModel,FieldfromtypingimportLiteralclassIntentResult(BaseModel):intent:Literal[knowledge,weather,general]confidence:floatField(ge0,le1,description意图识别置信度)reason:str让模型输出结构化结果structured_llmllm.with_structured_output(IntentResult)resultstructured_llm.invoke(用户问今天北京天气怎么样)print(result)print(result.intent)print(result.confidence)结构化输出的意义在于把模型输出转换为程序可使用的数据对象后续可以直接根据intent选择不同处理流程。需要注意结构化输出并不意味着模型绝对可靠。生产环境仍然需要对字段进行校验处理解析失败设置默认值限制枚举范围记录异常响应。六、实现一个简单的 RAG Chain6.1 创建向量库fromlangchain_ollamaimportOllamaEmbeddingsfromlangchain_chromaimportChroma embeddingsOllamaEmbeddings(modelnomic-embed-text)documents[系统发布需要经过开发、测试、预发布和生产审批。,员工报销需要在费用发生后30天内提交。,数据库变更必须先执行备份再执行变更脚本。]vectorstoreChroma.from_texts(textsdocuments,embeddingembeddings,collection_namecompany_knowledge,persist_directory./chroma_data)retrievervectorstore.as_retriever(search_kwargs{k:3})6.2 构建检索问答链fromlangchain_core.promptsimportChatPromptTemplate rag_promptChatPromptTemplate.from_template( 你是企业知识库助手。 请根据上下文回答问题。 如果上下文中没有答案请明确说明无法从知识库中找到依据。 不要根据常识补充未提供的信息。 上下文 {context} 问题 {question} )defformat_documents(docs):return\n\n.join(doc.page_contentfordocindocs)defrag_answer(question:str):docsretriever.invoke(question)contextformat_documents(docs)chainrag_prompt|llm responsechain.invoke({context:context,question:question})return{answer:response.content,sources:[doc.page_contentfordocindocs]}resultrag_answer(数据库变更前需要做什么)print(result[answer])一个可靠的 RAG 系统不应该只返回答案还应返回来源信息。这样用户可以判断答案是否有依据。七、为什么复杂任务需要 LangGraph假设我们要构建一个智能助手它需要完成以下流程判断用户意图如果是知识库问题执行知识检索如果是天气问题调用天气工具如果是普通问题直接让模型回答检查答案是否存在依据必要时重新检索最后输出结果。如果使用普通函数代码可能变成大量嵌套判断ifintentknowledge:docsretrieve()ifnotdocs:...elifintentweather:...else:...LangGraph 将这些步骤拆分成节点并通过边描述节点之间的跳转关系。这样流程更加清晰也方便增加重试、人工审批和持久化。八、定义 LangGraph 状态状态是图中所有节点共享的数据结构。fromtypingimportTypedDict,Listfromlangchain_core.messagesimportBaseMessageclassAgentState(TypedDict,totalFalse):question:strintent:strconfidence:floatdocuments:List[str]answer:strgrounded:boolmessages:List[BaseMessage]totalFalse表示字段可以不是一开始就全部存在。不同节点可以逐步补充状态。九、实现意图识别节点frompydanticimportBaseModelfromtypingimportLiteralclassIntent(BaseModel):name:Literal[knowledge,weather,general]confidence:floatintent_llmllm.with_structured_output(Intent)defclassify_intent(state:AgentState):resultintent_llm.invoke(f请判断下面问题属于哪一类知识库、天气或普通问题。\nf问题{state[question]})return{intent:result.name,confidence:result.confidence}节点函数接收状态返回需要更新的字段。它不需要手动传递所有变量。十、实现知识检索节点defretrieve_documents(state:AgentState):docsretriever.invoke(state[question])return{documents:[doc.page_contentfordocindocs]}为了避免将过多信息传给模型可以限制内容大小defretrieve_documents(state:AgentState):docsretriever.invoke(state[question])contents[]total_length0fordocindocs:textdoc.page_contentiftotal_lengthlen(text)5000:breakcontents.append(text)total_lengthlen(text)return{documents:contents}这是一个很重要的工程细节。检索结果越多不一定越好。过长的上下文可能导致推理速度变慢关键内容被淹没Token 成本增加模型回答出现偏离。十一、实现答案生成节点answer_promptChatPromptTemplate.from_template( 你是一个严谨的企业智能助手。 请根据提供的上下文回答问题。 上下文 {context} 用户问题 {question} 要求 1. 不要编造上下文之外的事实 2. 如果资料不足明确说明 3. 使用简洁、专业的中文回答。 )defgenerate_answer(state:AgentState):context\n\n.join(state.get(documents,[]))response(answer_prompt|llm).invoke({context:context,question:state[question]})return{answer:response.content}十二、实现答案校验节点大模型生成答案后可以让另一个模型判断答案是否有上下文支持。classGroundingResult(BaseModel):grounded:boolreason:strgrounding_llmllm.with_structured_output(GroundingResult)defcheck_grounding(state:AgentState):context\n\n.join(state.get(documents,[]))promptf 请判断下面的答案是否能够从上下文中得到支持。 上下文{context}问题{state[question]}答案{state[answer]}请判断答案是否有充分依据。 resultgrounding_llm.invoke(prompt)return{grounded:result.grounded}这种“生成 校验”的方式可以降低明显幻觉但它并不能从根本上保证事实正确。校验模型本身也可能判断错误因此关键业务仍然需要人工审核或规则系统兜底。十三、组装 LangGraphfromlanggraph.graphimportStateGraph,START,END builderStateGraph(AgentState)builder.add_node(classify,classify_intent)builder.add_node(retrieve,retrieve_documents)builder.add_node(generate,generate_answer)builder.add_node(check,check_grounding)定义意图路由defroute_intent(state:AgentState):intentstate.get(intent)ifintentknowledge:returnretrieveifintentweather:returnweatherreturngenerate_general为了简化示例天气和普通问题节点如下defweather_node(state:AgentState):return{answer:天气工具尚未配置请先接入实际天气API。}defgeneral_node(state:AgentState):responsellm.invoke(state[question])return{answer:response.content}builder.add_node(weather,weather_node)builder.add_node(generate_general,general_node)添加边builder.add_edge(START,classify)builder.add_conditional_edges(classify,route_intent,{retrieve:retrieve,weather:weather,generate_general:generate_general})builder.add_edge(retrieve,generate)builder.add_edge(generate,check)builder.add_edge(weather,END)builder.add_edge(generate_general,END)定义校验后的路由defroute_grounding(state:AgentState):ifstate.get(grounded):returnfinishreturnretry增加重试节点defretry_retrieve(state:AgentState):docsretriever.invoke(state[question] 请检索更具体、更相关的资料)return{documents:[doc.page_contentfordocindocs]}builder.add_node(retry,retry_retrieve)builder.add_conditional_edges(check,route_grounding,{finish:END,retry:retry})builder.add_edge(retry,generate)编译图graphbuilder.compile()执行resultgraph.invoke({question:数据库变更前需要执行什么操作})print(result[answer])print(是否有依据,result.get(grounded))完整流程如下START | classify | -- weather ------ END | -- generate_general - END | retrieve | generate | check | -- grounded ------ END | retry | generate十四、使用流式输出观察执行过程LangGraph 支持流式获取节点执行结果foreventingraph.stream({question:系统发布需要经过哪些流程},stream_modeupdates):print(event)输出可能类似{classify:{intent:knowledge,confidence:0.96}}{retrieve:{documents:[系统发布需要经过开发、测试、预发布和生产审批。]}}{generate:{answer:系统发布需要经过开发、测试、预发布和生产审批。}}{check:{grounded:True}}流式事件对于调试非常重要。传统的黑盒调用只能看到最终答案而图流式输出可以观察每个节点的状态变化。十五、接入工具调用智能 Agent 通常需要调用外部工具。下面定义一个简单的天气工具fromlangchain_core.toolsimporttooltooldefget_weather(city:str)-str:查询指定城市的天气。weather_data{北京:晴25摄氏度,上海:多云28摄氏度,广州:小雨30摄氏度}returnweather_data.get(city,暂时没有该城市的天气数据)模型绑定工具tool_llmllm.bind_tools([get_weather])responsetool_llm.invoke(请查询北京的天气)print(response.tool_calls)需要注意模型返回工具调用请求并不代表工具已经执行。应用程序需要解析tool_calls执行对应函数再把结果返回给模型。一个简单的执行逻辑defexecute_tools(message):outputs[]forcallinmessage.tool_calls:ifcall[name]get_weather:resultget_weather.invoke(call[args])outputs.append(result)returnoutputs在实际项目中应该建立工具注册表而不是使用大量if elseTOOLS{get_weather:get_weather}defexecute_tools(message):results[]forcallinmessage.tool_calls:toolTOOLS.get(call[name])iftoolisNone:raiseValueError(f未知工具{call[name]})resulttool.invoke(call[args])results.append(result)returnresults十六、LangChain 与 LangGraph 的职责划分可以使用下面的方式理解二者关系组件主要职责Chat Model调用大语言模型Prompt Template管理输入模板Output Parser解析模型输出Retriever检索外部知识Tool封装外部能力Chain串联固定处理步骤LangGraph编排条件分支和循环流程Checkpointer保存流程状态Human-in-the-loop引入人工审批如果流程是固定的例如输入 - Prompt - 模型 - 解析使用 LangChain Chain 就足够。如果流程包含条件、循环、重试、人工审批和多个 Agent则应当使用 LangGraph。十七、生产环境设计建议17.1 节点必须尽量保持单一职责一个节点最好只完成一件事。例如意图识别节点只负责分类检索节点只负责检索生成节点只负责生成校验节点只负责判断。这样更容易测试也容易替换。17.2 所有外部调用都要设置超时工具调用、向量数据库调用和模型调用都可能失败。不能让一个节点无限等待。17.3 对状态进行裁剪状态会随着流程不断增长。如果把完整对话、所有检索文档和所有工具结果都放进去长流程可能导致上下文过大。17.4 使用持久化保存状态复杂任务可能运行很长时间或者需要人工确认。此时应该使用 Checkpointer 保存状态以便中断后继续。17.5 不要让 Agent 拥有无限权限工具应该遵循最小权限原则。例如只读数据库工具与写入工具分开高风险操作必须人工审批限制工具参数范围记录每次工具调用对敏感数据进行脱敏。十八、测试一个 LangGraph 应用不要只测试最终答案还应该测试每个节点。deftest_intent():resultclassify_intent({question:数据库变更前需要做什么})assertresult[intent]in{knowledge,weather,general}测试检索节点deftest_retrieval():resultretrieve_documents({question:系统发布流程是什么})assertdocumentsinresultassertlen(result[documents])0测试图的完整执行deftest_graph():resultgraph.invoke({question:数据库变更前需要执行什么操作})assertresult.get(answer)对于大模型应用测试重点不只是字符串是否完全相等还包括是否返回结构化数据是否调用正确工具是否包含来源是否在资料不足时拒答是否能够从失败节点恢复是否不会泄露敏感信息。十九、总结LangChain 解决的是大模型应用开发中的组件复用问题LangGraph 解决的是复杂任务中的流程编排问题。一个比较完整的企业级智能应用通常可以拆分为用户输入 | 意图识别 | 流程路由 | 检索 / 工具 / 普通问答 | 结果生成 | 事实校验 | 人工审批或最终输出LangChain 适合封装模型、Prompt、工具和检索器LangGraph 适合表达条件分支、循环、重试和状态管理。在实践中不建议一开始就把所有逻辑都设计成复杂 Agent。更合理的路径是先使用 LangChain 实现最小可用链路再把固定流程拆成独立节点当出现条件分支和循环时引入 LangGraph最后补充状态持久化、监控、人工审批和安全控制。真正可靠的大模型应用不是简单地把模型接入系统而是围绕模型建立一套可观测、可恢复、可验证的工程流程。