引言：为什么需要LangChain和LangGraph？
在人工智能快速发展的今天，大语言模型（LLM）已成为开发者构建智能应用的核心工具。然而，直接将LLM接入应用会遇到诸多挑战：如何管理复杂的对话流程？如何处理多步骤任务？如何集成外部工具和数据源？LangChain和LangGraph正是为解决这些问题而生的框架。
LangChain是一个用于开发由语言模型驱动的应用程序的框架，而LangGraph则是构建在LangChain之上的状态机库，专门用于创建复杂、有状态的对话和工作流。本文将带你全面了解这两个框架的核心概念、使用场景和实际应用。
什么是LangChain？
核心概念
LangChain是一个开源框架，旨在简化基于大语言模型的应用程序开发。它提供了一套标准化的接口和组件，让开发者能够更轻松地构建复杂的LLM应用。
主要组件包括：
- 模型（Models）：支持多种LLM提供商（OpenAI、Anthropic、Google等）
- 提示词（Prompts）：模板化和管理提示词
- 链（Chains）：将多个LLM调用组合成序列
- 记忆（Memory）：在对话中保持状态
- 索引（Indexes）：与外部数据源集成
- 代理（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"]+=1returnstate3. 边（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: 生产环境注意事项
- 做好错误处理和重试机制
- 监控和日志记录
- 状态持久化考虑
- 性能测试和优化