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 llm=ChatOllama(model="qwen2.5:7b",temperature=0.2)response=llm.invoke("请解释什么是LangChain")print(response.content)

ChatOllama返回的是一个消息对象，而不是普通字符串。生成的文本通常位于：

response.content

3.1 使用消息角色

fromlangchain_core.messagesimportSystemMessage,HumanMessage messages=[SystemMessage(content="你是一名资深软件架构师。"),HumanMessage(content="如何设计一个高可用的消息队列系统？")]response=llm.invoke(messages)print(response.content)

消息角色包括：

• SystemMessage：定义模型角色和行为；
• HumanMessage：用户输入；
• AIMessage：模型输出；
• ToolMessage：工具执行结果。

通过消息类型组织上下文，比手动拼接字符串更加清晰。

四、使用 Prompt 模板

将 Prompt 直接写在业务代码里，会导致后续维护困难。LangChain 提供了模板机制。

fromlangchain_core.promptsimportChatPromptTemplate prompt=ChatPromptTemplate.from_messages([("system","你是一名专业技术作者，请使用清晰、准确的语言回答问题。"),("human","请围绕以下主题写一段技术说明：{topic}")])chain=prompt|llm result=chain.invoke({"topic":"向量数据库的索引原理"})print(result.content)

这里的|表示 Runnable 组合。执行过程如下：

输入变量 | Prompt模板 | ChatOllama | AIMessage

这比手动执行多个函数更加符合声明式编程思想。

五、结构化输出

大模型输出自然语言，程序很难直接解析。实际业务中通常需要模型返回固定结构。

定义 Pydantic 模型：

frompydanticimportBaseModel,FieldfromtypingimportLiteralclassIntentResult(BaseModel):intent:Literal["knowledge","weather","general"]confidence:float=Field(ge=0,le=1,description="意图识别置信度")reason:str

让模型输出结构化结果：

structured_llm=llm.with_structured_output(IntentResult)result=structured_llm.invoke("用户问：今天北京天气怎么样？")print(result)print(result.intent)print(result.confidence)

结构化输出的意义在于把模型输出转换为程序可使用的数据对象，后续可以直接根据intent选择不同处理流程。

需要注意，结构化输出并不意味着模型绝对可靠。生产环境仍然需要：

• 对字段进行校验；
• 处理解析失败；
• 设置默认值；
• 限制枚举范围；
• 记录异常响应。

六、实现一个简单的 RAG Chain

6.1 创建向量库

fromlangchain_ollamaimportOllamaEmbeddingsfromlangchain_chromaimportChroma embeddings=OllamaEmbeddings(model="nomic-embed-text")documents=["系统发布需要经过开发、测试、预发布和生产审批。","员工报销需要在费用发生后30天内提交。","数据库变更必须先执行备份，再执行变更脚本。"]vectorstore=Chroma.from_texts(texts=documents,embedding=embeddings,collection_name="company_knowledge",persist_directory="./chroma_data")retriever=vectorstore.as_retriever(search_kwargs={"k":3})

6.2 构建检索问答链

fromlangchain_core.promptsimportChatPromptTemplate rag_prompt=ChatPromptTemplate.from_template(""" 你是企业知识库助手。 请根据上下文回答问题。 如果上下文中没有答案，请明确说明无法从知识库中找到依据。 不要根据常识补充未提供的信息。 上下文： {context} 问题： {question} """)defformat_documents(docs):return"\n\n".join(doc.page_contentfordocindocs)defrag_answer(question:str):docs=retriever.invoke(question)context=format_documents(docs)chain=rag_prompt|llm response=chain.invoke({"context":context,"question":question})return{"answer":response.content,"sources":[doc.page_contentfordocindocs]}result=rag_answer("数据库变更前需要做什么？")print(result["answer"])

一个可靠的 RAG 系统不应该只返回答案，还应返回来源信息。这样用户可以判断答案是否有依据。

七、为什么复杂任务需要 LangGraph

假设我们要构建一个智能助手，它需要完成以下流程：

1. 判断用户意图；
2. 如果是知识库问题，执行知识检索；
3. 如果是天气问题，调用天气工具；
4. 如果是普通问题，直接让模型回答；
5. 检查答案是否存在依据；
6. 必要时重新检索；
7. 最后输出结果。

如果使用普通函数，代码可能变成大量嵌套判断：

ifintent=="knowledge":docs=retrieve()ifnotdocs:...elifintent=="weather":...else:...

LangGraph 将这些步骤拆分成节点，并通过边描述节点之间的跳转关系。这样流程更加清晰，也方便增加重试、人工审批和持久化。

八、定义 LangGraph 状态

状态是图中所有节点共享的数据结构。

fromtypingimportTypedDict,Listfromlangchain_core.messagesimportBaseMessageclassAgentState(TypedDict,total=False):question:strintent:strconfidence:floatdocuments:List[str]answer:strgrounded:boolmessages:List[BaseMessage]

total=False表示字段可以不是一开始就全部存在。不同节点可以逐步补充状态。

九、实现意图识别节点

frompydanticimportBaseModelfromtypingimportLiteralclassIntent(BaseModel):name:Literal["knowledge","weather","general"]confidence:floatintent_llm=llm.with_structured_output(Intent)defclassify_intent(state:AgentState):result=intent_llm.invoke(f"请判断下面问题属于哪一类：知识库、天气或普通问题。\n"f"问题：{state['question']}")return{"intent":result.name,"confidence":result.confidence}

节点函数接收状态，返回需要更新的字段。它不需要手动传递所有变量。

十、实现知识检索节点

defretrieve_documents(state:AgentState):docs=retriever.invoke(state["question"])return{"documents":[doc.page_contentfordocindocs]}

为了避免将过多信息传给模型，可以限制内容大小：

defretrieve_documents(state:AgentState):docs=retriever.invoke(state["question"])contents=[]total_length=0fordocindocs:text=doc.page_contentiftotal_length+len(text)>5000:breakcontents.append(text)total_length+=len(text)return{"documents":contents}

这是一个很重要的工程细节。检索结果越多不一定越好。过长的上下文可能导致：

• 推理速度变慢；
• 关键内容被淹没；
• Token 成本增加；
• 模型回答出现偏离。

十一、实现答案生成节点

answer_prompt=ChatPromptTemplate.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_llm=llm.with_structured_output(GroundingResult)defcheck_grounding(state:AgentState):context="\n\n".join(state.get("documents",[]))prompt=f""" 请判断下面的答案是否能够从上下文中得到支持。 上下文：{context}问题：{state["question"]}答案：{state["answer"]}请判断答案是否有充分依据。 """result=grounding_llm.invoke(prompt)return{"grounded":result.grounded}

这种“生成 + 校验”的方式可以降低明显幻觉，但它并不能从根本上保证事实正确。校验模型本身也可能判断错误，因此关键业务仍然需要人工审核或规则系统兜底。

十三、组装 LangGraph

fromlanggraph.graphimportStateGraph,START,END builder=StateGraph(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):intent=state.get("intent")ifintent=="knowledge":return"retrieve"ifintent=="weather":return"weather"return"generate_general"

为了简化示例，天气和普通问题节点如下：

defweather_node(state:AgentState):return{"answer":"天气工具尚未配置，请先接入实际天气API。"}defgeneral_node(state:AgentState):response=llm.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"):return"finish"return"retry"

增加重试节点：

defretry_retrieve(state:AgentState):docs=retriever.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")

编译图：

graph=builder.compile()

执行：

result=graph.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_mode="updates"):print(event)

输出可能类似：

{'classify':{'intent':'knowledge','confidence':0.96}}{'retrieve':{'documents':['系统发布需要经过开发、测试、预发布和生产审批。']}}{'generate':{'answer':'系统发布需要经过开发、测试、预发布和生产审批。'}}{'check':{'grounded':True}}

流式事件对于调试非常重要。传统的黑盒调用只能看到最终答案，而图流式输出可以观察每个节点的状态变化。

十五、接入工具调用

智能 Agent 通常需要调用外部工具。下面定义一个简单的天气工具：

fromlangchain_core.toolsimporttool@tooldefget_weather(city:str)->str:"""查询指定城市的天气。"""weather_data={"北京":"晴，25摄氏度","上海":"多云，28摄氏度","广州":"小雨，30摄氏度"}returnweather_data.get(city,"暂时没有该城市的天气数据")

模型绑定工具：

tool_llm=llm.bind_tools([get_weather])response=tool_llm.invoke("请查询北京的天气")print(response.tool_calls)

需要注意，模型返回工具调用请求并不代表工具已经执行。应用程序需要解析tool_calls，执行对应函数，再把结果返回给模型。

一个简单的执行逻辑：

defexecute_tools(message):outputs=[]forcallinmessage.tool_calls:ifcall["name"]=="get_weather":result=get_weather.invoke(call["args"])outputs.append(result)returnoutputs

在实际项目中，应该建立工具注册表，而不是使用大量if else：

TOOLS={"get_weather":get_weather}defexecute_tools(message):results=[]forcallinmessage.tool_calls:tool=TOOLS.get(call["name"])iftoolisNone:raiseValueError(f"未知工具：{call['name']}")result=tool.invoke(call["args"])results.append(result)returnresults

十六、LangChain 与 LangGraph 的职责划分

可以使用下面的方式理解二者关系：

如果流程是固定的，例如：

输入 -> Prompt -> 模型 -> 解析

使用 LangChain Chain 就足够。

如果流程包含条件、循环、重试、人工审批和多个 Agent，则应当使用 LangGraph。

十七、生产环境设计建议

17.1 节点必须尽量保持单一职责

一个节点最好只完成一件事。例如：

• 意图识别节点只负责分类；
• 检索节点只负责检索；
• 生成节点只负责生成；
• 校验节点只负责判断。

这样更容易测试，也容易替换。

17.2 所有外部调用都要设置超时

工具调用、向量数据库调用和模型调用都可能失败。不能让一个节点无限等待。

17.3 对状态进行裁剪

状态会随着流程不断增长。如果把完整对话、所有检索文档和所有工具结果都放进去，长流程可能导致上下文过大。

17.4 使用持久化保存状态

复杂任务可能运行很长时间，或者需要人工确认。此时应该使用 Checkpointer 保存状态，以便中断后继续。

17.5 不要让 Agent 拥有无限权限

工具应该遵循最小权限原则。例如：

• 只读数据库工具与写入工具分开；
• 高风险操作必须人工审批；
• 限制工具参数范围；
• 记录每次工具调用；
• 对敏感数据进行脱敏。

十八、测试一个 LangGraph 应用

不要只测试最终答案，还应该测试每个节点。

deftest_intent():result=classify_intent({"question":"数据库变更前需要做什么？"})assertresult["intent"]in{"knowledge","weather","general"}

测试检索节点：

deftest_retrieval():result=retrieve_documents({"question":"系统发布流程是什么？"})assert"documents"inresultassertlen(result["documents"])>0

测试图的完整执行：

deftest_graph():result=graph.invoke({"question":"数据库变更前需要执行什么操作？"})assertresult.get("answer")

对于大模型应用，测试重点不只是字符串是否完全相等，还包括：

• 是否返回结构化数据；
• 是否调用正确工具；
• 是否包含来源；
• 是否在资料不足时拒答；
• 是否能够从失败节点恢复；
• 是否不会泄露敏感信息。

十九、总结

LangChain 解决的是大模型应用开发中的组件复用问题，LangGraph 解决的是复杂任务中的流程编排问题。

一个比较完整的企业级智能应用通常可以拆分为：

用户输入 | 意图识别 | 流程路由 | 检索 / 工具 / 普通问答 | 结果生成 | 事实校验 | 人工审批或最终输出

LangChain 适合封装模型、Prompt、工具和检索器；LangGraph 适合表达条件分支、循环、重试和状态管理。

在实践中，不建议一开始就把所有逻辑都设计成复杂 Agent。更合理的路径是：

1. 先使用 LangChain 实现最小可用链路；
2. 再把固定流程拆成独立节点；
3. 当出现条件分支和循环时引入 LangGraph；
4. 最后补充状态持久化、监控、人工审批和安全控制。

真正可靠的大模型应用，不是简单地把模型接入系统，而是围绕模型建立一套可观测、可恢复、可验证的工程流程。