LangChain与LangGraph实战指南：从Agent到Graph的智能体开发

1. 从零到一：为什么我们需要LangChain和LangGraph？

如果你最近在捣鼓大语言模型（LLM），想把它们集成到自己的应用里，大概率会碰到一个词：LangChain。一开始你可能会想，我直接调用OpenAI的API不就行了吗？干嘛要引入一个框架？我自己也这么想过，直到我尝试构建一个稍微复杂点的应用，比如一个能联网搜索、读取本地文档、然后根据你的问题给出综合答案的聊天机器人。你会发现，事情很快就变得一团糟。

想象一下，你需要手动处理这些事：调用API、解析返回的JSON、把结果喂给下一个提示词、管理对话历史、处理可能出现的错误（比如网络超时或API限制）、还要考虑如何把长文档拆分成模型能处理的片段。这就像你打算盖个小木屋，结果发现自己得先从伐木和烧砖开始。LangChain的出现，就是为了解决这个“从伐木开始”的问题。它提供了一套标准化的“乐高积木”（官方叫Components），比如各种LLM的接口（OpenAI、Anthropic、本地模型）、提示词模板、文档加载器、文本分割器、向量数据库集成工具等等。你用这些积木来搭建应用，效率会高得多。

那么LangGraph又是什么？你可以把它看作是LangChain的“进阶版工作流引擎”。LangChain本身提供了链（Chains）和代理（Agents）来组合多个步骤，但当你的逻辑变得非常复杂，比如需要循环、条件分支、或者多个“执行者”（Agent）协同工作时，基础的链和代理用起来就有点力不从心了。LangGraph引入了“图”（Graph）的概念，让你能清晰地定义应用的状态（State）和节点（Nodes），以及节点之间流转的边（Edges）。它特别适合构建那种有复杂决策逻辑的智能体（Agent）。如果说LangChain帮你准备好了砖瓦和预制板，那么LangGraph就是给你提供了设计图纸和施工流程，让你能盖出结构更复杂、功能更强大的“智能大厦”。

这个开源教程项目（LangChain-OpenTutorial）的价值就在于此。它不是一个简单的API文档翻译，而是由一群一线开发者和实践者共同维护的实战指南。里面充满了他们在真实项目中踩过的坑、总结的最佳实践，以及对最新特性的深度解读。无论你是刚听说LangChain的新手，还是已经用它做过几个项目的老手，都能在这里找到对你有用的东西。

2. 核心概念拆解：Agent、Chain与Graph的实战定位

很多教程一上来就讲安装和“Hello World”，但我觉得，先搞清楚这几个核心概念到底能解决什么实际问题，比直接写代码更重要。这样你在设计自己的应用时，才能做出正确的选择。

2.1 Agent：你的“自动驾驶”执行者

Agent是LangChain里最吸引人也最容易让人困惑的概念。你可以把它理解为一个配备了“工具”（Tools）和“思考能力”的LLM。它的核心工作模式是“感知-思考-行动”循环。

1. 感知：接收用户的输入（比如“北京今天的天气怎么样？”）。
2. 思考：LLM内核会分析：“要回答这个问题，我需要知道北京的实时天气。我有没有能获取天气的工具？” 这个过程可能涉及查阅内部的提示词和可用工具列表。
3. 行动：LLM决定调用“天气查询工具”，并生成符合该工具输入格式的指令（如get_weather(“Beijing”)）。
4. 观察：工具执行，返回结果（“北京，晴，25摄氏度”）。
5. 再思考/再行动：LLM接收到结果，判断是否足以回答问题。如果足够，就组织语言回复给用户；如果不够（比如用户问“那明天呢？”），则可能开启新一轮的“思考-行动”。

实战心得：Agent的强大在于其自主性，但这也是调试的难点。你经常会遇到Agent陷入循环（反复调用同一个工具）、错误地选择工具、或者生成不符合工具输入格式的内容。在教程中，我们会重点分享如何通过设计更好的提示词、给工具添加清晰的描述、以及设置max_iterations（最大迭代次数）来约束Agent的行为。

2.2 Chain：确定性的工作流水线

如果说Agent是“自动驾驶”，那么Chain就是“预设导航路线”。Chain将一系列对LLM或其他工具的调用按固定顺序链接起来。它的流程是确定性的，没有基于结果的循环判断。

一个典型的Chain例子是“摘要生成链”：

1. 加载长文档。
2. 用文本分割器将其拆成小块。
3. 对每一小块进行摘要。
4. 将所有小块的摘要合并，再生成一个最终摘要。

为什么用Chain？因为简单、可靠、可预测。当你需要完成一个步骤清晰、不需要中途决策的任务时，Chain是最高效的选择。在LangChain-OpenTutorial里，你会看到大量关于如何构建复杂Chain的案例，比如结合检索器（Retriever）的问答链（RetrievalQA），它固定地先检索相关文档，再将文档和问题一起发给LLM生成答案。

2.3 LangGraph：为复杂Agent绘制蓝图

当你的Agent需要处理多轮对话、管理长期记忆、或者协调多个子Agent分工合作时，基础的Agent框架就显得有些笨拙。LangGraph应运而生。

它的核心是状态（State）和节点（Nodes）。

• 状态：一个共享的数据结构，在整个图执行过程中传递和修改。通常包含用户输入、对话历史、中间结果、工具执行结果等。
• 节点：图中的每个节点是一个函数，它读取状态，执行操作（如调用LLM、运行工具），并更新状态。
• 边：决定执行完一个节点后，下一个该执行哪个节点。边可以是有条件的，根据状态中的某个值来决定流向。

举个例子：构建一个客服Agent。

• 节点1（路由）：判断用户意图是“查询订单”还是“投诉”。
• 条件边：如果是“查询订单”，流向节点2；如果是“投诉”，流向节点3。
• 节点2（查询）：调用订单查询工具，将结果写入状态。
• 节点3（安抚）：调用LLM生成安抚性话语，并询问是否需要转接人工。
• 节点4（回复）：根据状态中的结果，组织最终回复给用户。

通过LangGraph，你可以清晰地可视化这个流程，并且非常方便地增加新的决策分支或处理步骤。教程中关于LangGraph的部分，会手把手教你如何定义状态模式（State Schema），编写节点函数，并设置条件边来实现复杂的业务逻辑。

3. 环境搭建与核心工具链选型

开始动手之前，搭一个好用的环境能事半功倍。这里我分享一套经过验证的、兼顾学习和生产需求的配置方案。

3.1 Python环境与包管理：隔离是关键

绝对不要在系统全局Python环境里直接pip install langchain！不同项目对库版本的依赖可能冲突。我强烈推荐使用conda或venv创建虚拟环境。

# 使用 conda (适合数据科学全家桶) conda create -n langchain-env python=3.10 conda activate langchain-env # 或者使用 venv (轻量) python -m venv langchain-env source langchain-env/bin/activate # Linux/Mac # langchain-env\Scripts\activate # Windows

包安装策略：LangChain生态的包很多，一次性全装没必要。建议按需安装。

# 核心库 pip install langchain langchain-community # 如果你要用OpenAI的模型 pip install openai # 如果你要用LangGraph pip install langgraph # 如果你需要用到本地文档处理（如PDF） pip install pypdf unstructured # 如果你需要用到向量数据库（如Chroma） pip install chromadb

教程仓库里的Jupyter Notebook（.ipynb）都设计为能在Google Colab上直接运行，这意味着它们通常会包含一个!pip install的单元格来安装所有依赖，这对初学者非常友好。但在本地开发时，我建议维护一个requirements.txt文件来精确控制版本。

3.2 模型选择：云端巨头 vs. 本地英雄

这是最重要的决策之一，直接关系到成本、速度和能力。

实操建议：

• 初学者和原型阶段：毫不犹豫地选择OpenAI的GPT-3.5-turbo。它成本低、速度快、API稳定，是学习LangChain各种功能的最佳搭档。你只需要一个OpenAI API Key。
• 深入学习和生产试探：可以尝试Anthropic的Claude-3或Google的Gemini，感受不同模型的风格。同时，用Ollama这类工具在本地跑一个7B参数的模型（如Llama 3），了解本地部署的流程和限制。
• 生产环境：需要综合评估效果、成本、响应延迟和数据合规要求。很多公司会采用混合策略：简单任务用低成本模型或本地模型，复杂任务用高性能云端模型。

在教程中，你会看到大量使用ChatOpenAI这个类的例子。这是LangChain为OpenAI模型提供的标准接口。如果你想换用其他模型，比如本地部署的，通常只需要换一个类，比如ChatOllama，而你的Chain或Agent的其余代码几乎不用变。这就是LangChain抽象层带来的好处。

3.3 向量数据库与记忆管理：让AI拥有“记忆”

LLM本身是“健忘”的，它不记得你之前说过什么。为了进行多轮对话，你需要把历史对话记录下来，并在每次提问时一并送给它。这就是“记忆”（Memory）组件。

简单记忆：ConversationBufferMemory，它会把所有历史对话都存下来。问题在于，对话长了之后，会消耗大量Token，可能超出模型上下文长度，且无关历史可能干扰当前问题。

优化记忆：ConversationSummaryMemory，它会让LLM定期对之前的对话进行摘要，只保存摘要，节省空间。ConversationBufferWindowMemory则只保留最近K轮对话。

对于更复杂的应用，比如让AI阅读你的个人文档并回答问题，就需要向量数据库。它的工作原理是：

1. 将文档分割成片段。
2. 用嵌入模型（Embedding Model）将每个片段转换为一个高维向量（可以理解为一串有意义的数字）。
3. 将这些向量存入向量数据库。
4. 当用户提问时，将问题也转换成向量。
5. 在向量数据库中搜索与问题向量最相似的几个文档片段（这个过程叫“检索”）。
6. 将这些相关片段作为上下文，和问题一起交给LLM，让它生成答案。

工具选型：ChromaDB轻量易上手，适合学习和中小项目；Milvus或Weaviate功能更强大，适合大规模生产环境。教程中多数例子使用ChromaDB，因为它无需额外服务，可以内存运行。

4. 实战演练：构建你的第一个智能体（Agent）

理论说了这么多，我们直接来撸代码。目标是构建一个能查询天气并能进行一般对话的智能体。

4.1 步骤一：定义工具

工具是Agent的手和脚。我们先定义一个模拟的天气查询工具。

from langchain.agents import tool from datetime import datetime @tool def get_weather(city: str) -> str: """根据城市名查询该城市的当前天气。""" # 这里模拟一个返回结果，真实情况应该调用天气API weather_data = { "Beijing": "晴朗，25摄氏度，微风。", "Shanghai": "多云，28摄氏度，湿度较高。", "New York": "小雨，18摄氏度，东北风3级。" } current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") return f"[{current_time}] {city}的天气是：{weather_data.get(city, '抱歉，未找到该城市天气信息。')}"

注意@tool装饰器和清晰的功能描述（docstring），这对Agent正确理解和使用工具至关重要。

4.2 步骤二：初始化Agent执行器

我们将使用OpenAI的模型和ReAct（Reasoning + Acting）框架来创建Agent。

from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain import hub # 1. 加载一个预设的ReAct提示词模板 prompt = hub.pull("hwchase17/react") # 2. 初始化LLM。请替换your_openai_api_key为你的真实密钥，或通过环境变量设置。 llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0, openai_api_key="your_openai_api_key") # 3. 定义工具列表 tools = [get_weather] # 4. 创建Agent agent = create_react_agent(llm, tools, prompt) # 5. 创建Agent执行器，并限制最大迭代次数防止死循环 agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=3, handle_parsing_errors=True)

关键参数解析：

• temperature=0：让输出更确定，减少随机性，适合工具调用场景。
• verbose=True：打印详细的执行过程，方便调试。
• max_iterations=3：安全阀，防止Agent陷入思考循环。
• handle_parsing_errors=True：当Agent输出的内容无法被解析为工具调用时，优雅地处理错误，而不是直接崩溃。

4.3 步骤三：运行与调试

现在，让我们来问它几个问题。

# 问题1：查询天气 result1 = agent_executor.invoke({"input": "北京今天天气怎么样？"}) print(result1["output"]) # 问题2：混合任务（它会先查天气，再结合常识回答） result2 = agent_executor.invoke({"input": "北京天气适合穿短袖吗？"}) print(result2["output"]) # 问题3：一般对话（不需要调用工具） result3 = agent_executor.invoke({"input": "你是谁？"}) print(result3["output"])

当verbose=True时，你会在控制台看到类似下面的思考过程：

> Entering new AgentExecutor chain... 我需要查询北京的天气来回答是否适合穿短袖。 Action: get_weather Action Input: {"city": "Beijing"} Observation: [2024-05-20 10:00:00] 北京的天气是：晴朗，25摄氏度，微风。 Thought: 用户问是否适合穿短袖。北京现在25度，晴朗，微风。这个温度通常比较舒适，适合穿短袖。 Final Answer: 根据当前天气（晴朗，25摄氏度），北京今天非常适合穿短袖。

这个过程清晰地展示了Agent的“思考-行动-观察”循环。通过观察这些日志，你可以判断Agent的决策是否合理，工具调用是否正确。

5. 进阶实战：用LangGraph构建一个决策型客服助手

现在我们来点更复杂的。假设我们要构建一个客服助手，它能根据用户意图自动路由：如果是简单问候，直接回复；如果是产品咨询，去查知识库；如果是投诉，则记录信息并提示转人工。

5.1 定义状态与节点

首先，我们定义整个图要维护的状态。

from typing import TypedDict, Annotated, Union from langgraph.graph.message import add_messages import operator class State(TypedDict): # 用户的最新输入 user_input: str # 完整的对话历史，LangGraph提供了专用类型来管理 messages: Annotated[list, add_messages] # 路由决策的结果 intent: str # 从知识库查到的信息 product_info: Union[str, None] # 是否需要转人工 need_human: bool

TypedDict让状态结构清晰。Annotated用于告诉LangGraph如何合并多个节点对messages字段的更新（这里用add_messages函数）。

然后，我们定义各个节点函数。每个函数都接收一个State字典，并返回一个包含更新后状态的字典。

# 节点A：意图识别 def intent_router_node(state: State): from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) # 构建一个提示词让LLM判断意图 system_prompt = """你是一个客服助手。请判断用户输入的意图类别，只返回以下三种之一： - greeting: 问候或闲聊 - inquiry: 产品咨询或功能询问 - complaint: 投诉或不满 用户输入：{user_input} """ prompt = system_prompt.format(user_input=state["user_input"]) response = llm.invoke(prompt) intent = response.content.strip().lower() return {"intent": intent} # 节点B：处理问候 def handle_greeting_node(state: State): friendly_response = "您好！我是客服助手，很高兴为您服务。请问有什么可以帮您？" return {"messages": [("assistant", friendly_response)]} # 节点C：处理产品咨询（模拟查询知识库） def handle_inquiry_node(state: State): # 这里模拟一个知识库查询 fake_knowledge_base = { "价格": "我们的基础版套餐每月99元。", "功能": "支持AI对话、文档分析和自动摘要。", "售后": "提供7x24小时在线技术支持。" } # 简单关键词匹配（实际应用应用嵌入和向量检索） info = "未找到相关信息。" for key in fake_knowledge_base: if key in state["user_input"]: info = fake_knowledge_base[key] break response = f"关于您咨询的【{state['user_input']}】，相关信息如下：{info}" return {"product_info": info, "messages": [("assistant", response)]} # 节点D：处理投诉 def handle_complaint_node(state: State): response = "非常抱歉给您带来了不好的体验。我们已经记录了您的问题【{}】，将尽快反馈给专人处理。请问是否需要现在转接人工客服？".format(state["user_input"]) return {"need_human": True, "messages": [("assistant", response)]}

5.2 构建图并设置条件边

这是LangGraph最核心的部分：定义执行流程。

from langgraph.graph import StateGraph, END # 1. 创建图 workflow = StateGraph(State) # 2. 添加节点 workflow.add_node("intent_router", intent_router_node) workflow.add_node("handle_greeting", handle_greeting_node) workflow.add_node("handle_inquiry", handle_inquiry_node) workflow.add_node("handle_complaint", handle_complaint_node) # 3. 设置入口点 workflow.set_entry_point("intent_router") # 4. 设置条件边：根据`intent`字段的值决定下一步走向 def decide_next_node(state: State): intent = state.get("intent") if intent == "greeting": return "handle_greeting" elif intent == "inquiry": return "handle_inquiry" elif intent == "complaint": return "handle_complaint" else: # 无法识别，默认按问候处理 return "handle_greeting" workflow.add_conditional_edges( "intent_router", decide_next_node, { "handle_greeting": "handle_greeting", "handle_inquiry": "handle_inquiry", "handle_complaint": "handle_complaint", } ) # 5. 从各个处理节点指向结束 workflow.add_edge("handle_greeting", END) workflow.add_edge("handle_inquiry", END) workflow.add_edge("handle_complaint", END) # 6. 编译图 app = workflow.compile()

5.3 运行与可视化

现在，我们可以运行这个图了。

# 初始化状态 initial_state = { "user_input": "你们的产品太差了，我要投诉！", "messages": [("user", "你们的产品太差了，我要投诉！")], "intent": "", "product_info": None, "need_human": False } # 执行图 final_state = app.invoke(initial_state) print("最终回复：", final_state["messages"][-1][1]) # 取最后一条助手消息 print("是否需要人工：", final_state["need_human"])

执行后，intent_router_node会判断意图为“complaint”，然后路由到handle_complaint_node，最终状态中的need_human会变为True，并生成相应的回复。

调试利器：你可以用app.get_graph().draw_mermaid()生成流程图的Mermaid代码（在支持的环境下），直观地看到你设计的整个决策流程。这对于理解和沟通复杂业务逻辑有巨大帮助。

6. 避坑指南与性能优化实战记录

在真实项目中，你会遇到各种各样的问题。下面是我和社区伙伴们总结的一些常见坑和优化技巧。

6.1 提示词工程：让Agent更听话

Agent不按你期望的方式调用工具？多半是提示词的问题。

问题：Agent忽略工具，直接用自己的知识回答。解决：在提示词中明确指令。使用create_react_agent时，它自带的提示词已经不错。但如果你自定义，务必包含：

• 工具描述：清晰说明每个工具是干什么的，输入格式是什么。
• 强制指令：明确告诉Agent“你必须使用提供的工具来回答问题。如果你没有合适的工具，请直接说明无法回答，不要编造信息。”
• 示例（Few-Shot）：在提示词里给一两个正确使用工具的示例，效果立竿见影。

6.2 错误处理与稳定性

问题：网络波动或API限制导致整个链崩溃。解决：使用重试和回退机制。

from langchain.llms import OpenAI from tenacity import retry, stop_after_attempt, wait_exponential # 为LLM调用添加重试逻辑 @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) def robust_llm_call(prompt): # ... 调用LLM的代码 pass # 或者在LangChain中，可以使用带有重试功能的自定义LLM包装类

对于关键应用，可以考虑设置一个“降级”方案，比如当主模型（GPT-4）失败或超时时，自动切换到备用模型（GPT-3.5或本地模型）。

6.3 成本与延迟优化

问题：应用响应慢，Token消耗大，成本高。解决：

1. 缓存：对相同的查询进行缓存。LangChain提供了InMemoryCache、SQLiteCache等。对于向量检索，可以缓存嵌入结果。
2. 摘要与压缩：在对话记忆或检索上下文中，使用ConversationSummaryMemory或上下文压缩检索器（ContextualCompressionRetriever），只保留最相关的信息，减少输入的Token数量。
3. 模型分级：对于简单的意图分类、路由等任务，使用小模型（如gpt-3.5-turbo甚至更小的本地模型）。只在需要复杂推理和生成的环节使用大模型（如GPT-4）。
4. 异步调用：如果你的应用有多个可以并行执行的步骤（比如同时查询多个数据源），使用LangChain的异步接口（ainvoke,abatch）可以显著降低整体延迟。

6.4 评估与监控

问题：如何知道我的Agent表现好不好？解决：建立评估体系。

• 单元测试：为每个工具和关键节点编写测试，确保功能正常。
• 端到端评估：准备一批有标准答案的测试问题（Q&A对），定期运行你的Agent，计算答案的相似度或使用LLM（如GPT-4）作为裁判来评分。
• 日志与追踪：利用LangSmith（LangChain官方平台）或自定义日志，记录每一次链的执行步骤、耗时、Token使用量和中间结果。这是分析和优化性能的黄金数据。

7. 从教程到项目：下一步学习路径建议

通过这个教程，你应该已经对LangChain和LangGraph的核心概念和基础用法有了实战级的理解。接下来，如果你想深入下去，我建议按这个路径来：

1. 吃透官方文档与教程案例：把LangChain-OpenTutorial仓库里的例子一个个跑通，并尝试修改它们，比如换用不同的模型、添加新的工具、修改提示词看看效果如何变化。
2. 复现一个经典应用：选择一个你感兴趣的场景，比如个人知识库问答、智能数据分析助手、自动化客服，尝试用LangChain完整地实现它。这个过程会遇到最多的问题，也是成长最快的阶段。
3. 深入研究高级模式：探索更复杂的Agent架构，如Plan-and-Execute（让Agent先制定计划再执行）、Multi-Agent（多个Agent协作）、以及利用LangGraph实现具有长期记忆和复杂状态管理的智能体。
4. 关注生态与部署：了解如何将你的LangChain应用打包成API服务（使用FastAPI）、如何与前端集成、以及如何部署到生产环境（考虑容器化、监控、扩缩容）。
5. 参与社区：像这个开源教程项目一样，社区的力量是巨大的。遇到问题时，在GitHub Issues、Discord或相关论坛提问。当你解决了某个棘手问题，不妨也像本教程的贡献者们一样，写一篇总结，回馈社区。在帮助别人的过程中，你自己的理解也会更加深刻。

记住，这个领域变化飞快，新的模型、工具和最佳实践不断涌现。保持动手实践的习惯，定期回看像LangChain-OpenTutorial这样的优质资源，是跟上节奏的最好方法。