2026/8/18 12:42:26
FEATURED · 精选文章 
如果你正在构建复杂的AI应用特别是那些需要多步骤决策、状态管理和循环逻辑的Agent系统那么你很可能已经感受到了传统LangChain在复杂流程编排上的局限性。当你的应用从简单的问答链发展到需要记忆、工具调用、条件分支和循环的复杂工作流时你会发现单纯用SequentialChain或LLMChain拼接代码会迅速变得难以维护和调试。这正是LangGraph要解决的核心痛点。它不是LangChain的替代品而是其“大脑”和“调度中心”。LangGraph将你的AI应用建模为一个有向图Graph节点是处理单元LLM调用、工具执行、代码运行边是控制流顺序、条件、循环。这种范式转变让开发者可以用声明式的方式构建复杂的、带状态的、可循环的AI应用就像用流程图设计业务逻辑一样直观。本文将为你提供一份从零到一的LangGraph实战指南。我们不仅会详细讲解其核心概念和安装部署更会通过一个完整的、可运行的案例带你理解如何构建一个具备“长期记忆”和“工具调用”能力的智能客服Agent。同时我们会将LangGraph与LangfuseAI应用可观测性平台、量化感知训练QAT和SFTTrainer监督微调训练器等热门技术栈结合探讨如何构建一个可观测、可优化、高性能的完整AI应用工程体系。读完本文你将能清晰理解LangGraph与LangChain的区别与定位知道何时该用它。独立完成LangGraph环境搭建并运行你的第一个Graph。掌握State设计、节点、边、条件边等核心概念并能设计自己的应用流程。学会将LangGraph应用与Langfuse集成实现完整的调用链追踪与评估。了解如何将训练好的模型特别是经过SFT和QAT优化的模型接入LangGraph应用并理解其中的工程考量。1. LangGraph为什么它是构建复杂AI Agent的“游戏规则改变者”在深入代码之前我们必须先建立一个关键认知LangGraph解决的是“复杂流程编排”问题而不是“基础功能调用”问题。想象一下这些场景一个客服Agent需要先理解用户意图然后查询知识库如果答案不完整再去调用外部API最后根据用户反馈决定是结束对话还是继续追问。一个数据分析Agent需要循环执行“提出问题 - 编写代码 - 执行代码 - 检查结果”的步骤直到得到满意的分析报告。一个游戏NPC需要根据玩家的历史对话和当前状态决定下一步是提供线索、发起战斗还是改变剧情分支。这些场景的共同点是有状态、多步骤、带循环和条件判断。用传统的LangChain链式结构来写你会陷入大量的if-else嵌套和手动状态传递中代码可读性和可维护性极差。LangGraph带来的核心改变图即程序你将应用逻辑画成一张图每个节点是一个清晰的功能单元如“调用LLM”、“执行工具”每条边代表明确的流转条件。这极大地提升了代码的可视化和可理解性。显式的状态管理所有节点共享一个中心化的State对象。你无需在函数间手动传递参数只需定义好State的结构LangGraph会自动帮你管理和更新。内置的循环与控制流通过conditional_edges条件边和将边指向END或某个节点你可以轻松实现while循环、if-else分支等复杂逻辑这是传统链难以优雅实现的。更好的调试与可观测性由于流程是结构化的你更容易追踪每个节点的输入输出也更容易与像Langfuse这样的可观测性平台集成进行全链路追踪。简单对比LangChain像“乐高积木”提供了丰富的组件Models, Prompts, Chains, Agents, Tools, Memory。适合快速搭建标准化的、线性的AI管道。LangGraph像“流程图设计器”和“调度引擎”。它使用LangChain的组件作为“积木”但负责如何将这些积木按照复杂的逻辑循环、分支组装并运行起来。它让LangChain如虎添翼。所以如果你的应用逻辑是简单的线性链LangChain足够。但一旦涉及循环、状态持久化、复杂决策LangGraph几乎是目前最优雅的解决方案。2. 核心概念解析State、Node、Edge与Graph理解LangGraph关键是理解它的几个核心抽象。我们将用一个“智能点餐助手”的类比来贯穿讲解。2.1 State应用的“记忆白板”State是一个字典或Pydantic模型它存储了应用在整个运行过程中的所有数据。你可以把它想象成餐厅服务员手里的点餐单。这张单子有固定的格式定义好的字段比如customer_query顾客需求、current_order当前已点菜品、need_confirmation是否需要确认等。每个服务环节节点都可以读取和修改这张单子上的内容。LangGraph负责在节点间传递这张最新的“点餐单”。from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator # 定义State的结构。我们使用TypedDict这是LangGraph推荐的方式。 class AgentState(TypedDict): # 存储对话历史。Annotated和add_messages是LangGraph提供的工具用于自动合并消息。 messages: Annotated[List, add_messages] # 用户的最新输入 user_input: str # Agent的思考过程或中间结果 agent_scratchpad: str # 一个标志位控制流程是否继续 should_continue: bool关键点Annotated[List, add_messages]是一个“缩减器”reducer它告诉LangGraph在更新messages字段时不是替换而是将新消息追加到列表末尾。这是实现对话记忆的关键。2.2 Node功能执行单元Node就是一个普通的Python函数或可调用对象它接收当前的State执行一些操作然后返回一个包含更新后字段的字典。 继续我们的类比Node就是餐厅里的不同角色接待员Node接收顾客原始需求(user_input)更新到点餐单(State)上。推荐员Node根据当前已点菜品(current_order)调用LLM推荐新菜并更新推荐结果到点餐单。确认员Node生成订单总结并询问顾客是否确认。def reception_node(state: AgentState) - dict: 接待节点处理用户初始输入 # 从State中获取用户输入 user_message state[“user_input”] # 将用户输入转换为AI消息格式并添加到历史中 # 注意我们返回的是要更新到State中的字段。 # LangGraph会自动将这个返回的字典与旧的State合并。 return {“messages”: [{“role”: “user”, “content”: user_message}]} def llm_agent_node(state: AgentState) - dict: AI代理节点调用LLM进行思考并可能调用工具 # 1. 从State中构建给LLM的提示词包括历史消息和工具描述 messages state[“messages”] # ... (这里会构造一个包含系统提示、历史、工具列表的messages) # 2. 调用LLM # response chat_model.invoke(messages) # 3. 解析LLM响应判断是直接回复还是调用了工具 # 如果是工具调用这里会执行工具并将结果追加到消息历史 # 我们将LLM的回复也追加到消息历史 ai_message {“role”: “assistant”, “content”: “这是AI的思考过程和回复...”} return {“messages”: [ai_message], “agent_scratchpad”: “一些中间思考...”}2.3 Edge控制流方向Edge定义了节点之间的执行顺序。分为两种普通边Linear Edge无条件地从上一个节点指向下一个节点。条件边Conditional Edge根据State中的某个条件值决定下一步走向哪个节点。这是实现分支和循环的关键。2.4 Graph将一切组装起来Graph是节点和边的容器。你创建Graph添加节点然后用边将它们连接起来。from langgraph.graph import StateGraph, END # 1. 创建一个Graph并指定它使用的State类型 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(“reception”, reception_node) workflow.add_node(“agent”, llm_agent_node) workflow.add_node(“confirm”, confirmation_node) # 3. 设置入口点第一个执行的节点 workflow.set_entry_point(“reception”) # 4. 添加普通边reception - agent workflow.add_edge(“reception”, “agent”) # 5. 添加条件边从agent节点出来后根据State中的should_continue字段决定下一步 def route_after_agent(state: AgentState): if state.get(“should_continue”, False): # 如果需要继续则返回下一个节点的名字这里我们循环回agent自身实现while效果 return “agent” else: # 如果不需要继续则结束流程 return “confirm” workflow.add_conditional_edges( “agent”, # 从哪个节点出来 route_after_agent, # 路由判断函数 {“agent”: “agent”, “confirm”: “confirm”} # 可能的目的地映射 ) # 6. 从confirm节点到结束 workflow.add_edge(“confirm”, END) # 7. 编译Graph得到一个可执行的对象 app workflow.compile()这个Graph定义了一个流程接待 - AI代理可能循环多次- 确认 - 结束。conditional_edges使得AI代理节点可以基于should_continue标志实现循环。3. 环境准备与安装LangGraph是一个Python库。我们建议在一个新的虚拟环境中安装以避免依赖冲突。3.1 创建并激活虚拟环境# 使用conda conda create -n langgraph-demo python3.10 conda activate langgraph-demo # 或使用venv python -m venv venv # Windows venv\Scripts\activate # Linux/Mac source venv/bin/activate3.2 安装核心库我们将安装LangGraph、LangChain提供LLM集成和基础工具、以及一个本地LLM的运行环境这里以使用Ollama运行开源模型为例。同时安装Langfuse用于后续的可观测性集成。# 安装 LangGraph 和 LangChain pip install langgraph langchain langchain-community # 安装 Langfuse SDK (用于追踪和评估) pip install langfuse # 安装 Ollama (用于本地运行大模型如 Llama 3.1, Qwen2.5 等) # 请根据你的操作系统从 https://ollama.com/ 下载并安装Ollama # 安装后在终端拉取一个模型例如 # ollama pull llama3.1:8b # 可选安装其他有用的工具链 pip install jupyterlab # 用于在笔记本中交互式开发 pip install python-dotenv # 用于管理环境变量3.3 验证安装创建一个简单的Python脚本test_install.py来验证基础环境。# test_install.py import langgraph import langchain print(f“LangGraph version: {langgraph.__version__}”) print(f“LangChain version: {langchain.__version__}”) # 尝试导入Langfuse try: import langfuse print(“Langfuse imported successfully.”) except ImportError as e: print(f“Failed to import Langfuse: {e}”)运行它python test_install.py如果看到版本号且没有报错说明核心环境已就绪。4. 实战案例构建一个具备长期记忆和工具调用的智能客服Agent现在我们来构建一个真实的智能客服Agent。它的功能是记忆对话历史能记住用户之前说过的话。调用工具能查询产品知识库模拟和获取当前时间。自主决策循环根据LLM的判断决定是直接回答、调用工具还是结束对话。4.1 定义State和工具首先我们定义State和两个简单的工具。# agent_demo.py from typing import TypedDict, List, Annotated from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_community.chat_models import ChatOllama # 使用Ollama from langchain.tools import tool from datetime import datetime import json # --- 1. 定义State --- class AgentState(TypedDict): # 对话消息历史使用add_messages缩减器自动追加 messages: Annotated[List, add_messages] # 一个标志由LLM或节点设置控制是否继续 should_continue: bool # --- 2. 定义工具 --- tool def search_product_knowledgebase(query: str) - str: 根据用户问题查询产品知识库。这是一个模拟函数。 # 模拟一个简单的知识库 kb { “退货政策”: “商品签收后7天内可无理由退货需保持商品完好。详情请见官网。”, “运费”: “订单满99元包邮不满99元收取10元运费。”, “保修期”: “所有电子产品提供一年保修从购买日算起。”, “客服时间”: “人工客服工作时间为每天9:00-21:00。” } # 简单关键词匹配 for key, answer in kb.items(): if key.lower() in query.lower(): return f“关于【{key}】: {answer}” return “在知识库中没有找到相关信息请尝试联系人工客服。” tool def get_current_time() - str: 获取当前的日期和时间。 now datetime.now() return now.strftime(“%Y-%m-%d %H:%M:%S”) # 将工具包装成列表供LLM使用 tools [search_product_knowledgebase, get_current_time] # --- 3. 初始化LLM --- # 确保你已经通过 ollama pull llama3.1:8b 拉取了模型 llm ChatOllama(model“llama3.1:8b”, temperature0) # 让LLM绑定工具这样它才知道自己能调用什么 llm_with_tools llm.bind_tools(tools)4.2 构建核心的Agent Node这个Node是Graph的核心它负责与LLM交互并处理工具调用。# --- 4. 定义Agent Node函数 --- def call_agent(state: AgentState): print(f“\n[DEBUG] Entering call_agent node. Message history length: {len(state[‘messages’])}”) # 1. 准备给LLM的消息。State中的messages已经包含了完整的历史。 messages state[“messages”] # 2. 调用绑定了工具的LLM response llm_with_tools.invoke(messages) # 3. 检查LLM的响应是否是工具调用 if response.tool_calls: print(f“[DEBUG] LLM decided to call tools: {response.tool_calls}”) # 初始化一个列表用于存放工具执行结果的消息 tool_messages [] for tool_call in response.tool_calls: # 根据工具名找到对应的工具函数 tool_to_call {t.name: t for t in tools}[tool_call[‘name’]] # 执行工具 tool_result tool_to_call.invoke(tool_call[‘args’]) # 将工具执行结果构造成ToolMessage并添加到列表 tool_messages.append(ToolMessage( contentstr(tool_result), tool_call_idtool_call[‘id’], nametool_call[‘name’] )) # 返回更新将LLM的响应包含工具调用请求和所有工具执行结果都追加到消息历史 # 注意这里我们返回的是要追加的消息列表LangGraph的add_messages缩减器会处理合并。 return {“messages”: [response] tool_messages, “should_continue”: True} else: # LLM直接给出了最终回复 print(f“[DEBUG] LLM gave final response: {response.content[:50]}...”) # 将AI的回复追加到历史 return {“messages”: [response], “should_continue”: False} # 可以结束对话了4.3 定义路由逻辑和构建Graph我们需要一个函数来判断在Agent Node执行后下一步应该去哪是继续循环让Agent处理工具结果还是结束# --- 5. 定义路由函数 --- def should_continue(state: AgentState) - str: “”“根据state中的should_continue标志决定下一步。 返回下一个节点的名称。”“” if state.get(“should_continue”, False): # 需要继续则返回”agent”节点名形成循环 return “agent” else: # 不需要继续则结束 return “__end__” # --- 6. 构建并编译Graph --- # 创建图 workflow StateGraph(AgentState) # 添加唯一的节点我们的核心Agent workflow.add_node(“agent”, call_agent) # 设置入口点 workflow.set_entry_point(“agent”) # 添加条件边。这是关键 # 从”agent”节点出来后根据should_continue函数的结果决定去向。 workflow.add_conditional_edges( “agent”, # 源节点 should_continue, # 路由函数 # 路由函数返回值到节点名的映射。”__end__”是LangGraph预定义的结束标识。 {“agent”: “agent”, “__end__”: END} ) # 编译成可执行的应用 app workflow.compile()4.4 运行与测试现在让我们运行这个Agent并模拟一段多轮对话。# --- 7. 运行Graph --- if __name__ “__main__”: # 初始化状态从用户的第一条消息开始 initial_state: AgentState { “messages”: [ # 系统提示词定义Agent的角色和能力 {“role”: “system”, “content”: “你是一个智能客服助手可以回答关于产品政策的问题也可以查询当前时间。请根据用户问题决定是直接回答还是调用工具。调用工具后请根据工具结果进行总结回复。用中文回答。”} ], “should_continue”: True } # 模拟用户输入 user_inputs [ “你们的退货政策是什么”, “那运费怎么算呢”, “现在几点了” ] for i, query in enumerate(user_inputs): print(f“\n{*50}”) print(f“用户第{i1}轮输入: {query}”) print(f“{*50}”) # 将用户输入转换为HumanMessage并更新到State中 # 注意我们通过app.invoke传入一个包含更新内容的字典。 # LangGraph会将这个字典与当前的State合并然后执行Graph。 inputs {“messages”: [HumanMessage(contentquery)]} # 执行Graph final_state app.invoke(inputs, initial_state) # 更新initial_state以便下一轮对话有历史记忆 initial_state final_state # 打印最后一轮AI的回复 last_message final_state[“messages”][-1] if hasattr(last_message, ‘content’): print(f“\n助手回复: {last_message.content}”) else: print(f“\n助手最后动作: {last_message}”) # 简单打印整个对话历史可选 # print(“\n当前完整对话历史:”) # for msg in final_state[“messages”]: # print(f“{msg.type}: {msg.content}”)运行这个脚本你将看到类似以下的输出清晰地展示了Agent的思考、工具调用和循环过程 用户第1轮输入: 你们的退货政策是什么 [DEBUG] Entering call_agent node. Message history length: 1 [DEBUG] LLM decided to call tools: [{‘name’: ‘search_product_knowledgebase’, ‘args’: {‘query’: ‘退货政策’}, ‘id’: ‘call_xxx’}] [DEBUG] Entering call_agent node. Message history length: 3 [DEBUG] LLM gave final response: 根据查询结果我们的退货政策是... 助手回复: 根据查询结果我们的退货政策是商品签收后7天内可无理由退货需保持商品完好。详情请见官网。这个输出表明第一轮LLM决定调用search_product_knowledgebase工具。工具执行后Graph根据should_continueTrue的条件循环再次进入call_agent节点。第二轮LLM收到了工具返回的结果并生成了最终的自然语言回复同时设置should_continueFalse。Graph结束我们得到了助手的最终回复。这就是LangGraph的核心魔力通过条件边和状态循环优雅地处理了“LLM调用工具 - 获取结果 - LLM再次处理”这个经典Agent流程。5. 集成Langfuse为你的Agent添加“眼睛”和“大脑”构建复杂的Agent只是第一步。在生产环境中你需要监控它的表现每次调用花了多少钱LLM的回复质量如何工具调用成功了吗这就是可观测性Observability的范畴而Langfuse是此领域的佼佼者。Langfuse可以自动追踪LangChain/LangGraph应用的每一次LLM调用、工具执行形成完整的溯源链Trace。你可以看到整个Graph的执行脉络每个节点的输入输出以及消耗的Token和成本。5.1 初始化Langfuse并集成到Graph中首先你需要在 Langfuse Cloud 注册并获取API密钥。# langfuse_integration.py from langfuse import Langfuse from langfuse.callback import CallbackHandler import os from dotenv import load_dotenv # 加载环境变量将你的Langfuse密钥保存在.env文件中 # LANGFUSE_SECRET_KEY“sk-lf-...” # LANGFUSE_PUBLIC_KEY“pk-lf-...” # LANGFUSE_HOST“https://cloud.langfuse.com” # 或你的自托管地址 load_dotenv() # 初始化Langfuse客户端 langfuse Langfuse( secret_keyos.getenv(“LANGFUSE_SECRET_KEY”), public_keyos.getenv(“LANGFUSE_PUBLIC_KEY”), hostos.getenv(“LANGFUSE_HOST”, “https://cloud.langfuse.com”) ) # 创建Langfuse回调处理器 langfuse_handler CallbackHandler( langfuselangfuse, # 为本次运行设置一个会话名称方便在Langfuse界面查找 session_name“my_langgraph_agent_demo” ) # --- 修改Graph的调用方式注入回调处理器 --- # 假设app是上一节编译好的Graph if __name__ “__main__”: initial_state {…} # 同上一节 inputs {“messages”: [HumanMessage(content“你们的退货政策是什么”)]} # 关键在invoke时传入callbacks参数 final_state app.invoke( inputs, initial_state, config{“callbacks”: [langfuse_handler]} # 注入Langfuse回调 ) # 确保所有追踪数据都发送到Langfuse服务器 langfuse_handler.langfuse.flush() print(“追踪数据已发送至Langfuse。”)运行这段代码后登录Langfuse Cloud的Dashboard你就能在“Traces”页面看到一个完整的追踪记录。点击进入可以看到清晰的层级结构Trace (根): 代表一次完整的app.invoke调用。Span (节点): 代表call_agent节点的执行。Generation (LLM调用): 在Span内部记录了一次具体的LLM调用包括提示词、回复、Token使用和成本。Event (工具调用): 记录工具search_product_knowledgebase的调用和结果。5.2 利用Langfuse进行提示词工程和评估Langfuse不仅是监控工具更是优化AI应用的利器。提示词版本管理你可以在Langfuse中修改系统提示词并快速比较不同提示词下Agent的表现。手动评分与反馈在界面上可以直接对某次Trace或LLM回复进行打分、添加评论这些数据可以用于后续的模型微调。数据集与评测你可以将一些测试用例导入为数据集然后让Agent批量运行自动计算关键指标如正确率、相关性形成评估报告。通过Langfuse你将Agent从“黑盒”变成了“白盒”能够系统地分析性能瓶颈、优化提示词、并收集高质量的训练数据。6. 接入优化模型SFTTrainer与量化感知训练QAT的工程实践我们的Agent核心是LLM。为了获得更好、更快、更便宜的模型我们常常需要对基础模型进行监督微调SFT和量化Quantization。这里我们探讨如何将经过这些流程优化后的模型接入LangGraph。6.1 监督微调SFT与SFTTrainerSFT是指在特定任务数据上继续训练预训练模型使其更擅长该任务。transformers库中的SFTTrainer是一个方便的工具。典型SFT流程准备数据整理成(instruction, output)的对话格式。加载基础模型如Qwen2.5-7B-Instruct。配置训练参数学习率、批次大小、训练轮数等。使用SFTTrainer训练。保存模型。训练完成后你会得到一个适配你任务的微调模型。在LangGraph中你只需要像使用任何其他ChatModel一样使用它。# 假设你有一个使用SFTTrainer微调后保存的模型目录 ./my_finetuned_model from langchain_community.chat_models import ChatOllama # 使用Ollama加载本地微调模型 # 首先你需要将HuggingFace格式的模型转换为Ollama支持的格式如GGUF并用Ollama加载。 # 例如: ollama create my-agent -f ./Modelfile (其中Modelfile指定了模型文件路径) # 然后在代码中指定这个模型名 sft_llm ChatOllama(model“my-agent”, temperature0, num_ctx4096) # 后续在构建LangGraph的Agent Node时使用这个sft_llm代替通用的llm即可。6.2 量化感知训练QAT与部署量化是将模型权重从高精度如FP32转换为低精度如INT8/INT4的过程能显著减少模型大小和推理延迟。量化感知训练QAT是在训练或微调过程中模拟量化效应让模型在训练时就适应低精度从而在真正量化后精度损失更小。对于LangGraph开发者你通常不需要自己实施QAT而是使用社区已经量化好的模型。例如从Hugging Face Model Hub下载TheBloke系列提供的GGUF量化模型。接入量化模型的步骤选择模型例如TheBloke/Llama-3.2-1B-Instruct-GGUF。下载量化文件选择适合你硬件的量化等级如Q4_K_M。使用Ollama或llama.cpp加载# 为Ollama创建Modelfile # FROM ./llama-3.2-1b-instruct.Q4_K_M.gguf # 然后 ollama create my-quantized-model -f ./Modelfile在LangGraph中调用quantized_llm ChatOllama(model“my-quantized-model”, temperature0)关键建议在LangGraph应用中优先考虑使用已经量化好的、适合你硬件的小尺寸模型。QAT是一个高级模型优化技术通常由模型提供商或算法工程师在发布模型前完成。应用开发者更应关注如何高效地集成和调用这些优化后的模型。7. 常见问题与排查指南在开发LangGraph应用时你可能会遇到以下典型问题。问题现象可能原因排查方式解决方案RuntimeError: ...或节点执行报错1. State结构定义与节点返回值不匹配。2. 工具函数参数传递错误。3. LLM响应格式不符合预期。1. 在节点函数开头打印state检查其结构。2. 检查工具tool装饰器的参数定义。3. 打印LLM的原始响应response。1. 确保节点返回的字典键名与State的TypedDict定义一致。2. 确保工具函数参数是基本类型str, int等。3. 使用try-except包裹LLM调用并检查response的属性。Graph陷入无限循环1. 条件边(conditional_edges)的路由逻辑有误始终返回循环条件。2. LLM在工具调用后仍然设置should_continueTrue。1. 在路由函数中打印state和判断逻辑。2. 检查系统提示词明确告诉LLM何时结束任务。1. 在路由函数中添加更严格的终止条件。2. 优化提示词例如“如果你已经获得了足够信息并给出了最终答案请将should_continue设为false。”Langfuse中没有Trace数据1. API密钥或Host配置错误。2. 回调处理器没有被正确注入。3. 网络问题或flush()未调用。1. 检查环境变量是否正确加载。2. 在invoke后检查langfuse_handler是否有错误属性。3. 查看Langfuse控制台或SDK日志。1. 使用langfuse.auth_check()验证连接。2. 确保config{“callbacks”: [langfuse_handler]}传入invoke。3. 调用langfuse_handler.langfuse.flush()并等待。Ollama模型加载失败或响应慢1. 模型名称错误或未下载。2. 硬件资源RAM/VRAM不足。3. Ollama服务未启动。1. 运行ollama list确认模型存在。2. 查看系统资源监控。3. 检查Ollama服务状态ollama serve。1. 使用ollama pull下载正确模型。2. 换用更小的量化模型如3B、7B的Q4量化。3. 确保Ollama在后台运行。工具调用未被LLM触发1. LLM未正确绑定工具。2. 系统提示词未明确说明工具用途。3. 工具描述不够清晰。1. 检查llm.bind_tools(tools)是否执行。2. 查看LLM收到的完整提示词。3. 检查tool装饰器中的函数文档字符串。1. 确认bind_tools调用成功。2. 在系统提示词中加入“你可以使用以下工具...”。3. 优化工具的描述使其更精准。8. 最佳实践与进阶建议掌握了基础之后遵循以下最佳实践能让你的LangGraph应用更加健壮和可维护。State设计要精简而充分精简只存储流程必需的数据。避免将整个对话历史、中间计算结果等全部塞进State这会影响序列化/反序列化效率。充分确保所有节点需要的信息都能从State中获取。常用的设计模式是将messages对话历史和scratchpad中间思考分开。节点函数保持单一职责每个节点只做一件事。例如一个节点专门调用LLM另一个节点专门处理工具执行结果。这提高了可测试性和复用性。善用条件边实现复杂逻辑不要试图在一个节点里用if-else处理所有分支。将不同的分支路径建模为不同的节点用条件边连接。这样逻辑更清晰也便于在LangGraph Studio等可视化工具中查看。集成Langfuse进行全链路可观测在开发初期就集成Langfuse。它能帮你快速定位是哪个节点、哪次LLM调用或工具执行出了问题。利用其评分和数据集功能持续优化你的Agent。模型选择与优化开发/测试环境使用响应快的轻量级模型如llama3.2:1b加速迭代。生产环境根据任务复杂度、精度要求和成本选择经过SFT和合适量化的模型。对于客服场景Qwen2.5-7B-Instruct或Llama-3.2-3B-Instruct的量化版本通常是很好的起点。错误处理与回退机制在节点函数中对LLM调用、工具调用进行try-except。设计一个“降级节点”当主要逻辑失败时可以转向一个更简单、更稳定的回复流程。使用LangGraph Studio进行可视化调试可选LangGraph官方提供了LangGraph Studio一个Web界面可以可视化你的Graph结构并逐步执行、检查State变化。这对于调试复杂工作流非常有帮助。通过本文的讲解你应该已经掌握了使用LangGraph构建复杂AI Agent的核心流程。从定义状态、设计节点和边到集成可观测性和优化模型这构成了开发现代AI应用的一个完整闭环。记住LangGraph的本质是将复杂的AI逻辑流程化、可视化、状态化。当你下次面对需要多步决策和循环的任务时不妨先画一张流程图然后用LangGraph将它实现出来。
LangGraph实战指南：构建可观测、可循环的复杂AI Agent系统
如果你正在构建复杂的AI应用特别是那些需要多步骤决策、状态管理和循环逻辑的Agent系统那么你很可能已经感受到了传统LangChain在复杂流程编排上的局限性。当你的应用从简单的问答链发展到需要记忆、工具调用、条件分支和循环的复杂工作流时你会发现单纯用SequentialChain或LLMChain拼接代码会迅速变得难以维护和调试。这正是LangGraph要解决的核心痛点。它不是LangChain的替代品而是其“大脑”和“调度中心”。LangGraph将你的AI应用建模为一个有向图Graph节点是处理单元LLM调用、工具执行、代码运行边是控制流顺序、条件、循环。这种范式转变让开发者可以用声明式的方式构建复杂的、带状态的、可循环的AI应用就像用流程图设计业务逻辑一样直观。本文将为你提供一份从零到一的LangGraph实战指南。我们不仅会详细讲解其核心概念和安装部署更会通过一个完整的、可运行的案例带你理解如何构建一个具备“长期记忆”和“工具调用”能力的智能客服Agent。同时我们会将LangGraph与LangfuseAI应用可观测性平台、量化感知训练QAT和SFTTrainer监督微调训练器等热门技术栈结合探讨如何构建一个可观测、可优化、高性能的完整AI应用工程体系。读完本文你将能清晰理解LangGraph与LangChain的区别与定位知道何时该用它。独立完成LangGraph环境搭建并运行你的第一个Graph。掌握State设计、节点、边、条件边等核心概念并能设计自己的应用流程。学会将LangGraph应用与Langfuse集成实现完整的调用链追踪与评估。了解如何将训练好的模型特别是经过SFT和QAT优化的模型接入LangGraph应用并理解其中的工程考量。1. LangGraph为什么它是构建复杂AI Agent的“游戏规则改变者”在深入代码之前我们必须先建立一个关键认知LangGraph解决的是“复杂流程编排”问题而不是“基础功能调用”问题。想象一下这些场景一个客服Agent需要先理解用户意图然后查询知识库如果答案不完整再去调用外部API最后根据用户反馈决定是结束对话还是继续追问。一个数据分析Agent需要循环执行“提出问题 - 编写代码 - 执行代码 - 检查结果”的步骤直到得到满意的分析报告。一个游戏NPC需要根据玩家的历史对话和当前状态决定下一步是提供线索、发起战斗还是改变剧情分支。这些场景的共同点是有状态、多步骤、带循环和条件判断。用传统的LangChain链式结构来写你会陷入大量的if-else嵌套和手动状态传递中代码可读性和可维护性极差。LangGraph带来的核心改变图即程序你将应用逻辑画成一张图每个节点是一个清晰的功能单元如“调用LLM”、“执行工具”每条边代表明确的流转条件。这极大地提升了代码的可视化和可理解性。显式的状态管理所有节点共享一个中心化的State对象。你无需在函数间手动传递参数只需定义好State的结构LangGraph会自动帮你管理和更新。内置的循环与控制流通过conditional_edges条件边和将边指向END或某个节点你可以轻松实现while循环、if-else分支等复杂逻辑这是传统链难以优雅实现的。更好的调试与可观测性由于流程是结构化的你更容易追踪每个节点的输入输出也更容易与像Langfuse这样的可观测性平台集成进行全链路追踪。简单对比LangChain像“乐高积木”提供了丰富的组件Models, Prompts, Chains, Agents, Tools, Memory。适合快速搭建标准化的、线性的AI管道。LangGraph像“流程图设计器”和“调度引擎”。它使用LangChain的组件作为“积木”但负责如何将这些积木按照复杂的逻辑循环、分支组装并运行起来。它让LangChain如虎添翼。所以如果你的应用逻辑是简单的线性链LangChain足够。但一旦涉及循环、状态持久化、复杂决策LangGraph几乎是目前最优雅的解决方案。2. 核心概念解析State、Node、Edge与Graph理解LangGraph关键是理解它的几个核心抽象。我们将用一个“智能点餐助手”的类比来贯穿讲解。2.1 State应用的“记忆白板”State是一个字典或Pydantic模型它存储了应用在整个运行过程中的所有数据。你可以把它想象成餐厅服务员手里的点餐单。这张单子有固定的格式定义好的字段比如customer_query顾客需求、current_order当前已点菜品、need_confirmation是否需要确认等。每个服务环节节点都可以读取和修改这张单子上的内容。LangGraph负责在节点间传递这张最新的“点餐单”。from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator # 定义State的结构。我们使用TypedDict这是LangGraph推荐的方式。 class AgentState(TypedDict): # 存储对话历史。Annotated和add_messages是LangGraph提供的工具用于自动合并消息。 messages: Annotated[List, add_messages] # 用户的最新输入 user_input: str # Agent的思考过程或中间结果 agent_scratchpad: str # 一个标志位控制流程是否继续 should_continue: bool关键点Annotated[List, add_messages]是一个“缩减器”reducer它告诉LangGraph在更新messages字段时不是替换而是将新消息追加到列表末尾。这是实现对话记忆的关键。2.2 Node功能执行单元Node就是一个普通的Python函数或可调用对象它接收当前的State执行一些操作然后返回一个包含更新后字段的字典。 继续我们的类比Node就是餐厅里的不同角色接待员Node接收顾客原始需求(user_input)更新到点餐单(State)上。推荐员Node根据当前已点菜品(current_order)调用LLM推荐新菜并更新推荐结果到点餐单。确认员Node生成订单总结并询问顾客是否确认。def reception_node(state: AgentState) - dict: 接待节点处理用户初始输入 # 从State中获取用户输入 user_message state[“user_input”] # 将用户输入转换为AI消息格式并添加到历史中 # 注意我们返回的是要更新到State中的字段。 # LangGraph会自动将这个返回的字典与旧的State合并。 return {“messages”: [{“role”: “user”, “content”: user_message}]} def llm_agent_node(state: AgentState) - dict: AI代理节点调用LLM进行思考并可能调用工具 # 1. 从State中构建给LLM的提示词包括历史消息和工具描述 messages state[“messages”] # ... (这里会构造一个包含系统提示、历史、工具列表的messages) # 2. 调用LLM # response chat_model.invoke(messages) # 3. 解析LLM响应判断是直接回复还是调用了工具 # 如果是工具调用这里会执行工具并将结果追加到消息历史 # 我们将LLM的回复也追加到消息历史 ai_message {“role”: “assistant”, “content”: “这是AI的思考过程和回复...”} return {“messages”: [ai_message], “agent_scratchpad”: “一些中间思考...”}2.3 Edge控制流方向Edge定义了节点之间的执行顺序。分为两种普通边Linear Edge无条件地从上一个节点指向下一个节点。条件边Conditional Edge根据State中的某个条件值决定下一步走向哪个节点。这是实现分支和循环的关键。2.4 Graph将一切组装起来Graph是节点和边的容器。你创建Graph添加节点然后用边将它们连接起来。from langgraph.graph import StateGraph, END # 1. 创建一个Graph并指定它使用的State类型 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(“reception”, reception_node) workflow.add_node(“agent”, llm_agent_node) workflow.add_node(“confirm”, confirmation_node) # 3. 设置入口点第一个执行的节点 workflow.set_entry_point(“reception”) # 4. 添加普通边reception - agent workflow.add_edge(“reception”, “agent”) # 5. 添加条件边从agent节点出来后根据State中的should_continue字段决定下一步 def route_after_agent(state: AgentState): if state.get(“should_continue”, False): # 如果需要继续则返回下一个节点的名字这里我们循环回agent自身实现while效果 return “agent” else: # 如果不需要继续则结束流程 return “confirm” workflow.add_conditional_edges( “agent”, # 从哪个节点出来 route_after_agent, # 路由判断函数 {“agent”: “agent”, “confirm”: “confirm”} # 可能的目的地映射 ) # 6. 从confirm节点到结束 workflow.add_edge(“confirm”, END) # 7. 编译Graph得到一个可执行的对象 app workflow.compile()这个Graph定义了一个流程接待 - AI代理可能循环多次- 确认 - 结束。conditional_edges使得AI代理节点可以基于should_continue标志实现循环。3. 环境准备与安装LangGraph是一个Python库。我们建议在一个新的虚拟环境中安装以避免依赖冲突。3.1 创建并激活虚拟环境# 使用conda conda create -n langgraph-demo python3.10 conda activate langgraph-demo # 或使用venv python -m venv venv # Windows venv\Scripts\activate # Linux/Mac source venv/bin/activate3.2 安装核心库我们将安装LangGraph、LangChain提供LLM集成和基础工具、以及一个本地LLM的运行环境这里以使用Ollama运行开源模型为例。同时安装Langfuse用于后续的可观测性集成。# 安装 LangGraph 和 LangChain pip install langgraph langchain langchain-community # 安装 Langfuse SDK (用于追踪和评估) pip install langfuse # 安装 Ollama (用于本地运行大模型如 Llama 3.1, Qwen2.5 等) # 请根据你的操作系统从 https://ollama.com/ 下载并安装Ollama # 安装后在终端拉取一个模型例如 # ollama pull llama3.1:8b # 可选安装其他有用的工具链 pip install jupyterlab # 用于在笔记本中交互式开发 pip install python-dotenv # 用于管理环境变量3.3 验证安装创建一个简单的Python脚本test_install.py来验证基础环境。# test_install.py import langgraph import langchain print(f“LangGraph version: {langgraph.__version__}”) print(f“LangChain version: {langchain.__version__}”) # 尝试导入Langfuse try: import langfuse print(“Langfuse imported successfully.”) except ImportError as e: print(f“Failed to import Langfuse: {e}”)运行它python test_install.py如果看到版本号且没有报错说明核心环境已就绪。4. 实战案例构建一个具备长期记忆和工具调用的智能客服Agent现在我们来构建一个真实的智能客服Agent。它的功能是记忆对话历史能记住用户之前说过的话。调用工具能查询产品知识库模拟和获取当前时间。自主决策循环根据LLM的判断决定是直接回答、调用工具还是结束对话。4.1 定义State和工具首先我们定义State和两个简单的工具。# agent_demo.py from typing import TypedDict, List, Annotated from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_community.chat_models import ChatOllama # 使用Ollama from langchain.tools import tool from datetime import datetime import json # --- 1. 定义State --- class AgentState(TypedDict): # 对话消息历史使用add_messages缩减器自动追加 messages: Annotated[List, add_messages] # 一个标志由LLM或节点设置控制是否继续 should_continue: bool # --- 2. 定义工具 --- tool def search_product_knowledgebase(query: str) - str: 根据用户问题查询产品知识库。这是一个模拟函数。 # 模拟一个简单的知识库 kb { “退货政策”: “商品签收后7天内可无理由退货需保持商品完好。详情请见官网。”, “运费”: “订单满99元包邮不满99元收取10元运费。”, “保修期”: “所有电子产品提供一年保修从购买日算起。”, “客服时间”: “人工客服工作时间为每天9:00-21:00。” } # 简单关键词匹配 for key, answer in kb.items(): if key.lower() in query.lower(): return f“关于【{key}】: {answer}” return “在知识库中没有找到相关信息请尝试联系人工客服。” tool def get_current_time() - str: 获取当前的日期和时间。 now datetime.now() return now.strftime(“%Y-%m-%d %H:%M:%S”) # 将工具包装成列表供LLM使用 tools [search_product_knowledgebase, get_current_time] # --- 3. 初始化LLM --- # 确保你已经通过 ollama pull llama3.1:8b 拉取了模型 llm ChatOllama(model“llama3.1:8b”, temperature0) # 让LLM绑定工具这样它才知道自己能调用什么 llm_with_tools llm.bind_tools(tools)4.2 构建核心的Agent Node这个Node是Graph的核心它负责与LLM交互并处理工具调用。# --- 4. 定义Agent Node函数 --- def call_agent(state: AgentState): print(f“\n[DEBUG] Entering call_agent node. Message history length: {len(state[‘messages’])}”) # 1. 准备给LLM的消息。State中的messages已经包含了完整的历史。 messages state[“messages”] # 2. 调用绑定了工具的LLM response llm_with_tools.invoke(messages) # 3. 检查LLM的响应是否是工具调用 if response.tool_calls: print(f“[DEBUG] LLM decided to call tools: {response.tool_calls}”) # 初始化一个列表用于存放工具执行结果的消息 tool_messages [] for tool_call in response.tool_calls: # 根据工具名找到对应的工具函数 tool_to_call {t.name: t for t in tools}[tool_call[‘name’]] # 执行工具 tool_result tool_to_call.invoke(tool_call[‘args’]) # 将工具执行结果构造成ToolMessage并添加到列表 tool_messages.append(ToolMessage( contentstr(tool_result), tool_call_idtool_call[‘id’], nametool_call[‘name’] )) # 返回更新将LLM的响应包含工具调用请求和所有工具执行结果都追加到消息历史 # 注意这里我们返回的是要追加的消息列表LangGraph的add_messages缩减器会处理合并。 return {“messages”: [response] tool_messages, “should_continue”: True} else: # LLM直接给出了最终回复 print(f“[DEBUG] LLM gave final response: {response.content[:50]}...”) # 将AI的回复追加到历史 return {“messages”: [response], “should_continue”: False} # 可以结束对话了4.3 定义路由逻辑和构建Graph我们需要一个函数来判断在Agent Node执行后下一步应该去哪是继续循环让Agent处理工具结果还是结束# --- 5. 定义路由函数 --- def should_continue(state: AgentState) - str: “”“根据state中的should_continue标志决定下一步。 返回下一个节点的名称。”“” if state.get(“should_continue”, False): # 需要继续则返回”agent”节点名形成循环 return “agent” else: # 不需要继续则结束 return “__end__” # --- 6. 构建并编译Graph --- # 创建图 workflow StateGraph(AgentState) # 添加唯一的节点我们的核心Agent workflow.add_node(“agent”, call_agent) # 设置入口点 workflow.set_entry_point(“agent”) # 添加条件边。这是关键 # 从”agent”节点出来后根据should_continue函数的结果决定去向。 workflow.add_conditional_edges( “agent”, # 源节点 should_continue, # 路由函数 # 路由函数返回值到节点名的映射。”__end__”是LangGraph预定义的结束标识。 {“agent”: “agent”, “__end__”: END} ) # 编译成可执行的应用 app workflow.compile()4.4 运行与测试现在让我们运行这个Agent并模拟一段多轮对话。# --- 7. 运行Graph --- if __name__ “__main__”: # 初始化状态从用户的第一条消息开始 initial_state: AgentState { “messages”: [ # 系统提示词定义Agent的角色和能力 {“role”: “system”, “content”: “你是一个智能客服助手可以回答关于产品政策的问题也可以查询当前时间。请根据用户问题决定是直接回答还是调用工具。调用工具后请根据工具结果进行总结回复。用中文回答。”} ], “should_continue”: True } # 模拟用户输入 user_inputs [ “你们的退货政策是什么”, “那运费怎么算呢”, “现在几点了” ] for i, query in enumerate(user_inputs): print(f“\n{*50}”) print(f“用户第{i1}轮输入: {query}”) print(f“{*50}”) # 将用户输入转换为HumanMessage并更新到State中 # 注意我们通过app.invoke传入一个包含更新内容的字典。 # LangGraph会将这个字典与当前的State合并然后执行Graph。 inputs {“messages”: [HumanMessage(contentquery)]} # 执行Graph final_state app.invoke(inputs, initial_state) # 更新initial_state以便下一轮对话有历史记忆 initial_state final_state # 打印最后一轮AI的回复 last_message final_state[“messages”][-1] if hasattr(last_message, ‘content’): print(f“\n助手回复: {last_message.content}”) else: print(f“\n助手最后动作: {last_message}”) # 简单打印整个对话历史可选 # print(“\n当前完整对话历史:”) # for msg in final_state[“messages”]: # print(f“{msg.type}: {msg.content}”)运行这个脚本你将看到类似以下的输出清晰地展示了Agent的思考、工具调用和循环过程 用户第1轮输入: 你们的退货政策是什么 [DEBUG] Entering call_agent node. Message history length: 1 [DEBUG] LLM decided to call tools: [{‘name’: ‘search_product_knowledgebase’, ‘args’: {‘query’: ‘退货政策’}, ‘id’: ‘call_xxx’}] [DEBUG] Entering call_agent node. Message history length: 3 [DEBUG] LLM gave final response: 根据查询结果我们的退货政策是... 助手回复: 根据查询结果我们的退货政策是商品签收后7天内可无理由退货需保持商品完好。详情请见官网。这个输出表明第一轮LLM决定调用search_product_knowledgebase工具。工具执行后Graph根据should_continueTrue的条件循环再次进入call_agent节点。第二轮LLM收到了工具返回的结果并生成了最终的自然语言回复同时设置should_continueFalse。Graph结束我们得到了助手的最终回复。这就是LangGraph的核心魔力通过条件边和状态循环优雅地处理了“LLM调用工具 - 获取结果 - LLM再次处理”这个经典Agent流程。5. 集成Langfuse为你的Agent添加“眼睛”和“大脑”构建复杂的Agent只是第一步。在生产环境中你需要监控它的表现每次调用花了多少钱LLM的回复质量如何工具调用成功了吗这就是可观测性Observability的范畴而Langfuse是此领域的佼佼者。Langfuse可以自动追踪LangChain/LangGraph应用的每一次LLM调用、工具执行形成完整的溯源链Trace。你可以看到整个Graph的执行脉络每个节点的输入输出以及消耗的Token和成本。5.1 初始化Langfuse并集成到Graph中首先你需要在 Langfuse Cloud 注册并获取API密钥。# langfuse_integration.py from langfuse import Langfuse from langfuse.callback import CallbackHandler import os from dotenv import load_dotenv # 加载环境变量将你的Langfuse密钥保存在.env文件中 # LANGFUSE_SECRET_KEY“sk-lf-...” # LANGFUSE_PUBLIC_KEY“pk-lf-...” # LANGFUSE_HOST“https://cloud.langfuse.com” # 或你的自托管地址 load_dotenv() # 初始化Langfuse客户端 langfuse Langfuse( secret_keyos.getenv(“LANGFUSE_SECRET_KEY”), public_keyos.getenv(“LANGFUSE_PUBLIC_KEY”), hostos.getenv(“LANGFUSE_HOST”, “https://cloud.langfuse.com”) ) # 创建Langfuse回调处理器 langfuse_handler CallbackHandler( langfuselangfuse, # 为本次运行设置一个会话名称方便在Langfuse界面查找 session_name“my_langgraph_agent_demo” ) # --- 修改Graph的调用方式注入回调处理器 --- # 假设app是上一节编译好的Graph if __name__ “__main__”: initial_state {…} # 同上一节 inputs {“messages”: [HumanMessage(content“你们的退货政策是什么”)]} # 关键在invoke时传入callbacks参数 final_state app.invoke( inputs, initial_state, config{“callbacks”: [langfuse_handler]} # 注入Langfuse回调 ) # 确保所有追踪数据都发送到Langfuse服务器 langfuse_handler.langfuse.flush() print(“追踪数据已发送至Langfuse。”)运行这段代码后登录Langfuse Cloud的Dashboard你就能在“Traces”页面看到一个完整的追踪记录。点击进入可以看到清晰的层级结构Trace (根): 代表一次完整的app.invoke调用。Span (节点): 代表call_agent节点的执行。Generation (LLM调用): 在Span内部记录了一次具体的LLM调用包括提示词、回复、Token使用和成本。Event (工具调用): 记录工具search_product_knowledgebase的调用和结果。5.2 利用Langfuse进行提示词工程和评估Langfuse不仅是监控工具更是优化AI应用的利器。提示词版本管理你可以在Langfuse中修改系统提示词并快速比较不同提示词下Agent的表现。手动评分与反馈在界面上可以直接对某次Trace或LLM回复进行打分、添加评论这些数据可以用于后续的模型微调。数据集与评测你可以将一些测试用例导入为数据集然后让Agent批量运行自动计算关键指标如正确率、相关性形成评估报告。通过Langfuse你将Agent从“黑盒”变成了“白盒”能够系统地分析性能瓶颈、优化提示词、并收集高质量的训练数据。6. 接入优化模型SFTTrainer与量化感知训练QAT的工程实践我们的Agent核心是LLM。为了获得更好、更快、更便宜的模型我们常常需要对基础模型进行监督微调SFT和量化Quantization。这里我们探讨如何将经过这些流程优化后的模型接入LangGraph。6.1 监督微调SFT与SFTTrainerSFT是指在特定任务数据上继续训练预训练模型使其更擅长该任务。transformers库中的SFTTrainer是一个方便的工具。典型SFT流程准备数据整理成(instruction, output)的对话格式。加载基础模型如Qwen2.5-7B-Instruct。配置训练参数学习率、批次大小、训练轮数等。使用SFTTrainer训练。保存模型。训练完成后你会得到一个适配你任务的微调模型。在LangGraph中你只需要像使用任何其他ChatModel一样使用它。# 假设你有一个使用SFTTrainer微调后保存的模型目录 ./my_finetuned_model from langchain_community.chat_models import ChatOllama # 使用Ollama加载本地微调模型 # 首先你需要将HuggingFace格式的模型转换为Ollama支持的格式如GGUF并用Ollama加载。 # 例如: ollama create my-agent -f ./Modelfile (其中Modelfile指定了模型文件路径) # 然后在代码中指定这个模型名 sft_llm ChatOllama(model“my-agent”, temperature0, num_ctx4096) # 后续在构建LangGraph的Agent Node时使用这个sft_llm代替通用的llm即可。6.2 量化感知训练QAT与部署量化是将模型权重从高精度如FP32转换为低精度如INT8/INT4的过程能显著减少模型大小和推理延迟。量化感知训练QAT是在训练或微调过程中模拟量化效应让模型在训练时就适应低精度从而在真正量化后精度损失更小。对于LangGraph开发者你通常不需要自己实施QAT而是使用社区已经量化好的模型。例如从Hugging Face Model Hub下载TheBloke系列提供的GGUF量化模型。接入量化模型的步骤选择模型例如TheBloke/Llama-3.2-1B-Instruct-GGUF。下载量化文件选择适合你硬件的量化等级如Q4_K_M。使用Ollama或llama.cpp加载# 为Ollama创建Modelfile # FROM ./llama-3.2-1b-instruct.Q4_K_M.gguf # 然后 ollama create my-quantized-model -f ./Modelfile在LangGraph中调用quantized_llm ChatOllama(model“my-quantized-model”, temperature0)关键建议在LangGraph应用中优先考虑使用已经量化好的、适合你硬件的小尺寸模型。QAT是一个高级模型优化技术通常由模型提供商或算法工程师在发布模型前完成。应用开发者更应关注如何高效地集成和调用这些优化后的模型。7. 常见问题与排查指南在开发LangGraph应用时你可能会遇到以下典型问题。问题现象可能原因排查方式解决方案RuntimeError: ...或节点执行报错1. State结构定义与节点返回值不匹配。2. 工具函数参数传递错误。3. LLM响应格式不符合预期。1. 在节点函数开头打印state检查其结构。2. 检查工具tool装饰器的参数定义。3. 打印LLM的原始响应response。1. 确保节点返回的字典键名与State的TypedDict定义一致。2. 确保工具函数参数是基本类型str, int等。3. 使用try-except包裹LLM调用并检查response的属性。Graph陷入无限循环1. 条件边(conditional_edges)的路由逻辑有误始终返回循环条件。2. LLM在工具调用后仍然设置should_continueTrue。1. 在路由函数中打印state和判断逻辑。2. 检查系统提示词明确告诉LLM何时结束任务。1. 在路由函数中添加更严格的终止条件。2. 优化提示词例如“如果你已经获得了足够信息并给出了最终答案请将should_continue设为false。”Langfuse中没有Trace数据1. API密钥或Host配置错误。2. 回调处理器没有被正确注入。3. 网络问题或flush()未调用。1. 检查环境变量是否正确加载。2. 在invoke后检查langfuse_handler是否有错误属性。3. 查看Langfuse控制台或SDK日志。1. 使用langfuse.auth_check()验证连接。2. 确保config{“callbacks”: [langfuse_handler]}传入invoke。3. 调用langfuse_handler.langfuse.flush()并等待。Ollama模型加载失败或响应慢1. 模型名称错误或未下载。2. 硬件资源RAM/VRAM不足。3. Ollama服务未启动。1. 运行ollama list确认模型存在。2. 查看系统资源监控。3. 检查Ollama服务状态ollama serve。1. 使用ollama pull下载正确模型。2. 换用更小的量化模型如3B、7B的Q4量化。3. 确保Ollama在后台运行。工具调用未被LLM触发1. LLM未正确绑定工具。2. 系统提示词未明确说明工具用途。3. 工具描述不够清晰。1. 检查llm.bind_tools(tools)是否执行。2. 查看LLM收到的完整提示词。3. 检查tool装饰器中的函数文档字符串。1. 确认bind_tools调用成功。2. 在系统提示词中加入“你可以使用以下工具...”。3. 优化工具的描述使其更精准。8. 最佳实践与进阶建议掌握了基础之后遵循以下最佳实践能让你的LangGraph应用更加健壮和可维护。State设计要精简而充分精简只存储流程必需的数据。避免将整个对话历史、中间计算结果等全部塞进State这会影响序列化/反序列化效率。充分确保所有节点需要的信息都能从State中获取。常用的设计模式是将messages对话历史和scratchpad中间思考分开。节点函数保持单一职责每个节点只做一件事。例如一个节点专门调用LLM另一个节点专门处理工具执行结果。这提高了可测试性和复用性。善用条件边实现复杂逻辑不要试图在一个节点里用if-else处理所有分支。将不同的分支路径建模为不同的节点用条件边连接。这样逻辑更清晰也便于在LangGraph Studio等可视化工具中查看。集成Langfuse进行全链路可观测在开发初期就集成Langfuse。它能帮你快速定位是哪个节点、哪次LLM调用或工具执行出了问题。利用其评分和数据集功能持续优化你的Agent。模型选择与优化开发/测试环境使用响应快的轻量级模型如llama3.2:1b加速迭代。生产环境根据任务复杂度、精度要求和成本选择经过SFT和合适量化的模型。对于客服场景Qwen2.5-7B-Instruct或Llama-3.2-3B-Instruct的量化版本通常是很好的起点。错误处理与回退机制在节点函数中对LLM调用、工具调用进行try-except。设计一个“降级节点”当主要逻辑失败时可以转向一个更简单、更稳定的回复流程。使用LangGraph Studio进行可视化调试可选LangGraph官方提供了LangGraph Studio一个Web界面可以可视化你的Graph结构并逐步执行、检查State变化。这对于调试复杂工作流非常有帮助。通过本文的讲解你应该已经掌握了使用LangGraph构建复杂AI Agent的核心流程。从定义状态、设计节点和边到集成可观测性和优化模型这构成了开发现代AI应用的一个完整闭环。记住LangGraph的本质是将复杂的AI逻辑流程化、可视化、状态化。当你下次面对需要多步决策和循环的任务时不妨先画一张流程图然后用LangGraph将它实现出来。 
