在实际的大模型应用开发中我们常常面临一个核心矛盾如何将多个独立的 AI 能力如调用模型、检索知识、执行工具编排成一个稳定、可控、可观测的复杂工作流。简单的链式调用LangChain在遇到循环、分支、状态管理等复杂逻辑时会显得力不从心。这正是 LangGraph 要解决的问题它基于有向图Graph的思想将工作流中的每个步骤定义为节点Node通过边Edge来控制流程的流转特别适合构建具备长期记忆、多轮对话和复杂决策能力的智能体Agent。本文将带你从零开始全面掌握 LangGraph 的核心概念、环境搭建、基础与高级用法并结合 Langfuse 实现工作流的可观测性。我们还会探讨如何将训练好的模型通过量化感知训练QAT进行优化并使用 SFTTrainer 进行监督微调最终构建一个从开发、训练、优化到部署观测的完整实践链路。无论你是希望构建复杂的 AI 应用还是想深入理解现代 AI 工程化工具链这篇文章都将提供清晰的路径和可运行的代码。1. 理解 LangGraph为什么是图而不仅仅是链在深入代码之前必须厘清 LangGraph 的设计哲学。LangChain 的Chain对象本质是一个线性的、预定义的调用序列。当你需要根据上一步的输出动态决定下一步或者需要维护一个跨越多次调用的会话状态时单纯的链就会变得非常笨拙。LangGraph 的核心抽象是状态图State Graph。你可以将其理解为一个状态机状态State一个字典Dict保存了工作流运行过程中的所有信息例如用户输入、模型响应、检索到的文档、工具执行结果等。这个状态会在节点间传递和更新。节点Node一个函数它接收当前状态执行某些操作如调用大模型、运行计算然后返回一个更新后的状态字典。边Edge决定了流程的走向。分为条件边Conditional Edge和普通边。条件边允许你根据当前状态的某些值动态选择下一个要执行的节点这是实现分支和循环的关键。这种图结构天然支持以下复杂模式循环Cycles例如一个“思考-行动-观察”的 ReAct 智能体可以循环执行直到问题解决。分支Branching根据模型输出是“需要搜索”还是“直接回答”走向不同的处理节点。并行Parallelism理论上可以定义并行执行的节点虽然当前版本需结合异步或其他方式实现。长期记忆Long-term Memory状态可以持久化到数据库使得工作流在多次调用间能记住历史上下文。与 LangChain 的关系LangGraph 不是 LangChain 的替代品而是其补充和增强。你依然会使用 LangChain 的ChatOpenAI、PromptTemplate、Retriever等组件作为图中的“零件”但用 LangGraph 来组装和驱动这台“机器”。2. 环境准备与核心依赖安装开始构建之前需要建立一个干净的 Python 环境。强烈建议使用虚拟环境来管理依赖。2.1 创建并激活虚拟环境# 使用 conda (推荐用于管理复杂的 Python 和 CUDA 环境) conda create -n langgraph-demo python3.10 conda activate langgraph-demo # 或者使用 venv python -m venv venv # 在 Windows 上激活 venv\Scripts\activate # 在 macOS/Linux 上激活 source venv/bin/activate2.2 安装核心库我们将安装 LangChain、LangGraph、Langfuse用于追踪和观测以及 PyTorch为后续的 QAT 和 SFT 做准备。# 安装 LangChain 和 LangGraph 核心库 pip install langchain langgraph langchain-openai # 安装 Langfuse 用于可观测性 (可选但强烈推荐) pip install langfuse # 安装 PyTorch (请根据你的 CUDA 版本到官网 https://pytorch.org/ 获取对应命令) # 例如对于 CUDA 11.8 pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 # 或仅安装 CPU 版本 pip install torch torchvision torchaudio # 安装 transformers 和 datasets 库用于模型训练 pip install transformers datasets accelerate peft trl2.3 配置 API 密钥大部分组件需要外部服务的 API 密钥。建议将它们设置为环境变量而不是硬编码在代码中。# 在终端中临时设置 (Linux/macOS) export OPENAI_API_KEYyour-openai-api-key export LANGFUSE_SECRET_KEYyour-langfuse-secret-key export LANGFUSE_PUBLIC_KEYyour-langfuse-public-key export LANGFUSE_HOSThttps://cloud.langfuse.com # 或你的自托管地址 # 在 Windows PowerShell 中 $env:OPENAI_API_KEYyour-openai-api-key $env:LANGFUSE_SECRET_KEYyour-langfuse-secret-key $env:LANGFUSE_PUBLIC_KEYyour-langfuse-public-key $env:LANGFUSE_HOSThttps://cloud.langfuse.com你也可以在 Python 代码中初始化时传入但环境变量是更安全、更灵活的方式。3. 构建你的第一个 LangGraph一个简单的聊天代理让我们从一个最简单的例子开始构建一个能调用 OpenAI 模型并回复的图。这个图只有一个节点。3.1 定义状态结构首先我们需要定义工作流的状态包含哪些字段。使用TypedDict可以让代码更清晰。from typing import TypedDict, Annotated from typing_extensions import TypedDict import operator class AgentState(TypedDict): 定义图的状态结构。 # 用户输入的问题 input: str # 模型的输出 output: str # 可以扩展其他字段如聊天历史、中间步骤等 # chat_history: list3.2 创建节点函数节点函数接收状态执行任务并返回更新后的状态。这里我们创建一个调用模型的节点。from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage # 初始化模型 llm ChatOpenAI(modelgpt-3.5-turbo) def call_model(state: AgentState) - AgentState: 节点函数调用大模型生成回复。 print(f【节点 call_model】收到输入: {state[input]}) # 构建消息 messages [HumanMessage(contentstate[input])] # 调用模型 response llm.invoke(messages) # 更新状态 return {output: response.content} # 注意返回的字典会被合并到原始状态中。这里我们更新了 output 字段。3.3 构建图并编译现在我们将节点添加到图中并定义流程从__start__开始到call_model节点然后到__end__结束。from langgraph.graph import StateGraph, END # 1. 创建一个图构建器并指定状态的结构 workflow StateGraph(AgentState) # 2. 添加节点。第一个参数是节点名第二个是节点函数。 workflow.add_node(model, call_model) # 3. 设置入口点。图将从 model 节点开始执行。 workflow.set_entry_point(model) # 4. 设置出口点。执行完 model 节点后图将结束。 workflow.add_edge(model, END) # 5. 编译图得到一个可执行的对象 app workflow.compile()3.4 运行图并查看结果编译后的app有一个invoke方法传入初始状态即可运行。# 定义初始状态 initial_state {input: 请用中文介绍一下 LangGraph 是什么, output: } # 运行图 final_state app.invoke(initial_state) print(\n 最终输出 ) print(final_state[output])运行上述代码你将看到模型返回的关于 LangGraph 的介绍。你已经在运行一个最简单的 LangGraph 工作流了。4. 进阶构建具备工具调用能力的 ReAct 智能体真正的智能体需要与环境交互。我们构建一个经典的 ReActReasoning Acting智能体它可以根据问题决定是直接回答还是调用一个搜索工具这里用模拟工具代替。4.1 扩展状态与工具定义from typing import List from langchain_core.tools import tool from langchain_core.messages import AIMessage, HumanMessage, SystemMessage class ReActState(TypedDict): ReAct 智能体的状态。 messages: Annotated[List, operator.add] # 关键这是一个消息累加器 # LangGraph 的 Annotated 和 operator.add 表示新消息会追加到列表中而不是替换。 # 定义一个模拟的搜索工具 tool def search_tool(query: str) - str: 一个模拟的搜索引擎工具。输入搜索词返回模拟结果。 print(f【工具调用】搜索关键词: {query}) # 模拟返回结果 return f根据搜索 {query}得到的结果是LangGraph 是一个用于构建复杂、有状态多智能体应用的框架。4.2 创建多个节点模型、工具、路由现在我们需要三个节点agent节点分析当前对话历史和状态决定下一步是“回答”还是“使用工具”。tools节点执行被调用的工具。router节点逻辑上这不是一个独立节点而是通过条件边实现的逻辑。from langchain.agents import create_react_agent from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.agents.output_parsers import ReActSingleInputOutputParser # 准备工具列表 tools [search_tool] # 构建 ReAct 提示词模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的助手。你可以使用工具。如果你不知道答案请使用工具搜索。请用中文回答。\n\n可用工具\n{tools}), MessagesPlaceholder(variable_namemessages), # 历史消息 (user, {input}), ]) # 创建 ReAct 代理执行器它本身是一个链 agent_executor create_react_agent(llm, tools, prompt) def agent_node(state: ReActState): 节点执行代理生成下一步动作思考、行动或最终答案。 print(【节点 agent】正在思考...) # 从状态中获取最新的用户输入简化处理实际应从 messages 最后提取 # 这里我们假设最后一条消息是用户输入 user_input state[messages][-1].content if state[messages] else if not user_input: user_input state.get(input, ) # 兼容初始输入 # 调用代理执行器 response agent_executor.invoke({input: user_input, tools: tools, messages: state[messages]}) # 代理的响应可能是一个 AIMessage其中包含工具调用或最终答案 return {messages: [response[messages][-1]]} # 返回最新的 AIMessage def tool_node(state: ReActState): 节点执行工具调用。 print(【节点 tools】执行工具...) # 上一步 agent 节点的输出应该是一个 AIMessage其中包含 ToolCall last_message state[messages][-1] tool_calls last_message.tool_calls results [] for tool_call in tool_calls: tool_name tool_call[name] tool_args tool_call[args] # 找到对应的工具并执行 tool_to_use next(tool for tool in tools if tool.name tool_name) result tool_to_use.invoke(tool_args) results.append(fTool {tool_name} result: {result}) # 返回工具执行结果格式化为 ToolMessage from langchain_core.messages import ToolMessage tool_messages [ ToolMessage(contentresult, tool_call_idtool_calls[i][id]) for i, result in enumerate(results) ] return {messages: tool_messages}4.3 构建带条件边的图关键部分来了我们需要让图在agent节点之后根据其输出决定是走向tools节点继续行动还是直接结束给出最终答案。from langgraph.graph import StateGraph, END def should_continue(state: ReActState) - str: 条件路由函数判断下一步是继续使用工具还是结束。 last_message state[messages][-1] # 如果最后一条消息是 AIMessage 且包含工具调用则去执行工具 if hasattr(last_message, tool_calls) and last_message.tool_calls: return tools # 否则流程结束 else: return END # 构建图 workflow StateGraph(ReActState) workflow.add_node(agent, agent_node) workflow.add_node(tools, tool_node) # 设置入口点 workflow.set_entry_point(agent) # 添加条件边从 agent 出来根据 should_continue 函数的结果决定去向 workflow.add_conditional_edges( agent, should_continue, { tools: tools, # 如果返回 tools则前往 tools 节点 END: END # 如果返回 END则结束 } ) # 添加普通边从 tools 节点执行完后必须回到 agent 节点进行下一步思考 workflow.add_edge(tools, agent) # 编译图 react_app workflow.compile()4.4 运行并观察循环# 初始化状态包含一条用户消息 initial_state { messages: [HumanMessage(contentLangGraph 和 LangChain 有什么区别)], input: # 兼容字段 } # 运行图。注意由于可能存在循环我们需要设置最大步数或让它自然结束。 print(开始运行 ReAct 智能体...) try: # 使用 stream 可以观察每一步的输出 for event in react_app.stream(initial_state, stream_modevalues): event_messages event.get(messages, []) if event_messages: print(f\n--- 步骤输出 ---) print(event_messages[-1]) except KeyboardInterrupt: print(\n流程被中断。) # 或者直接获取最终结果 final_state react_app.invoke(initial_state) print(\n 最终对话记录 ) for msg in final_state[messages]: print(f{type(msg).__name__}: {msg.content[:100]}...)运行这段代码你会看到智能体先“思考”然后决定调用search_tool工具执行后返回结果智能体再次“思考”并生成最终答案。这就是一个完整的、具备循环能力的 ReAct 智能体工作流。5. 集成 Langfuse 实现可观测性当工作流变得复杂时追踪每一次调用、查看输入输出、分析性能和成本变得至关重要。Langfuse 是一个开源的 LLM 应用可观测性平台可以无缝集成到 LangGraph 中。5.1 初始化 Langfuse 并包装 LLMfrom langfuse import Langfuse from langfuse.callback import CallbackHandler from langchain_openai import ChatOpenAI # 初始化 Langfuse (确保环境变量已设置) langfuse Langfuse() # 创建 Langfuse 回调处理器 langfuse_handler CallbackHandler() # 使用回调处理器创建 LLM observed_llm ChatOpenAI( modelgpt-3.5-turbo, callbacks[langfuse_handler] # 关键将回调注入 LLM ) # 现在所有用 observed_llm 进行的调用都会被记录到 Langfuse5.2 在 LangGraph 中集成追踪LangGraph 的compile()方法接受checkpointer参数我们可以利用它和 Langfuse 进行更细粒度的追踪。更简单的方式是在调用invoke时传入回调。# 修改之前的 agent_node 函数使用被观测的 LLM # 首先用 observed_llm 重新创建 agent_executor agent_executor_observed create_react_agent(observed_llm, tools, prompt) def agent_node_observed(state: ReActState): 使用可观测 LLM 的代理节点。 print(【节点 agent】正在思考...) user_input state[messages][-1].content if state[messages] else if not user_input: user_input state.get(input, ) # 调用时也传入 langfuse_handler以追踪整个链的执行 response agent_executor_observed.invoke( { input: user_input, tools: tools, messages: state[messages] }, config{callbacks: [langfuse_handler]} # 传入配置 ) return {messages: [response[messages][-1]]} # 重新构建并运行图 workflow_observed StateGraph(ReActState) workflow_observed.add_node(agent, agent_node_observed) workflow_observed.add_node(tools, tool_node) # tool_node 不变 workflow_observed.set_entry_point(agent) workflow_observed.add_conditional_edges( agent, should_continue, {tools: tools, END: END} ) workflow_observed.add_edge(tools, agent) observed_app workflow_observed.compile() # 运行并追踪 print(运行带观测的智能体...) final_state observed_app.invoke( {messages: [HumanMessage(content什么是量化感知训练)]}, config{callbacks: [langfuse_handler]} # 为整个图的执行也添加回调 ) print(\n执行完成。请访问 Langfuse Cloud 控制台查看详细的追踪记录、Token 消耗和延迟。)运行后登录 Langfuse Cloud 或你的自托管实例你可以在Traces页面看到一个完整的追踪链包含了每次 LLM 调用、工具调用的详细信息这对于调试和优化至关重要。6. 模型优化量化感知训练QAT与监督微调SFT当你拥有一个定制化的、运行良好的智能体工作流后可能会考虑部署私有模型以降低成本或提升性能。这时对开源大模型进行微调SFT和优化如量化就成为了关键步骤。6.1 监督微调SFT概览与数据准备SFT 使用高质量的指令-回答对数据让预训练模型适应特定任务或风格。我们使用 Hugging Face 的transformers和trl库。首先准备数据。数据格式通常是一个 JSONL 文件每条记录包含instruction和output。from datasets import Dataset import json # 示例数据 data [ {instruction: 用友好的语气介绍 LangGraph。, output: 你好LangGraph 是一个超棒的工具它能帮你像搭积木一样构建复杂的AI工作流...}, {instruction: 解释一下量化感知训练。, output: 量化感知训练是一种模型压缩技术它在训练过程中就模拟低精度计算...}, # ... 更多数据 ] # 创建 Hugging Face Dataset dataset Dataset.from_list(data) # 通常需要将数据转换为模型需要的对话格式 def format_conversation(example): # 这里以 ChatML 格式为例 formatted_text f|im_start|user\n{example[instruction]}|im_end|\n|im_start|assistant\n{example[output]}|im_end| return {text: formatted_text} formatted_dataset dataset.map(format_conversation)6.2 使用 SFTTrainer 进行微调trl库的SFTTrainer简化了 SFT 流程。from transformers import AutoTokenizer, AutoModelForCausalLM, TrainingArguments from trl import SFTTrainer import torch # 选择模型例如 Qwen1.5-7B-Chat model_name Qwen/Qwen1.5-7B-Chat tokenizer AutoTokenizer.from_pretrained(model_name, trust_remote_codeTrue) # 设置 padding token if tokenizer.pad_token is None: tokenizer.pad_token tokenizer.eos_token model AutoModelForCausalLM.from_pretrained( model_name, torch_dtypetorch.bfloat16, # 使用 BF16 节省显存 device_mapauto, # 自动分配到 GPU trust_remote_codeTrue ) # 定义训练参数 training_args TrainingArguments( output_dir./sft_results, num_train_epochs3, per_device_train_batch_size4, gradient_accumulation_steps4, save_steps500, logging_steps50, learning_rate2e-5, fp16False, bf16True, # 使用 BF16 混合精度训练 remove_unused_columnsFalse, ) # 创建 Trainer trainer SFTTrainer( modelmodel, argstraining_args, train_datasetformatted_dataset, dataset_text_fieldtext, max_seq_length1024, tokenizertokenizer, packingTrue, # 将多个样本打包以提高效率 ) # 开始训练 print(开始监督微调...) trainer.train() trainer.save_model(./my_finetuned_model) tokenizer.save_pretrained(./my_finetuned_model) print(微调完成模型已保存。)6.3 量化感知训练QAT简介量化是将模型权重和激活从高精度如 FP32转换为低精度如 INT8的过程以大幅减少模型大小和推理延迟。量化感知训练在训练或微调过程中模拟量化误差让模型提前适应低精度环境从而在真正量化后精度损失最小。对于 LLM通常使用GPTQ训练后量化或AWQ进行权重量化。而 QAT 更常用于视觉模型或需要从头训练的场景。对于 LLM 微调一个更常见的实践是先进行 SFT然后对微调后的模型进行训练后量化。以下是使用bitsandbytes库进行加载时量化Load-in 4-bit的示例这虽然不是严格的 QAT但是一种极其流行的、高效的推理优化手段可以在微调后直接应用。from transformers import BitsAndBytesConfig import torch # 配置 4-bit 量化 bnb_config BitsAndBytesConfig( load_in_4bitTrue, bnb_4bit_quant_typenf4, # 使用 NormalFloat4 量化类型 bnb_4bit_compute_dtypetorch.bfloat16, bnb_4bit_use_double_quantTrue, # 使用双重量化进一步压缩 ) # 以量化方式加载我们微调好的模型 quantized_model AutoModelForCausalLM.from_pretrained( ./my_finetuned_model, quantization_configbnb_config, device_mapauto, trust_remote_codeTrue ) # Tokenizer 不变 quantized_tokenizer AutoTokenizer.from_pretrained(./my_finetuned_model) print(模型已以 4-bit 量化格式加载显存占用大幅降低。)现在quantized_model就可以像普通模型一样用于推理但显存占用只有原来的四分之一左右非常适合部署。7. 常见问题、排查与最佳实践7.1 LangGraph 常见问题排查问题现象可能原因检查与解决KeyError或状态字段丢失节点函数返回的字典键与状态定义不匹配或未提供初始值。1. 检查TypedDict定义的所有字段是否在初始state中都存在。2. 确保每个节点函数返回的字典包含它要更新的所有键。图陷入无限循环条件边逻辑有误或节点未正确修改导致循环条件始终满足。1. 在should_continue函数中打印日志检查路由逻辑。2. 确保工具节点执行后返回的消息能让agent节点判断出下一步是END。3. 使用app.invoke(..., config{recursion_limit”: 50})设置递归上限。工具调用不被识别AIMessage 的tool_calls格式不正确或工具定义与模型不匹配。1. 使用print(last_message)查看 AIMessage 结构。2. 确保在提示词中正确描述了工具 ({tools})。3. 使用bind_tools方法将工具绑定到 LLMllm_with_tools llm.bind_tools(tools)。Langfuse 看不到追踪记录API 密钥或主机配置错误回调未正确注入。1. 检查环境变量LANGFUSE_*是否正确设置。2. 确保在调用invoke或stream时将CallbackHandler实例传入config。3. 查看 Langfuse 控制台的项目选择是否正确。7.2 模型训练与优化注意事项数据质量高于数量SFT 需要数百到数千条高质量的对话数据。低质量数据会导致模型性能下降。仔细清洗和构造你的指令数据。小心过拟合如果训练数据很少模型可能会记住数据而不是学习泛化模式。使用验证集监控损失并考虑早停Early Stopping。资源管理全参数微调 7B 模型需要足够的 GPU 显存如 A100 40G。考虑使用参数高效微调PEFT如 LoRA可以大幅降低显存需求。from peft import LoraConfig, get_peft_model lora_config LoraConfig( r8, lora_alpha32, target_modules[q_proj, v_proj], # 针对不同模型结构需调整 lora_dropout0.1, biasnone, task_typeCAUSAL_LM, ) model get_peft_model(model, lora_config) # 然后使用此 model 进行训练量化部署bitsandbytes的 4-bit 量化非常适合推理。对于生产环境可以考虑更先进的量化方案如 GPTQ、AWQ以获得更好的精度-效率平衡并使用专门的推理引擎如 vLLM、TGI来部署。7.3 LangGraph 设计最佳实践状态设计最小化只将需要在节点间传递的数据放入状态。避免将整个会话历史都塞进去可以考虑使用外部的记忆存储如ChatMessageHistory。节点职责单一每个节点应只完成一件明确的事情如“调用模型”、“检索文档”、“执行工具”。这提高了图的可读性和可测试性。善用条件边这是实现复杂业务逻辑的核心。确保路由函数 (should_continue) 逻辑清晰并处理好所有可能的分支。集成检查点Checkpoint对于长会话或需要持久化的流程使用 LangGraph 的检查点功能可以将状态保存到数据库实现工作流的暂停、恢复和长期记忆。全面的可观测性在项目早期就集成像 Langfuse 这样的观测工具。记录每一次 LLM 调用、工具调用、耗时和成本这对于性能优化、调试和成本控制不可或缺。从 LangGraph 构建可控的工作流到用 Langfuse 观测每一个细节再到通过 SFT 和量化定制优化你的模型这条工具链代表了当前构建生产级 AI 应用的最佳实践。开始时可以从简单的线性图入手逐步引入条件逻辑、工具和记忆最终你将能够设计出应对复杂、多步骤现实任务的强大智能体系统。