尧图网站建设 
1. 先搞清楚 LangGraph、LangChain 和 Agent 到底能帮你做什么如果你正在看 AI 大模型应用开发尤其是想搞点能“自己动起来”的智能应用那 LangGraph、LangChain 和 Agent 这三个词肯定绕不开。很多人一上来就被各种教程和概念搞晕了不知道从哪下手。其实你不需要先啃完 500 多集视频核心就一件事用代码把大模型的能力组织成一个能处理复杂、多步骤任务的“工作流”或“智能体”。LangChain 是你的工具箱和脚手架。它提供了连接大模型、处理文本、管理记忆、调用工具的基础组件。但当你需要处理“先做 A再做 B根据 B 的结果决定做 C 还是 D”这类有状态、有分支的任务时原生的 LangChain 链条Chain就显得有点力不从心了。这时候LangGraph 就上场了。LangGraph 的核心价值是让你能用“图”Graph的思维来设计和运行智能体Agent。你可以把每个步骤比如调用一次大模型、执行一个工具、做一个判断看作一个节点Node步骤之间的流转关系看作边Edge。这样你就能清晰地构建出循环、条件分支、并行执行等复杂逻辑。这对于开发客服助手、数据分析机器人、自动化流程等需要多轮交互和决策的应用至关重要。而Agent智能体在这里可以理解为一个具备自主规划、调用工具、持续执行能力的程序实体。LangGraph 是构建这类高级 Agent 的强力框架。所以这个主题解决的实际问题是当你不再满足于让大模型简单问答而是希望它能像“员工”一样按照你设定的流程自主、可靠地完成一连串任务时你应该用什么工具、怎么搭建、如何调试。这篇文章就围绕这个目标拆解从环境准备到实战落地的全过程。2. 环境准备别在依赖和版本上栽跟头动手之前环境是第一个坎。很多“跑不通”的问题根源都在这里。我建议完全按照以下顺序来可以避开 80% 的初期坑。2.1 基础 Python 环境首先需要一个干净的 Python 环境。强烈建议使用conda或venv创建独立的虚拟环境避免包冲突。# 使用 conda 创建环境假设你已安装 Anaconda 或 Miniconda conda create -n langgraph-demo python3.10 -y conda activate langgraph-demo # 或者使用 venv python -m venv langgraph-demo # Windows langgraph-demo\Scripts\activate # macOS/Linux source langgraph-demo/bin/activatePython 版本选择3.10或3.11是目前最稳妥的对新旧包的兼容性最好。不建议直接用最新的 3.12 或 3.13有些底层依赖可能还没适配。2.2 核心库安装核心就是安装langchain和langgraph。注意langchain是一个大家族我们通常需要安装核心包和社区包。pip install langchain langgraph这行命令会安装最新稳定版。但“最新”有时意味着 API 变动。如果你跟着某个特定教程比如标题提到的 549 集最好先确认一下教程使用的版本。你可以通过pip install langchain0.1.0 langgraph0.0.1来指定版本这里的版本号是举例需替换为实际版本。一个关键点langchain包本身不包含大模型。你需要额外安装对应模型的 SDK 或使用 HTTP 客户端。例如如果你用 OpenAI 的模型pip install openai如果你打算本地部署开源模型比如通过 Ollama则需要安装对应的集成包pip install langchain-communitylangchain-community包含了大量第三方集成如 Ollama、Hugging Face 等是扩展能力所必需的。2.3 模型访问配置无论你用云端 API 还是本地模型都需要配置访问权限或地址。OpenAI/Azure 等云端 API需要设置环境变量。# Windows (cmd) setx OPENAI_API_KEY your-api-key-here # Windows (PowerShell) $env:OPENAI_API_KEY your-api-key-here # macOS/Linux export OPENAI_API_KEYyour-api-key-here更推荐在代码中通过os.environ设置或在.env文件中管理。本地模型如 Ollama确保 Ollama 服务已启动。通常安装后它会自动运行在http://localhost:11434。LangChain 通过langchain-community中的ChatOllama类来调用。2.4 验证安装环境装好后不要急着写复杂 Agent先用一个最简单的 LangChain 调用验证整个链路是否通畅。import os from langchain_openai import ChatOpenAI # 1. 设置 API Key (如果是本地模型这部分不同) os.environ[OPENAI_API_KEY] your-api-key # 2. 初始化一个最简单的聊天模型 llm ChatOpenAI(modelgpt-3.5-turbo) # 3. 发起一次调用 response llm.invoke(Hello, world!) print(response.content)如果这一步能成功输出 “Hello, world!” 或其他问候语的回复说明你的 Python 环境、langchain、openaiSDK 和网络连接都是好的。这是所有后续工作的基石。注意如果这里就报错比如ModuleNotFoundError回去检查pip install的包名是否正确如果是认证错误检查 API Key如果是网络超时检查代理或防火墙设置。务必把这一步调通再继续。3. 从 LangChain Chain 到 LangGraph State理解核心范式转变在直接动手写 LangGraph 之前必须理解它和传统 LangChain Chain 的根本区别。这能帮你避免用 Chain 的思维去写 Graph导致代码别扭、功能受限。3.1 LangChain Chain线性管道在 LangChain 中一个典型的 Chain 是这样的from langchain.prompts import ChatPromptTemplate from langchain.chains import LLMChain prompt ChatPromptTemplate.from_template(请将{input}翻译成英文。) chain LLMChain(llmllm, promptprompt) result chain.run(input今天天气真好) print(result) # 输出The weather is really nice today.这是一个线性流程输入 - 模板格式化 - 调用 LLM - 输出。它强大、简单适合大多数单步或固定顺序的多步任务。3.2 LangGraph 的 State状态和 Node节点LangGraph 引入了两个核心概念State一个字典或 Pydantic 模型用于在整个工作流执行过程中传递和共享数据。它就像是智能体的“记忆白板”所有节点都能读写上面的信息。Node一个函数接收当前的 State执行一些操作如调用 LLM、运行工具然后返回一个对 State 的更新。工作流Graph由多个 Node 和连接它们的边决定下一个执行哪个 Node组成。执行引擎会根据边的逻辑在 Node 之间跳转直到到达终点。3.3 一个极简的 LangGraph 示例对话循环我们来实现一个最简单的智能体它和你对话并且能记住之前说过的话短期记忆。from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, END # 1. 定义状态结构。这是一个“白板”。 class AgentState(TypedDict): messages: Annotated[list, operator.add] # 关键这是一个累加器会自动追加消息。 # 2. 定义节点函数 def call_model(state: AgentState): 节点调用大模型生成回复 from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-3.5-turbo) # 将历史消息列表直接传给模型 response llm.invoke(state[messages]) # 将模型的回复也加入到消息列表中 new_message {role: assistant, content: response.content} # 返回对状态的更新。这里更新了 messages 字段。 return {messages: [new_message]} # 3. 构建图 graph_builder StateGraph(AgentState) # 添加一个节点命名为 “model” graph_builder.add_node(model, call_model) # 设置入口点从 “model” 节点开始 graph_builder.set_entry_point(model) # 设置出口点“model” 节点执行完后就结束。 graph_builder.add_edge(model, END) # 编译图得到一个可执行对象 graph graph_builder.compile() # 4. 运行这个智能体 initial_state {messages: [{role: user, content: 你好我叫小明。}]} result graph.invoke(initial_state) print(f第一轮回复: {result[messages][-1][content]}) # 继续对话状态中已经包含了历史 next_state {messages: result[messages] [{role: user, content: 我刚才说我叫什么名字}]} result2 graph.invoke(next_state) print(f第二轮回复: {result2[messages][-1][content]}) # 模型应该能回答“小明”这个例子虽然简单但包含了 LangGraph 的所有精髓AgentState定义了工作流中流动的数据。call_model是一个节点函数它读写 State。StateGraph用来组装节点和边。graph.invoke()是执行入口。关键理解Annotated[list, operator.add]这个类型提示是 LangGraph 的魔法。它告诉框架messages这个字段在每次节点返回更新时应该用append追加的方式而不是覆盖。这就是实现“记忆”功能的底层机制。4. 构建实用智能体加入工具调用与条件路由只会对话的 Agent 能力有限。真正的智能体需要能“动手”做事情比如搜索网络、查询数据库、执行代码。这就是工具调用Tool Calling。同时智能体需要能根据结果做判断决定下一步做什么这就是条件路由Conditional Edge。4.1 为智能体装备工具我们扩展上面的例子给智能体加一个“计算器”工具和一个“网络搜索”工具这里用模拟。from langchain.tools import tool from langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor from langgraph.prebuilt import create_react_agent_executor import json # 1. 定义工具 tool def calculator(expression: str) - str: 计算一个数学表达式的值。支持 , -, *, /, **。 try: # 警告实际生产中应对 eval 做严格安全检查这里仅为演示。 result eval(expression) return f计算结果: {result} except Exception as e: return f计算错误: {e} tool def search_web(query: str) - str: 模拟网络搜索。返回模拟结果。 # 这里模拟返回真实场景可集成 SerperAPI、Tavily 等。 return f关于 {query} 的模拟搜索结果相关资讯1相关资讯2。 # 工具列表 tools [calculator, search_web] # 2. 使用 LangGraph 预置的 ReAct 智能体执行器 # ReAct 是一个经典的 Agent 推理框架Reasoning Acting llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) agent_executor create_react_agent_executor(llm, tools) # 3. 运行智能体 question 先计算 (15 27) * 2 等于多少然后搜索一下最新的AI新闻。 inputs {messages: [(user, question)]} # 注意create_react_agent_executor 返回的本身就是一个 LangGraph 对象 for step in agent_executor.stream(inputs): print(f步骤输出: {step}) print(---)create_react_agent_executor是 LangGraph 提供的一个高级封装它内部已经构建好了一个标准的 ReAct 智能体图。这个图包含的节点通常有Agent 节点分析当前状态和任务决定下一步是调用工具还是直接回复。工具调用节点执行被选中的工具。路由逻辑根据 Agent 节点的输出决定下一步是去执行工具还是结束流程。当你运行这段代码时你会看到流式输出展示智能体“思考”决定调用哪个工具、“行动”调用工具并获取结果、再“思考”的完整过程。4.2 实现自定义条件路由预置的执行器很方便但有时你需要更精细的控制。比如一个客服智能体如果用户问的是“产品价格”就路由到“查询价格表”节点如果是“投诉”就路由到“创建工单”节点。下面我们手动构建一个带条件路由的图。from langgraph.graph import StateGraph, END from typing import Literal from langchain_core.messages import HumanMessage, AIMessage # 1. 定义更丰富的状态 class RoutingState(TypedDict): messages: Annotated[list, operator.add] needs_human: bool # 一个标志位表示是否需要人工介入 # 2. 定义几个节点 def classify_intent(state: RoutingState): 节点1分类用户意图 latest_msg state[messages][-1].content if state[messages] else # 这里简化处理实际应用中应该用更复杂的LLM调用或分类器 if 价格 in latest_msg or 多少钱 in latest_msg: return {intent: price_query, needs_human: False} elif 投诉 in latest_msg or 不满意 in latest_msg: return {intent: complaint, needs_human: True} # 投诉需要人工 else: return {intent: general, needs_human: False} def handle_price_query(state: RoutingState): 节点2处理价格查询 # 模拟查询数据库或知识库 price_info 产品A: 100元产品B: 200元。 new_msg AIMessage(contentf根据您的查询价格信息如下{price_info}) return {messages: [new_msg]} def handle_complaint(state: RoutingState): 节点3处理投诉创建工单 new_msg AIMessage(content已为您创建投诉工单 #12345客服专员将尽快联系您。) return {messages: [new_msg]} def escalate_to_human(state: RoutingState): 节点4转接人工 new_msg AIMessage(content您的问题已转接给人工客服请稍候。) return {messages: [new_msg]} # 3. 构建图 builder StateGraph(RoutingState) # 添加节点 builder.add_node(classify, classify_intent) builder.add_node(answer_price, handle_price_query) builder.add_node(create_ticket, handle_complaint) builder.add_node(human_agent, escalate_to_human) # 设置入口 builder.set_entry_point(classify) # 定义条件路由函数 def route_after_classify(state: RoutingState) - Literal[answer_price, create_ticket, human_agent, __end__]: 根据 classify 节点设置的 intent 和 needs_human 决定下一步 intent state.get(intent, general) needs_human state.get(needs_human, False) if needs_human: return human_agent # 需要人工去 human_agent 节点 elif intent price_query: return answer_price # 价格查询去 answer_price 节点 elif intent complaint: return create_ticket # 投诉去 create_ticket 节点 else: return __end__ # 其他情况直接结束 # 添加条件边 builder.add_conditional_edges( classify, # 从哪个节点出发 route_after_classify, # 路由决策函数 { answer_price: answer_price, create_ticket: create_ticket, human_agent: human_agent, __end__: END } ) # 为其他节点添加普通边执行完就去结束 builder.add_edge(answer_price, END) builder.add_edge(create_ticket, END) builder.add_edge(human_agent, END) # 编译图 routing_graph builder.compile() # 4. 测试 print(测试1: 询问价格) result1 routing_graph.invoke({messages: [HumanMessage(content产品A多少钱)]}) print(result1[messages][-1].content) print(\n测试2: 我要投诉) result2 routing_graph.invoke({messages: [HumanMessage(content我对服务不满意我要投诉)]}) print(result2[messages][-1].content)这个例子展示了 LangGraph 最强大的能力之一基于状态的动态路由。add_conditional_edges让你能根据任意复杂的业务逻辑来决定工作流的走向这是构建复杂、健壮智能体的基础。5. 记忆Memory管理从短期对话到长期记忆智能体的“记忆”是其智能的核心体现。在 LangGraph 中记忆主要通过State来管理。但根据记忆的时长和用途我们需要不同的策略。5.1 短期对话记忆Conversation Memory上面的例子已经实现了短期记忆把所有消息都保存在state[‘messages’]列表中。每次新的交互都传入完整的历史模型就能基于上下文进行回复。这是最简单也最常用的方式。问题当对话轮次非常多时这个列表会变得很长可能超出模型的上下文窗口限制Context Window导致最开始的对话被“遗忘”。解决方案摘要式记忆Summarization定期比如每 10 轮对话让模型对之前的对话历史进行总结然后用一个简短的摘要替换掉冗长的原始历史只保留最近几轮原始对话。滑动窗口记忆Sliding Window只保留最近 N 条消息。这可以通过在状态更新函数中截断列表来实现但会丢失早期信息。向量存储记忆VectorStore-Backed Memory将历史对话存入向量数据库如 Chroma, FAISS每次需要时根据当前问题检索最相关的历史片段。这是实现“长期记忆”和“知识库”的关键。5.2 实现基于向量数据库的长期记忆我们来实现一个能将对话存入向量库并能根据当前问题检索相关历史的智能体。from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_core.documents import Document from langchain.chains import create_history_aware_retriever, create_retrieval_chain from langchain.chains.combine_documents import create_stuff_documents_chain from langchain.chains import create_retrieval_chain from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator # 1. 初始化向量数据库这里使用内存版的 Chroma embeddings OpenAIEmbeddings() vectorstore Chroma(embedding_functionembeddings, persist_directoryNone) # 内存存储 # 2. 定义状态。除了消息我们还需要存储“记忆”的 ID。 class LongMemoryState(TypedDict): messages: Annotated[list, operator.add] memory_ids: list # 用于存储本轮对话存入向量库的文档 ID # 3. 定义节点将对话存入记忆库 def save_to_memory(state: LongMemoryState): 将最新的用户和助理对话保存到向量数据库 if len(state[messages]) 2: return {memory_ids: []} # 没有完整的对话轮次不保存 # 获取最近一轮对话假设最后两条消息是 user 和 assistant recent_convo state[messages][-2:] # 将对话内容合并成一个文本 convo_text \n.join([f{msg.type}: {msg.content} for msg in recent_convo]) # 创建文档对象 doc Document(page_contentconvo_text) # 存入向量库并获取存储的 ID ids vectorstore.add_documents([doc]) # 返回新的 memory_ids追加到状态中 return {memory_ids: state.get(memory_ids, []) ids} # 4. 定义节点从记忆库中检索相关历史 def retrieve_from_memory(state: LongMemoryState): 根据当前问题从向量库中检索相关历史对话 if not state[messages]: return {relevant_history: } latest_question state[messages][-1].content # 从向量库中检索与当前问题最相关的 3 个历史片段 docs vectorstore.similarity_search(latest_question, k3) relevant_history \n---\n.join([doc.page_content for doc in docs]) return {relevant_history: relevant_history} # 5. 定义节点调用 LLM 生成回复结合检索到的历史 def generate_with_memory(state: LongMemoryState): 结合检索到的长期记忆和短期对话历史生成回复 llm ChatOpenAI(modelgpt-3.5-turbo) # 构建提示词注入相关历史 prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的助手。请参考以下相关历史对话来回答用户问题。\n相关历史\n{relevant_history}), MessagesPlaceholder(variable_namemessages), # 短期对话历史 ]) chain prompt | llm # 调用链传入相关历史和短期消息 response chain.invoke({ relevant_history: state.get(relevant_history, ), messages: state[messages] }) new_msg AIMessage(contentresponse.content) return {messages: [new_msg]} # 6. 构建图 builder StateGraph(LongMemoryState) builder.add_node(save_memory, save_to_memory) builder.add_node(retrieve_memory, retrieve_from_memory) builder.add_node(generate, generate_with_memory) builder.set_entry_point(save_memory) # 流程先保存记忆 - 然后检索记忆 - 最后生成回复 builder.add_edge(save_memory, retrieve_memory) builder.add_edge(retrieve_memory, generate) builder.add_edge(generate, END) memory_graph builder.compile() # 7. 测试 print(第一轮对话询问编程语言) state1 memory_graph.invoke({ messages: [HumanMessage(contentPython 和 Java 哪个更适合初学者)], memory_ids: [] }) print(f助理回复: {state1[messages][-1].content}\n) print(第二轮对话基于之前的话题深入) state2 memory_graph.invoke({ messages: state1[messages] [HumanMessage(content你刚才提到的 Python它的主要应用领域是什么)], memory_ids: state1[memory_ids] }) print(f助理回复: {state2[messages][-1].content}) # 注意在这个简化示例中第二轮的检索可能会找到第一轮对话的向量记录从而在回复中体现“记忆”。这个架构让智能体具备了“长期记忆”能力。即使对话间隔了很久只要问题相关它就能从向量库中找回当时的讨论上下文。这对于构建知识型助手、个性化聊天机器人至关重要。5.3 记忆的持久化与加载上面的例子中Chroma使用了内存存储程序重启后记忆就消失了。在生产环境中你需要指定persist_directory参数将向量库保存到磁盘。在智能体初始化时加载已存在的向量库。考虑记忆的清理策略比如按时间过期、按重要性打分等。6. 实战避坑与生产化考量把 Demo 跑起来只是第一步。要让智能体真正可用、可靠你需要关注以下这些实战细节。6.1 错误处理与重试智能体在调用外部工具、模型 API 时很可能失败。一个健壮的 Graph 必须有错误处理机制。节点级 Try-Catch在每个可能出错的节点函数内部进行异常捕获并返回一个标记错误的状态更新。def call_external_api(state): try: result some_unstable_api() return {api_result: result} except Exception as e: # 记录错误并决定下一步是重试、转人工还是优雅失败 return {error: str(e), should_retry: True}LangGraph 的检查点Checkpoint与回滚LangGraph 支持检查点机制可以在每个节点执行后保存状态。如果后续节点失败理论上可以回滚到上一个检查点。这对于长流程、高价值任务非常重要。设置超时Timeout在调用工具或模型时务必设置超时防止因网络或服务问题导致整个智能体挂起。6.2 流式输出与用户体验对于需要长时间运行的复杂智能体流式输出Streaming用户体验好很多。LangGraph 原生支持stream方法。# 使用之前创建的 agent_executor 或 graph inputs {messages: [(user, 请写一首关于春天的诗。)]} for chunk in agent_executor.stream(inputs): # chunk 是一个字典包含当前步骤的输出 if agent in chunk: print(f思考: {chunk[agent][messages][-1].content}) elif tools in chunk: print(f执行工具: {chunk[tools][messages][-1].content}) elif messages in chunk: # 最终输出消息 for msg in chunk[messages]: if msg.type ai: print(msg.content, end, flushTrue) # 流式打印 AI 回复在前端界面上你可以将这些chunk实时推送给用户展示智能体的“思考过程”和“打字机效果”的回复。6.3 性能监控与调试日志记录在每个节点的入口和出口记录详细的日志包括输入状态、输出状态、耗时、错误信息。使用结构化日志如 JSON 格式便于后续分析。可视化LangGraph 的一个巨大优势是图结构清晰。使用graph.get_graph().draw_mermaid()可以生成 Mermaid 图代码可视化你的智能体工作流这对于复杂流程的调试和团队沟通极其有用。跟踪Tracing集成 LangSmithLangChain 官方的监控平台或类似的 APM 工具可以追踪每次调用的链式关系、耗时、token 消耗和成本是生产部署的必备项。6.4 与 Web 框架集成如 Flask/FastAPI智能体最终需要以 API 的形式提供服务。以 FastAPI 为例from fastapi import FastAPI, HTTPException from pydantic import BaseModel from your_agent_module import compiled_graph # 导入你编译好的 LangGraph app FastAPI() class ChatRequest(BaseModel): message: str session_id: str # 用于区分不同会话/用户 class ChatResponse(BaseModel): reply: str session_id: str app.post(/chat, response_modelChatResponse) async def chat_endpoint(request: ChatRequest): try: # 1. 根据 session_id 从数据库或缓存中加载历史状态 # initial_state load_state(request.session_id) # 这里简化处理每次都是新会话 initial_state {messages: [(user, request.message)]} # 2. 调用智能体图 result compiled_graph.invoke(initial_state) # 3. 提取最终回复 final_reply result[messages][-1].content # 4. 将更新后的状态保存回数据库或缓存 # save_state(request.session_id, result) return ChatResponse(replyfinal_reply, session_idrequest.session_id) except Exception as e: raise HTTPException(status_code500, detailstr(e))关键点在于状态管理。你需要将会话 IDsession_id与每个用户的AgentState持久化关联起来确保每次对话都能在正确的记忆上下文中进行。7. 总结从入门到实战的路径建议看完这些你可能觉得内容很多。别担心按照这个路径来你会走得更稳第一步夯实基础。确保 Python 环境、LangChain/LangGraph 安装、基础模型调用OpenAI 或 Ollama全部跑通。这是所有工作的地基。第二步理解 State 和 Node。亲手写一个最简单的、带状态传递的 LangGraph比如那个对话循环。理解State如何流动Node如何更新它。第三步玩转工具调用。使用create_react_agent_executor快速创建一个能调用工具的智能体观察它的 ReAct 推理过程。尝试自己定义一两个简单工具。第四步设计条件路由。针对一个具体场景如客服分类手动构建一个带有add_conditional_edges的图。这是构建复杂业务逻辑的核心。第五步引入记忆管理。从简单的对话历史列表升级到基于向量数据库的长期记忆。理解检索增强生成RAG如何与智能体结合。第六步生产化改造。为你的智能体加上错误处理、日志、流式输出并把它封装成 API。考虑状态持久化数据库和并发请求处理。最后也是最关键的一点不要追求一次性构建一个万能智能体。从解决一个非常具体、微小的问题开始比如“查天气并推荐穿衣”把它做完善。然后逐步增加分支、工具和记忆。在迭代过程中你会对 LangGraph 的抽象有更深的理解从而能更自信地驾驭更复杂的智能体架构。 LangGraph实战指南：从零构建具备记忆与工具调用的智能体工作流
1. 先搞清楚 LangGraph、LangChain 和 Agent 到底能帮你做什么如果你正在看 AI 大模型应用开发尤其是想搞点能“自己动起来”的智能应用那 LangGraph、LangChain 和 Agent 这三个词肯定绕不开。很多人一上来就被各种教程和概念搞晕了不知道从哪下手。其实你不需要先啃完 500 多集视频核心就一件事用代码把大模型的能力组织成一个能处理复杂、多步骤任务的“工作流”或“智能体”。LangChain 是你的工具箱和脚手架。它提供了连接大模型、处理文本、管理记忆、调用工具的基础组件。但当你需要处理“先做 A再做 B根据 B 的结果决定做 C 还是 D”这类有状态、有分支的任务时原生的 LangChain 链条Chain就显得有点力不从心了。这时候LangGraph 就上场了。LangGraph 的核心价值是让你能用“图”Graph的思维来设计和运行智能体Agent。你可以把每个步骤比如调用一次大模型、执行一个工具、做一个判断看作一个节点Node步骤之间的流转关系看作边Edge。这样你就能清晰地构建出循环、条件分支、并行执行等复杂逻辑。这对于开发客服助手、数据分析机器人、自动化流程等需要多轮交互和决策的应用至关重要。而Agent智能体在这里可以理解为一个具备自主规划、调用工具、持续执行能力的程序实体。LangGraph 是构建这类高级 Agent 的强力框架。所以这个主题解决的实际问题是当你不再满足于让大模型简单问答而是希望它能像“员工”一样按照你设定的流程自主、可靠地完成一连串任务时你应该用什么工具、怎么搭建、如何调试。这篇文章就围绕这个目标拆解从环境准备到实战落地的全过程。2. 环境准备别在依赖和版本上栽跟头动手之前环境是第一个坎。很多“跑不通”的问题根源都在这里。我建议完全按照以下顺序来可以避开 80% 的初期坑。2.1 基础 Python 环境首先需要一个干净的 Python 环境。强烈建议使用conda或venv创建独立的虚拟环境避免包冲突。# 使用 conda 创建环境假设你已安装 Anaconda 或 Miniconda conda create -n langgraph-demo python3.10 -y conda activate langgraph-demo # 或者使用 venv python -m venv langgraph-demo # Windows langgraph-demo\Scripts\activate # macOS/Linux source langgraph-demo/bin/activatePython 版本选择3.10或3.11是目前最稳妥的对新旧包的兼容性最好。不建议直接用最新的 3.12 或 3.13有些底层依赖可能还没适配。2.2 核心库安装核心就是安装langchain和langgraph。注意langchain是一个大家族我们通常需要安装核心包和社区包。pip install langchain langgraph这行命令会安装最新稳定版。但“最新”有时意味着 API 变动。如果你跟着某个特定教程比如标题提到的 549 集最好先确认一下教程使用的版本。你可以通过pip install langchain0.1.0 langgraph0.0.1来指定版本这里的版本号是举例需替换为实际版本。一个关键点langchain包本身不包含大模型。你需要额外安装对应模型的 SDK 或使用 HTTP 客户端。例如如果你用 OpenAI 的模型pip install openai如果你打算本地部署开源模型比如通过 Ollama则需要安装对应的集成包pip install langchain-communitylangchain-community包含了大量第三方集成如 Ollama、Hugging Face 等是扩展能力所必需的。2.3 模型访问配置无论你用云端 API 还是本地模型都需要配置访问权限或地址。OpenAI/Azure 等云端 API需要设置环境变量。# Windows (cmd) setx OPENAI_API_KEY your-api-key-here # Windows (PowerShell) $env:OPENAI_API_KEY your-api-key-here # macOS/Linux export OPENAI_API_KEYyour-api-key-here更推荐在代码中通过os.environ设置或在.env文件中管理。本地模型如 Ollama确保 Ollama 服务已启动。通常安装后它会自动运行在http://localhost:11434。LangChain 通过langchain-community中的ChatOllama类来调用。2.4 验证安装环境装好后不要急着写复杂 Agent先用一个最简单的 LangChain 调用验证整个链路是否通畅。import os from langchain_openai import ChatOpenAI # 1. 设置 API Key (如果是本地模型这部分不同) os.environ[OPENAI_API_KEY] your-api-key # 2. 初始化一个最简单的聊天模型 llm ChatOpenAI(modelgpt-3.5-turbo) # 3. 发起一次调用 response llm.invoke(Hello, world!) print(response.content)如果这一步能成功输出 “Hello, world!” 或其他问候语的回复说明你的 Python 环境、langchain、openaiSDK 和网络连接都是好的。这是所有后续工作的基石。注意如果这里就报错比如ModuleNotFoundError回去检查pip install的包名是否正确如果是认证错误检查 API Key如果是网络超时检查代理或防火墙设置。务必把这一步调通再继续。3. 从 LangChain Chain 到 LangGraph State理解核心范式转变在直接动手写 LangGraph 之前必须理解它和传统 LangChain Chain 的根本区别。这能帮你避免用 Chain 的思维去写 Graph导致代码别扭、功能受限。3.1 LangChain Chain线性管道在 LangChain 中一个典型的 Chain 是这样的from langchain.prompts import ChatPromptTemplate from langchain.chains import LLMChain prompt ChatPromptTemplate.from_template(请将{input}翻译成英文。) chain LLMChain(llmllm, promptprompt) result chain.run(input今天天气真好) print(result) # 输出The weather is really nice today.这是一个线性流程输入 - 模板格式化 - 调用 LLM - 输出。它强大、简单适合大多数单步或固定顺序的多步任务。3.2 LangGraph 的 State状态和 Node节点LangGraph 引入了两个核心概念State一个字典或 Pydantic 模型用于在整个工作流执行过程中传递和共享数据。它就像是智能体的“记忆白板”所有节点都能读写上面的信息。Node一个函数接收当前的 State执行一些操作如调用 LLM、运行工具然后返回一个对 State 的更新。工作流Graph由多个 Node 和连接它们的边决定下一个执行哪个 Node组成。执行引擎会根据边的逻辑在 Node 之间跳转直到到达终点。3.3 一个极简的 LangGraph 示例对话循环我们来实现一个最简单的智能体它和你对话并且能记住之前说过的话短期记忆。from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, END # 1. 定义状态结构。这是一个“白板”。 class AgentState(TypedDict): messages: Annotated[list, operator.add] # 关键这是一个累加器会自动追加消息。 # 2. 定义节点函数 def call_model(state: AgentState): 节点调用大模型生成回复 from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-3.5-turbo) # 将历史消息列表直接传给模型 response llm.invoke(state[messages]) # 将模型的回复也加入到消息列表中 new_message {role: assistant, content: response.content} # 返回对状态的更新。这里更新了 messages 字段。 return {messages: [new_message]} # 3. 构建图 graph_builder StateGraph(AgentState) # 添加一个节点命名为 “model” graph_builder.add_node(model, call_model) # 设置入口点从 “model” 节点开始 graph_builder.set_entry_point(model) # 设置出口点“model” 节点执行完后就结束。 graph_builder.add_edge(model, END) # 编译图得到一个可执行对象 graph graph_builder.compile() # 4. 运行这个智能体 initial_state {messages: [{role: user, content: 你好我叫小明。}]} result graph.invoke(initial_state) print(f第一轮回复: {result[messages][-1][content]}) # 继续对话状态中已经包含了历史 next_state {messages: result[messages] [{role: user, content: 我刚才说我叫什么名字}]} result2 graph.invoke(next_state) print(f第二轮回复: {result2[messages][-1][content]}) # 模型应该能回答“小明”这个例子虽然简单但包含了 LangGraph 的所有精髓AgentState定义了工作流中流动的数据。call_model是一个节点函数它读写 State。StateGraph用来组装节点和边。graph.invoke()是执行入口。关键理解Annotated[list, operator.add]这个类型提示是 LangGraph 的魔法。它告诉框架messages这个字段在每次节点返回更新时应该用append追加的方式而不是覆盖。这就是实现“记忆”功能的底层机制。4. 构建实用智能体加入工具调用与条件路由只会对话的 Agent 能力有限。真正的智能体需要能“动手”做事情比如搜索网络、查询数据库、执行代码。这就是工具调用Tool Calling。同时智能体需要能根据结果做判断决定下一步做什么这就是条件路由Conditional Edge。4.1 为智能体装备工具我们扩展上面的例子给智能体加一个“计算器”工具和一个“网络搜索”工具这里用模拟。from langchain.tools import tool from langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor from langgraph.prebuilt import create_react_agent_executor import json # 1. 定义工具 tool def calculator(expression: str) - str: 计算一个数学表达式的值。支持 , -, *, /, **。 try: # 警告实际生产中应对 eval 做严格安全检查这里仅为演示。 result eval(expression) return f计算结果: {result} except Exception as e: return f计算错误: {e} tool def search_web(query: str) - str: 模拟网络搜索。返回模拟结果。 # 这里模拟返回真实场景可集成 SerperAPI、Tavily 等。 return f关于 {query} 的模拟搜索结果相关资讯1相关资讯2。 # 工具列表 tools [calculator, search_web] # 2. 使用 LangGraph 预置的 ReAct 智能体执行器 # ReAct 是一个经典的 Agent 推理框架Reasoning Acting llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) agent_executor create_react_agent_executor(llm, tools) # 3. 运行智能体 question 先计算 (15 27) * 2 等于多少然后搜索一下最新的AI新闻。 inputs {messages: [(user, question)]} # 注意create_react_agent_executor 返回的本身就是一个 LangGraph 对象 for step in agent_executor.stream(inputs): print(f步骤输出: {step}) print(---)create_react_agent_executor是 LangGraph 提供的一个高级封装它内部已经构建好了一个标准的 ReAct 智能体图。这个图包含的节点通常有Agent 节点分析当前状态和任务决定下一步是调用工具还是直接回复。工具调用节点执行被选中的工具。路由逻辑根据 Agent 节点的输出决定下一步是去执行工具还是结束流程。当你运行这段代码时你会看到流式输出展示智能体“思考”决定调用哪个工具、“行动”调用工具并获取结果、再“思考”的完整过程。4.2 实现自定义条件路由预置的执行器很方便但有时你需要更精细的控制。比如一个客服智能体如果用户问的是“产品价格”就路由到“查询价格表”节点如果是“投诉”就路由到“创建工单”节点。下面我们手动构建一个带条件路由的图。from langgraph.graph import StateGraph, END from typing import Literal from langchain_core.messages import HumanMessage, AIMessage # 1. 定义更丰富的状态 class RoutingState(TypedDict): messages: Annotated[list, operator.add] needs_human: bool # 一个标志位表示是否需要人工介入 # 2. 定义几个节点 def classify_intent(state: RoutingState): 节点1分类用户意图 latest_msg state[messages][-1].content if state[messages] else # 这里简化处理实际应用中应该用更复杂的LLM调用或分类器 if 价格 in latest_msg or 多少钱 in latest_msg: return {intent: price_query, needs_human: False} elif 投诉 in latest_msg or 不满意 in latest_msg: return {intent: complaint, needs_human: True} # 投诉需要人工 else: return {intent: general, needs_human: False} def handle_price_query(state: RoutingState): 节点2处理价格查询 # 模拟查询数据库或知识库 price_info 产品A: 100元产品B: 200元。 new_msg AIMessage(contentf根据您的查询价格信息如下{price_info}) return {messages: [new_msg]} def handle_complaint(state: RoutingState): 节点3处理投诉创建工单 new_msg AIMessage(content已为您创建投诉工单 #12345客服专员将尽快联系您。) return {messages: [new_msg]} def escalate_to_human(state: RoutingState): 节点4转接人工 new_msg AIMessage(content您的问题已转接给人工客服请稍候。) return {messages: [new_msg]} # 3. 构建图 builder StateGraph(RoutingState) # 添加节点 builder.add_node(classify, classify_intent) builder.add_node(answer_price, handle_price_query) builder.add_node(create_ticket, handle_complaint) builder.add_node(human_agent, escalate_to_human) # 设置入口 builder.set_entry_point(classify) # 定义条件路由函数 def route_after_classify(state: RoutingState) - Literal[answer_price, create_ticket, human_agent, __end__]: 根据 classify 节点设置的 intent 和 needs_human 决定下一步 intent state.get(intent, general) needs_human state.get(needs_human, False) if needs_human: return human_agent # 需要人工去 human_agent 节点 elif intent price_query: return answer_price # 价格查询去 answer_price 节点 elif intent complaint: return create_ticket # 投诉去 create_ticket 节点 else: return __end__ # 其他情况直接结束 # 添加条件边 builder.add_conditional_edges( classify, # 从哪个节点出发 route_after_classify, # 路由决策函数 { answer_price: answer_price, create_ticket: create_ticket, human_agent: human_agent, __end__: END } ) # 为其他节点添加普通边执行完就去结束 builder.add_edge(answer_price, END) builder.add_edge(create_ticket, END) builder.add_edge(human_agent, END) # 编译图 routing_graph builder.compile() # 4. 测试 print(测试1: 询问价格) result1 routing_graph.invoke({messages: [HumanMessage(content产品A多少钱)]}) print(result1[messages][-1].content) print(\n测试2: 我要投诉) result2 routing_graph.invoke({messages: [HumanMessage(content我对服务不满意我要投诉)]}) print(result2[messages][-1].content)这个例子展示了 LangGraph 最强大的能力之一基于状态的动态路由。add_conditional_edges让你能根据任意复杂的业务逻辑来决定工作流的走向这是构建复杂、健壮智能体的基础。5. 记忆Memory管理从短期对话到长期记忆智能体的“记忆”是其智能的核心体现。在 LangGraph 中记忆主要通过State来管理。但根据记忆的时长和用途我们需要不同的策略。5.1 短期对话记忆Conversation Memory上面的例子已经实现了短期记忆把所有消息都保存在state[‘messages’]列表中。每次新的交互都传入完整的历史模型就能基于上下文进行回复。这是最简单也最常用的方式。问题当对话轮次非常多时这个列表会变得很长可能超出模型的上下文窗口限制Context Window导致最开始的对话被“遗忘”。解决方案摘要式记忆Summarization定期比如每 10 轮对话让模型对之前的对话历史进行总结然后用一个简短的摘要替换掉冗长的原始历史只保留最近几轮原始对话。滑动窗口记忆Sliding Window只保留最近 N 条消息。这可以通过在状态更新函数中截断列表来实现但会丢失早期信息。向量存储记忆VectorStore-Backed Memory将历史对话存入向量数据库如 Chroma, FAISS每次需要时根据当前问题检索最相关的历史片段。这是实现“长期记忆”和“知识库”的关键。5.2 实现基于向量数据库的长期记忆我们来实现一个能将对话存入向量库并能根据当前问题检索相关历史的智能体。from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_core.documents import Document from langchain.chains import create_history_aware_retriever, create_retrieval_chain from langchain.chains.combine_documents import create_stuff_documents_chain from langchain.chains import create_retrieval_chain from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator # 1. 初始化向量数据库这里使用内存版的 Chroma embeddings OpenAIEmbeddings() vectorstore Chroma(embedding_functionembeddings, persist_directoryNone) # 内存存储 # 2. 定义状态。除了消息我们还需要存储“记忆”的 ID。 class LongMemoryState(TypedDict): messages: Annotated[list, operator.add] memory_ids: list # 用于存储本轮对话存入向量库的文档 ID # 3. 定义节点将对话存入记忆库 def save_to_memory(state: LongMemoryState): 将最新的用户和助理对话保存到向量数据库 if len(state[messages]) 2: return {memory_ids: []} # 没有完整的对话轮次不保存 # 获取最近一轮对话假设最后两条消息是 user 和 assistant recent_convo state[messages][-2:] # 将对话内容合并成一个文本 convo_text \n.join([f{msg.type}: {msg.content} for msg in recent_convo]) # 创建文档对象 doc Document(page_contentconvo_text) # 存入向量库并获取存储的 ID ids vectorstore.add_documents([doc]) # 返回新的 memory_ids追加到状态中 return {memory_ids: state.get(memory_ids, []) ids} # 4. 定义节点从记忆库中检索相关历史 def retrieve_from_memory(state: LongMemoryState): 根据当前问题从向量库中检索相关历史对话 if not state[messages]: return {relevant_history: } latest_question state[messages][-1].content # 从向量库中检索与当前问题最相关的 3 个历史片段 docs vectorstore.similarity_search(latest_question, k3) relevant_history \n---\n.join([doc.page_content for doc in docs]) return {relevant_history: relevant_history} # 5. 定义节点调用 LLM 生成回复结合检索到的历史 def generate_with_memory(state: LongMemoryState): 结合检索到的长期记忆和短期对话历史生成回复 llm ChatOpenAI(modelgpt-3.5-turbo) # 构建提示词注入相关历史 prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的助手。请参考以下相关历史对话来回答用户问题。\n相关历史\n{relevant_history}), MessagesPlaceholder(variable_namemessages), # 短期对话历史 ]) chain prompt | llm # 调用链传入相关历史和短期消息 response chain.invoke({ relevant_history: state.get(relevant_history, ), messages: state[messages] }) new_msg AIMessage(contentresponse.content) return {messages: [new_msg]} # 6. 构建图 builder StateGraph(LongMemoryState) builder.add_node(save_memory, save_to_memory) builder.add_node(retrieve_memory, retrieve_from_memory) builder.add_node(generate, generate_with_memory) builder.set_entry_point(save_memory) # 流程先保存记忆 - 然后检索记忆 - 最后生成回复 builder.add_edge(save_memory, retrieve_memory) builder.add_edge(retrieve_memory, generate) builder.add_edge(generate, END) memory_graph builder.compile() # 7. 测试 print(第一轮对话询问编程语言) state1 memory_graph.invoke({ messages: [HumanMessage(contentPython 和 Java 哪个更适合初学者)], memory_ids: [] }) print(f助理回复: {state1[messages][-1].content}\n) print(第二轮对话基于之前的话题深入) state2 memory_graph.invoke({ messages: state1[messages] [HumanMessage(content你刚才提到的 Python它的主要应用领域是什么)], memory_ids: state1[memory_ids] }) print(f助理回复: {state2[messages][-1].content}) # 注意在这个简化示例中第二轮的检索可能会找到第一轮对话的向量记录从而在回复中体现“记忆”。这个架构让智能体具备了“长期记忆”能力。即使对话间隔了很久只要问题相关它就能从向量库中找回当时的讨论上下文。这对于构建知识型助手、个性化聊天机器人至关重要。5.3 记忆的持久化与加载上面的例子中Chroma使用了内存存储程序重启后记忆就消失了。在生产环境中你需要指定persist_directory参数将向量库保存到磁盘。在智能体初始化时加载已存在的向量库。考虑记忆的清理策略比如按时间过期、按重要性打分等。6. 实战避坑与生产化考量把 Demo 跑起来只是第一步。要让智能体真正可用、可靠你需要关注以下这些实战细节。6.1 错误处理与重试智能体在调用外部工具、模型 API 时很可能失败。一个健壮的 Graph 必须有错误处理机制。节点级 Try-Catch在每个可能出错的节点函数内部进行异常捕获并返回一个标记错误的状态更新。def call_external_api(state): try: result some_unstable_api() return {api_result: result} except Exception as e: # 记录错误并决定下一步是重试、转人工还是优雅失败 return {error: str(e), should_retry: True}LangGraph 的检查点Checkpoint与回滚LangGraph 支持检查点机制可以在每个节点执行后保存状态。如果后续节点失败理论上可以回滚到上一个检查点。这对于长流程、高价值任务非常重要。设置超时Timeout在调用工具或模型时务必设置超时防止因网络或服务问题导致整个智能体挂起。6.2 流式输出与用户体验对于需要长时间运行的复杂智能体流式输出Streaming用户体验好很多。LangGraph 原生支持stream方法。# 使用之前创建的 agent_executor 或 graph inputs {messages: [(user, 请写一首关于春天的诗。)]} for chunk in agent_executor.stream(inputs): # chunk 是一个字典包含当前步骤的输出 if agent in chunk: print(f思考: {chunk[agent][messages][-1].content}) elif tools in chunk: print(f执行工具: {chunk[tools][messages][-1].content}) elif messages in chunk: # 最终输出消息 for msg in chunk[messages]: if msg.type ai: print(msg.content, end, flushTrue) # 流式打印 AI 回复在前端界面上你可以将这些chunk实时推送给用户展示智能体的“思考过程”和“打字机效果”的回复。6.3 性能监控与调试日志记录在每个节点的入口和出口记录详细的日志包括输入状态、输出状态、耗时、错误信息。使用结构化日志如 JSON 格式便于后续分析。可视化LangGraph 的一个巨大优势是图结构清晰。使用graph.get_graph().draw_mermaid()可以生成 Mermaid 图代码可视化你的智能体工作流这对于复杂流程的调试和团队沟通极其有用。跟踪Tracing集成 LangSmithLangChain 官方的监控平台或类似的 APM 工具可以追踪每次调用的链式关系、耗时、token 消耗和成本是生产部署的必备项。6.4 与 Web 框架集成如 Flask/FastAPI智能体最终需要以 API 的形式提供服务。以 FastAPI 为例from fastapi import FastAPI, HTTPException from pydantic import BaseModel from your_agent_module import compiled_graph # 导入你编译好的 LangGraph app FastAPI() class ChatRequest(BaseModel): message: str session_id: str # 用于区分不同会话/用户 class ChatResponse(BaseModel): reply: str session_id: str app.post(/chat, response_modelChatResponse) async def chat_endpoint(request: ChatRequest): try: # 1. 根据 session_id 从数据库或缓存中加载历史状态 # initial_state load_state(request.session_id) # 这里简化处理每次都是新会话 initial_state {messages: [(user, request.message)]} # 2. 调用智能体图 result compiled_graph.invoke(initial_state) # 3. 提取最终回复 final_reply result[messages][-1].content # 4. 将更新后的状态保存回数据库或缓存 # save_state(request.session_id, result) return ChatResponse(replyfinal_reply, session_idrequest.session_id) except Exception as e: raise HTTPException(status_code500, detailstr(e))关键点在于状态管理。你需要将会话 IDsession_id与每个用户的AgentState持久化关联起来确保每次对话都能在正确的记忆上下文中进行。7. 总结从入门到实战的路径建议看完这些你可能觉得内容很多。别担心按照这个路径来你会走得更稳第一步夯实基础。确保 Python 环境、LangChain/LangGraph 安装、基础模型调用OpenAI 或 Ollama全部跑通。这是所有工作的地基。第二步理解 State 和 Node。亲手写一个最简单的、带状态传递的 LangGraph比如那个对话循环。理解State如何流动Node如何更新它。第三步玩转工具调用。使用create_react_agent_executor快速创建一个能调用工具的智能体观察它的 ReAct 推理过程。尝试自己定义一两个简单工具。第四步设计条件路由。针对一个具体场景如客服分类手动构建一个带有add_conditional_edges的图。这是构建复杂业务逻辑的核心。第五步引入记忆管理。从简单的对话历史列表升级到基于向量数据库的长期记忆。理解检索增强生成RAG如何与智能体结合。第六步生产化改造。为你的智能体加上错误处理、日志、流式输出并把它封装成 API。考虑状态持久化数据库和并发请求处理。最后也是最关键的一点不要追求一次性构建一个万能智能体。从解决一个非常具体、微小的问题开始比如“查天气并推荐穿衣”把它做完善。然后逐步增加分支、工具和记忆。在迭代过程中你会对 LangGraph 的抽象有更深的理解从而能更自信地驾驭更复杂的智能体架构。