一文读懂LangChain/LangGraph

引言：为什么需要LangChain和LangGraph？

在人工智能快速发展的今天，大语言模型（LLM）已成为开发者构建智能应用的核心工具。然而，直接将LLM接入应用会遇到诸多挑战：如何管理复杂的对话流程？如何处理多步骤任务？如何集成外部工具和数据源？LangChain和LangGraph正是为解决这些问题而生的框架。

LangChain是一个用于开发由语言模型驱动的应用程序的框架，而LangGraph则是构建在LangChain之上的状态机库，专门用于创建复杂、有状态的对话和工作流。本文将带你全面了解这两个框架的核心概念、使用场景和实际应用。

什么是LangChain？

核心概念

LangChain是一个开源框架，旨在简化基于大语言模型的应用程序开发。它提供了一套标准化的接口和组件，让开发者能够更轻松地构建复杂的LLM应用。

主要组件包括：

1. 模型（Models）：支持多种LLM提供商（OpenAI、Anthropic、Google等）
2. 提示词（Prompts）：模板化和管理提示词
3. 链（Chains）：将多个LLM调用组合成序列
4. 记忆（Memory）：在对话中保持状态
5. 索引（Indexes）：与外部数据源集成
6. 代理（Agents）：让LLM决定使用哪些工具

基本使用示例

fromlangchain.llmsimportOpenAIfromlangchain.promptsimportPromptTemplatefromlangchain.chainsimportLLMChain# 初始化LLMllm=OpenAI(temperature=0.7)# 创建提示词模板prompt=PromptTemplate(input_variables=["product"],template="为{product}写一段营销文案，突出其三个主要特点。")# 创建链chain=LLMChain(llm=llm,prompt=prompt)# 运行链result=chain.run("智能手表")print(result)

什么是LangGraph？

从LangChain到LangGraph

虽然LangChain的链（Chains）可以处理简单的线性流程，但在面对复杂、有分支、有循环的工作流时显得力不从心。LangGraph应运而生，它引入了图（Graph）的概念，让开发者能够构建任意复杂度的状态机。

LangGraph的核心优势：

• 支持循环和条件分支
• 明确的状态管理
• 可视化工作流
• 更好的错误处理和重试机制

状态机概念

在LangGraph中，每个节点代表一个处理步骤，边代表状态转移。这种设计模式特别适合：

• 多轮对话系统
• 复杂决策流程
• 需要外部工具调用的任务
• 需要人工干预的工作流

LangGraph的核心组件

1. 状态（State）

状态是一个字典，在整个图执行过程中传递和更新。你可以定义状态的类型注解来获得更好的类型检查。

fromtypingimportTypedDict,Annotatedfromtyping_extensionsimportTypedDictimportoperatorclassState(TypedDict):messages:Annotated[list,operator.add]# 消息列表topic:str# 当前话题step:int# 当前步骤

2. 节点（Nodes）

节点是执行具体任务的函数，接收状态作为输入，返回更新后的状态。

defgenerate_response(state:State)->State:"""生成AI回复的节点"""# 获取最新用户消息last_message=state["messages"][-1]# 调用LLM生成回复response=llm.invoke(f"用户说：{last_message}\n请回复：")# 更新状态state["messages"].append({"role":"assistant","content":response})state["step"]+=1returnstate

3. 边（Edges）

边定义了节点之间的流转条件，可以是固定的，也可以是基于状态的动态条件。

defshould_continue(state:State)->str:"""决定下一步走向哪个节点"""ifstate["step"]>=5:return"end"# 结束对话elif"帮助"instate["messages"][-1]["content"]:return"help_node"# 转到帮助节点else:return"generate_response"# 继续生成回复

4. 图（Graph）

将节点和边组合成完整的图结构。

fromlanggraph.graphimportStateGraph,END# 创建图构建器workflow=StateGraph(State)# 添加节点workflow.add_node("generate_response",generate_response)workflow.add_node("help_node",provide_help)workflow.add_node("human_input",get_human_input)# 设置入口点workflow.set_entry_point("generate_response")# 添加条件边workflow.add_conditional_edges("generate_response",should_continue,{"end":END,"help_node":"help_node","generate_response":"generate_response"})# 添加固定边workflow.add_edge("help_node","generate_response")# 编译图app=workflow.compile()

实际应用场景

场景一：智能客服系统

fromlanggraph.graphimportStateGraph,ENDfromtypingimportTypedDict,AnnotatedimportoperatorclassChatState(TypedDict):messages:Annotated[list,operator.add]user_query:strcontext:listrequires_human:booldefclassify_intent(state:ChatState):"""分类用户意图"""# 使用LLM判断用户意图intent=llm.classify(state["user_query"])state["intent"]=intentreturnstatedefretrieve_knowledge(state:ChatState):"""检索相关知识"""ifstate["intent"]=="technical":# 从知识库检索docs=vectorstore.similarity_search(state["user_query"])state["context"]=docsreturnstatedefgenerate_answer(state:ChatState):"""生成回答"""response=llm.generate_with_context(query=state["user_query"],context=state["context"])state["messages"].append({"role":"assistant","content":response})returnstatedefescalate_to_human(state:ChatState):"""转接人工"""state["requires_human"]=Truereturnstate# 构建客服图workflow=StateGraph(ChatState)workflow.add_node("classify",classify_intent)workflow.add_node("retrieve",retrieve_knowledge)workflow.add_node("answer",generate_answer)workflow.add_node("escalate",escalate_to_human)workflow.set_entry_point("classify")workflow.add_edge("classify","retrieve")workflow.add_edge("retrieve","answer")workflow.add_edge("answer",END)# 条件边：如果用户不满意，转人工workflow.add_conditional_edges("answer",lambdas:"escalate"ifs.get("dissatisfied",False)elseEND,{"escalate":"escalate",END:END})

场景二：数据分析工作流

最佳实践

1. 状态设计原则

• 保持状态简洁：只存储必要的信息
• 使用类型注解：提高代码可读性和可维护性
• 考虑状态持久化：对于长时间运行的工作流

2. 错误处理

defsafe_node_function(state:State):"""带错误处理的节点函数"""try:# 主要逻辑result=some_operation(state)state["result"]=result state["error"]=NoneexceptExceptionase:# 错误处理state["error"]=str(e)state["should_retry"]=Truereturnstatedefretry_logic(state:State):"""重试逻辑"""ifstate.get("should_retry",False)andstate.get("retry_count",0)<3:state["retry_count"]=state.get("retry_count",0)+1return"previous_node"# 返回上一个节点重试else:return"error_handler"# 转到错误处理节点

3. 性能优化

• 并行执行：对于独立的节点，考虑并行处理
• 缓存机制：缓存昂贵的LLM调用结果
• 流式输出：对于长文本生成，使用流式响应

常见问题解答

Q1: LangChain和LangGraph有什么区别？

LangChain更像是一个工具箱，提供了构建LLM应用所需的各种组件。LangGraph则是一个专门用于构建复杂工作流的框架，特别擅长处理有状态、多步骤、有分支的任务。

Q2: 什么时候应该使用LangGraph？

当你需要：

• 多轮对话系统
• 复杂决策流程
• 需要循环或条件分支的任务
• 可视化工作流
• 更好的状态管理

Q3: 学习曲线如何？

如果你已经熟悉Python和基本的LLM概念，LangChain的学习曲线相对平缓。LangGraph需要理解状态机和图的概念，但官方文档和示例非常丰富。

Q4: 生产环境注意事项

• 做好错误处理和重试机制
• 监控和日志记录
• 状态持久化考虑
• 性能测试和优化