在构建复杂AI应用时你是否遇到过这样的困境单个LLM调用逻辑简单但一旦涉及多步骤决策、状态循环或与外部工具交互代码就迅速变得难以维护LangChain的链式调用在处理复杂工作流时显得力不从心而LangGraph的出现正是为了解决这一痛点。本文将为你带来LangGraph从零到一的完整实战指南不仅涵盖其核心概念与安装部署还会深入讲解如何结合Langfuse进行可观测性监控并拓展至模型微调SFT与量化QAT等高级主题为你构建生产级AI智能体应用提供一站式解决方案。1. LangGraph核心概念与架构解析在深入代码之前我们必须理解LangGraph要解决的根本问题以及它的设计哲学。这将帮助你在后续开发中做出更合理的设计决策。1.1 什么是LangGraph与LangChain有何区别LangGraph是一个基于图Graph的编程框架专门用于构建有状态、多参与者的AI应用。它的核心思想是将应用逻辑建模为一个有向图图中的节点Node代表一个执行单元如调用LLM、执行工具、条件判断边Edge代表节点之间的流转路径。这与LangChain有本质区别LangChain侧重于“链”Chain是一种线性的、预定义顺序的执行流程。它适合串联多个LLM调用或工具但难以处理循环、分支和复杂状态管理。LangGraph侧重于“图”Graph支持任意复杂的拓扑结构包括循环、条件分支、并行执行。它内置了状态管理机制非常适合构建需要长期记忆、反复决策的智能体Agent或工作流。简单来说LangChain是“流水线”而LangGraph是“调度中心”。对于聊天机器人、游戏NPC、自动化工作流等需要根据上下文动态决定下一步行动的场景LangGraph是更自然的选择。1.2 LangGraph的核心组件理解以下几个核心组件是使用LangGraph的基础State状态这是一个贯穿整个图执行过程的共享数据容器。通常是一个Pydantic模型或TypedDict定义了工作流中需要传递和修改的所有数据。例如在智能体应用中状态可能包含用户消息、对话历史、已执行工具的结果等。Node节点图中的一个执行单元。它是一个函数接收当前State作为输入并返回一个更新后的State。节点可以执行任何操作调用LLM、运行计算、调用API等。Edge边定义节点之间的流转逻辑。分为两种普通边Edge无条件地从上一个节点指向下一个节点。条件边Conditional Edge根据State中的某个值动态决定下一个节点是哪个。这是实现分支逻辑的关键。Graph图将节点和边组装起来的容器。LangGraph提供了StateGraph类来帮助构建图。CompiledGraph编译图将定义好的Graph编译成一个可执行对象。编译过程会进行验证并可以添加中断Interruption、检查点Checkpointing等高级功能。这种基于状态和图的模型使得LangGraph能够优雅地处理智能体中常见的“思考-行动-观察”循环。2. 环境准备与安装我们将在一个干净的环境中进行确保依赖的兼容性。本文示例基于Python 3.9这是大多数AI框架的推荐版本。2.1 创建虚拟环境与安装核心库首先创建一个新的项目目录并设置虚拟环境这是管理Python项目依赖的最佳实践。# 创建项目目录并进入 mkdir langgraph-tutorial cd langgraph-tutorial # 创建并激活虚拟环境以conda为例也可使用venv conda create -n langgraph-env python3.10 -y conda activate langgraph-env接下来安装LangGraph及其常用配套库。我们一并安装LangChain因为LangGraph通常与LangChain的组件如LLM、工具协同工作。# 安装LangGraph和LangChain核心包 pip install langgraph langchain langchain-core # 安装一个LLM提供商这里以OpenAI为例需自有API Key pip install langchain-openai # 安装Langfuse用于可观测性可选但强烈推荐 pip install langfuse # 安装开发常用工具 pip install jupyter python-dotenv2.2 配置环境变量为了安全地管理API密钥我们使用.env文件。在项目根目录创建.env文件。# 创建.env文件 touch .env将你的API密钥填入.env文件。切记不要将此文件提交到版本控制系统如Git。# .env 文件内容 OPENAI_API_KEYsk-your-openai-api-key-here LANGFUSE_SECRET_KEYsk-lf-your-langfuse-secret-key-here LANGFUSE_PUBLIC_KEYpk-lf-your-langfuse-public-key-here LANGFUSE_HOSThttps://cloud.langfuse.com # 或你的自托管地址在Python代码中使用dotenv加载这些变量。# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的所有变量 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) LANGFUSE_SECRET_KEY os.getenv(LANGFUSE_SECRET_KEY) LANGFUSE_PUBLIC_KEY os.getenv(LANGFUSE_PUBLIC_KEY) LANGFUSE_HOST os.getenv(LANGFUSE_HOST)3. LangGraph基础实战构建第一个智能体工作流让我们通过一个经典的“ReAct智能体”示例来上手。这个智能体能够根据用户问题决定是调用工具如计算器、搜索引擎还是直接回答。3.1 定义状态State状态是工作流的“记忆”。我们使用TypedDict来定义。# agent_state.py from typing import TypedDict, List, Annotated import operator from typing_extensions import TypedDict # 使用注解来简化状态更新LangGraph的推荐方式 class AgentState(TypedDict): # 用户输入的问题 input: str # 智能体生成的思考步骤 thoughts: Annotated[List[str], operator.add] # 从工具调用获得的结果 tool_results: Annotated[List[str], operator.add] # 最终给用户的回答 answer: str关键解释Annotated[List[str], operator.add]这是一个神奇的注解。它告诉LangGraph当多个节点修改thoughts或tool_results字段时应该使用operator.add即列表的extend操作来合并更新而不是覆盖。这是实现状态累积的关键。3.2 创建工具Tools智能体需要工具来与世界交互。我们定义两个简单的工具。# tools.py from langchain.tools import tool from datetime import datetime tool def get_current_time(placeholder: str None) - str: 获取当前的日期和时间。当用户询问时间、日期、今天星期几时使用此工具。 now datetime.now() return now.strftime(%Y-%m-%d %H:%M:%S %A) tool def calculator(expression: str) - str: 计算一个数学表达式。只支持基本运算, -, *, /, (), **。 # 警告在生产环境中直接eval是危险的这里仅作演示。 # 实际应用应使用安全计算库如ast.literal_eval或专用计算引擎。 try: # 限制可用的命名空间增加一点安全性 allowed_names {} result eval(expression, {__builtins__: None}, allowed_names) return f计算结果: {expression} {result} except Exception as e: return f计算错误: {e} # 将工具包装成列表供智能体使用 tools [get_current_time, calculator]3.3 构建LangGraph现在是核心部分将节点和边组装成图。# graph_builder.py from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate from agent_state import AgentState from tools import tools import config # 1. 初始化LLM llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, api_keyconfig.OPENAI_API_KEY) # 2. 创建ReAct智能体这是LangChain的Agent prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的助手可以回答问题和使用工具。请一步一步思考。), (human, {input}), (placeholder, {agent_scratchpad}) # 代理会在此处填充其思考和行动 ]) agent create_react_agent(llm, tools, prompt) # 3. 创建代理执行器 agent_executor AgentExecutor(agentagent, toolstools, verboseFalse, handle_parsing_errorsTrue) # 4. 定义图节点 def call_agent(state: AgentState): 调用智能体决定下一步是执行工具还是结束。 # 调用LangChain的AgentExecutor result agent_executor.invoke({input: state[input]}) # 更新状态 state[thoughts].append(result.get(intermediate_steps, [])) state[answer] result[output] return state def process_tool_result(state: AgentState): 处理工具调用的结果本例中已集成在call_agent中此处为展示节点定义 # 在实际更复杂的图中这里可能包含结果后处理逻辑 print(f工具执行结果: {state[tool_results]}) return state # 5. 构建图 graph_builder StateGraph(AgentState) # 添加节点 graph_builder.add_node(agent, call_agent) graph_builder.add_node(process_result, process_tool_result) # 设置入口点 graph_builder.set_entry_point(agent) # 添加边从agent节点出来后直接结束因为ReAct循环在agent内部已完成 graph_builder.add_edge(agent, END) # 如果需要更复杂的分支可以添加条件边 # graph_builder.add_conditional_edges( # agent, # lambda state: needs_tool if some_condition else final_answer, # {needs_tool: process_result, final_answer: END} # ) # 编译图 graph graph_builder.compile() # 6. 运行图 if __name__ __main__: # 初始化状态 initial_state: AgentState { input: 现在几点了如果现在是下午2点请计算 (15 7) * 3 等于多少, thoughts: [], tool_results: [], answer: } # 执行图 final_state graph.invoke(initial_state) print(\n 最终回答 ) print(final_state[answer]) print(\n 思考过程 ) for thought in final_state[thoughts]: print(f- {thought})运行上述代码你将看到智能体首先调用get_current_time工具获取时间然后可能直接计算或再次调用calculator工具最后生成包含时间和计算结果的回答。这个流程完全由LLM驱动LangGraph负责状态流转。4. 集成Langfuse实现可观测性当智能体工作流变得复杂调试和优化就变得困难。Langfuse是一个开源的LLM应用可观测性平台可以跟踪每次调用、记录输入输出、分析延迟和成本。4.1 初始化Langfuse并包装LLM首先确保你已在 Langfuse Cloud 创建项目并获取了密钥。# langfuse_integration.py from langfuse import Langfuse from langfuse.callback import CallbackHandler from langchain_openai import ChatOpenAI import config # 1. 初始化Langfuse客户端 langfuse Langfuse( secret_keyconfig.LANGFUSE_SECRET_KEY, public_keyconfig.LANGFUSE_PUBLIC_KEY, hostconfig.LANGFUSE_HOST ) # 2. 创建Langfuse回调处理器 langfuse_handler CallbackHandler() # 3. 创建集成了Langfuse的LLM observed_llm ChatOpenAI( modelgpt-3.5-turbo, temperature0, api_keyconfig.OPENAI_API_KEY, callbacks[langfuse_handler] # 关键注入回调 ) # 4. 在原有的graph_builder.py中将llm替换为observed_llm # from langfuse_integration import observed_llm as llm4.2 在LangGraph中集成追踪为了追踪整个图的执行而不仅仅是LLM调用我们需要使用Langfuse的trace上下文管理器。# graph_with_tracing.py from langfuse.decorators import observe, langfuse_context # 假设graph是之前编译好的图 from graph_builder import graph import asyncio observe() # 使用装饰器自动创建Trace def run_agent_with_trace(user_input: str): 运行智能体并自动记录Trace到Langfuse initial_state { input: user_input, thoughts: [], tool_results: [], answer: } # 可以手动设置Trace的元数据 langfuse_context.update_current_trace( nameReAct Agent Workflow, user_iddemo_user_123, metadata{model: gpt-3.5-turbo, environment: testing} ) result graph.invoke(initial_state) return result[answer] if __name__ __main__: answer run_agent_with_trace(请计算圆周率小数点后5位是多少) print(f回答: {answer}) # 注意Langfuse是异步上报数据可能需要稍等片刻才能在Dashboard看到运行后登录Langfuse Dashboard你将看到一个清晰的Trace记录包含了整个图执行的时序、每个节点的耗时、LLM调用的输入/输出和Token使用量。这对于性能分析、调试和成本核算至关重要。5. 进阶主题与模型微调SFT和量化QAT结合LangGraph负责工作流编排而模型的“能力”则来源于底层LLM。为了获得更专业、更高效或更私有的模型我们常常需要对基础模型进行微调SFT和量化QAT。5.1 使用SFTTrainer进行监督微调假设我们想微调一个模型使其更擅长执行我们定义的工具调用逻辑。我们可以使用Hugging Face的transformers库中的SFTTrainer。环境准备pip install transformers datasets accelerate peft trl torch微调示例脚本# sft_finetune.py from datasets import load_dataset from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments from trl import SFTTrainer import torch # 1. 加载模型和分词器以小型模型为例如Qwen1.5-1.8B model_name Qwen/Qwen1.5-1.8B model AutoModelForCausalLM.from_pretrained( model_name, torch_dtypetorch.bfloat16, device_mapauto ) tokenizer AutoTokenizer.from_pretrained(model_name) tokenizer.pad_token tokenizer.eos_token # 设置填充token # 2. 准备数据集这里需要你准备工具调用格式的对话数据 # 示例数据格式每条数据是一个包含“工具调用”和“正确响应”的对话 def format_tool_data(example): # 假设example包含query和tool_call_sequence text fHuman: {example[query]}\nAssistant: {example[tool_call_sequence]} return {text: text} # 加载或创建你的数据集 # dataset load_dataset(your_dataset_path) # 这里用一个虚拟示例 from datasets import Dataset data [ {query: 现在时间是多少, tool_call_sequence: 我需要调用get_current_time工具。当前时间是2023-10-27 14:30:00 Friday。}, {query: 计算35*2, tool_call_sequence: 我需要调用calculator工具。计算过程35*213。} ] dataset Dataset.from_list(data) dataset dataset.map(format_tool_data) # 3. 配置训练参数 training_args TrainingArguments( output_dir./sft_finetuned_model, num_train_epochs3, per_device_train_batch_size4, gradient_accumulation_steps4, warmup_steps100, logging_steps10, save_steps500, evaluation_strategyno, learning_rate2e-5, fp16True, # 根据硬件调整 push_to_hubFalse, # 设为True可上传到Hugging Face Hub ) # 4. 创建SFTTrainer trainer SFTTrainer( modelmodel, tokenizertokenizer, argstraining_args, train_datasetdataset, dataset_text_fieldtext, max_seq_length512, ) # 5. 开始训练 trainer.train() trainer.save_model(./sft_finetuned_model_final) print(微调完成模型已保存。)微调后的模型可以替换掉原来LangGraph中使用的通用LLM如ChatOpenAI使你的智能体在特定工具调用任务上表现更精准、更符合你的业务逻辑。5.2 量化感知训练QAT简介与集成量化是将模型权重从高精度如FP32转换为低精度如INT8的过程以大幅减少模型大小和推理延迟。量化感知训练QAT在训练过程中模拟量化效果让模型提前适应精度损失从而在真正量化后保持更高的精度。为什么在边缘设备如Jetson上需要QAT直接在训练好的模型上做后训练量化PTQ可能会导致精度显著下降。QAT通过在训练前向传播中插入“伪量化”节点让模型在训练阶段就学习到量化噪声从而获得更鲁棒的量化模型。使用PyTorch进行QAT的基本步骤# qat_example.py (概念性代码) import torch import torch.nn as nn from torch.quantization import QuantStub, DeQuantStub, prepare_qat, convert class SimpleModelForQAT(nn.Module): def __init__(self): super().__init__() self.quant QuantStub() # 量化入口 self.linear nn.Linear(10, 5) self.dequant DeQuantStub() # 反量化出口 def forward(self, x): x self.quant(x) x self.linear(x) x self.dequant(x) return x # 1. 创建模型并设置为训练模式 model SimpleModelForQAT() model.train() # 2. 准备QAT模型 model.qconfig torch.quantization.get_default_qat_qconfig(fbgemm) # 后端配置 model_prepared prepare_qat(model) # 3. 进行训练在你的训练循环中使用model_prepared # ... 你的训练代码 ... # 4. 训练完成后转换为真正的量化模型 model_quantized convert(model_prepared.eval()) # 5. 保存和加载量化模型 torch.jit.save(torch.jit.script(model_quantized), quantized_model.pt)将量化模型集成到LangGraph 量化后的模型通常通过torch.jit或onnxruntime加载。你可以创建一个自定义的LangChain LLM封装类来调用它。# custom_quantized_llm.py from langchain_core.language_models.llms import BaseLLM from langchain_core.callbacks import CallbackManagerForLLMRun from langchain_core.outputs import LLMResult, Generation from typing import Any, List, Optional, Dict import torch class QuantizedLLMWrapper(BaseLLM): 封装量化PyTorch模型的LangChain LLM接口 model_path: str tokenizer: Any # Hugging Face Tokenizer device: str cpu def __init__(self, model_path: str, tokenizer, **kwargs): super().__init__(**kwargs) self.model_path model_path self.tokenizer tokenizer self.model torch.jit.load(model_path, map_locationself.device) self.model.eval() def _call( self, prompt: str, stop: Optional[List[str]] None, run_manager: Optional[CallbackManagerForLLMRun] None, **kwargs: Any, ) - str: # 编码输入 inputs self.tokenizer(prompt, return_tensorspt) with torch.no_grad(): # 注意实际推理逻辑需根据你的模型结构调整 outputs self.model(inputs.input_ids) generated_ids outputs.argmax(dim-1) response self.tokenizer.decode(generated_ids[0], skip_special_tokensTrue) return response property def _llm_type(self) - str: return quantized_llm # 在LangGraph中使用 # from custom_quantized_llm import QuantizedLLMWrapper # llm QuantizedLLMWrapper(model_path./quantized_model.pt, tokenizertokenizer)通过结合SFT和QAT你可以在LangGraph中部署一个既专精于你的任务又能在资源受限环境下高效运行的定制化AI智能体。6. 常见问题与排查思路在实际开发中你可能会遇到以下典型问题。问题现象可能原因排查步骤与解决方案Graph编译或执行时报State字段错误1. State的TypedDict定义与节点中实际访问的键名不匹配。2. 使用了未在State中定义的键。3.Annotated注解的更新操作符使用错误。1. 仔细检查TypedDict的字段名和类型。2. 确保所有节点只读取和写入已定义的State字段。3. 对于列表累积使用Annotated[List, operator.add]对于字符串追加可使用operator.concat。智能体陷入无限循环或重复调用工具1. 图中存在未正确终止的循环。2. LLM的提示词未能引导其做出结束决策。3. 条件边的判断逻辑有误。1. 使用langgraph的checkpointer或设置max_iterations参数限制循环次数。2. 在系统提示词中明确告知模型“在获得最终答案后请输出最终答案并结束”。3. 打印条件边判断函数的输入State检查逻辑。Langfuse Dashboard上看不到Trace数据1. API密钥或主机地址配置错误。2. 回调处理器未正确附加到LLM或函数。3. 数据上报是异步的有延迟。4. 代码执行环境无网络连接。1. 检查.env文件和环境变量。2. 确保callbacks[langfuse_handler]已传入LLM初始化。3. 等待10-30秒刷新页面或调用langfuse.flush()强制同步。4. 检查网络对于生产环境考虑使用Langfuse SDK的离线模式或队列。集成自定义模型如SFT/QAT后响应慢或错误1. 模型未正确加载到指定设备如GPU。2. 自定义LLM封装类的_call方法逻辑错误。3. 量化模型与推理框架不兼容。1. 检查torch.cuda.is_available()和model.to(device)。2. 单独测试你的自定义LLM类确保输入输出格式符合LangChain预期。3. 确保量化模型如TorchScript, ONNX的导出和加载版本一致。多线程/异步环境下State混乱1. Graph实例被多个线程共享State未隔离。2. 节点函数中修改了可变对象如字典、列表产生了副作用。1. 为每个会话或请求创建新的Graph实例或使用graph.with_config(configurable{...})。2. 遵循函数式编程思想节点函数应返回新的State字典而不是修改传入的State。LangGraph的Annotated机制已处理了部分问题。7. 最佳实践与工程建议将LangGraph应用于生产环境需要关注以下几点状态设计精简而明确State应只包含工作流必需的数据。避免将整个会话历史或大量中间结果塞入State这会影响序列化效率和内存使用。考虑将大数据存储在外部缓存如Redis只在State中保留索引或关键标识。节点功能单一且可测试每个节点应只负责一个明确的职责。这使得单元测试变得容易。确保节点函数是纯函数或接近纯函数便于调试和复现问题。利用检查点Checkpointing实现持久化对于长时间运行的工作流如需要人工审核的流程使用Checkpointer将图的状态持久化到数据库。这样即使服务重启工作流也能从断点恢复。为图配置超时和重试在编译图时使用interrupt_before或interrupt_after配置超时中断。对于可能失败的节点如调用外部API在节点逻辑内实现重试机制或使用tenacity等重试库。实现全面的可观测性不仅集成Langfuse还要在关键节点添加业务日志和指标如执行时间、成功率。这有助于监控系统健康度和分析性能瓶颈。版本化你的图定义当工作流逻辑更新时图的定义也会变化。对定义图的代码进行版本控制并考虑如何平滑迁移正在运行中的旧版本图实例的状态。安全第一谨慎处理用户输入和工具调用。对传入LLM的提示词进行清理防止提示注入。像我们之前示例中的calculator工具在生产中必须用安全的数学表达式解析库如ast.literal_eval或numexpr替代eval。LangGraph为构建复杂、有状态的AI应用提供了一个强大而优雅的范式。从简单的ReAct智能体到包含人工审核、外部系统集成的多步骤业务流程它都能胜任。结合Langfuse的可观测性、SFT的领域定制和QAT的推理优化你可以打造出高效、可靠且易于维护的下一代AI应用。建议从本文的示例出发逐步尝试设计自己的状态模型和图结构在实践中深入掌握这一强大工具。 LangGraph实战：构建可观测、可微调的AI智能体工作流
在构建复杂AI应用时你是否遇到过这样的困境单个LLM调用逻辑简单但一旦涉及多步骤决策、状态循环或与外部工具交互代码就迅速变得难以维护LangChain的链式调用在处理复杂工作流时显得力不从心而LangGraph的出现正是为了解决这一痛点。本文将为你带来LangGraph从零到一的完整实战指南不仅涵盖其核心概念与安装部署还会深入讲解如何结合Langfuse进行可观测性监控并拓展至模型微调SFT与量化QAT等高级主题为你构建生产级AI智能体应用提供一站式解决方案。1. LangGraph核心概念与架构解析在深入代码之前我们必须理解LangGraph要解决的根本问题以及它的设计哲学。这将帮助你在后续开发中做出更合理的设计决策。1.1 什么是LangGraph与LangChain有何区别LangGraph是一个基于图Graph的编程框架专门用于构建有状态、多参与者的AI应用。它的核心思想是将应用逻辑建模为一个有向图图中的节点Node代表一个执行单元如调用LLM、执行工具、条件判断边Edge代表节点之间的流转路径。这与LangChain有本质区别LangChain侧重于“链”Chain是一种线性的、预定义顺序的执行流程。它适合串联多个LLM调用或工具但难以处理循环、分支和复杂状态管理。LangGraph侧重于“图”Graph支持任意复杂的拓扑结构包括循环、条件分支、并行执行。它内置了状态管理机制非常适合构建需要长期记忆、反复决策的智能体Agent或工作流。简单来说LangChain是“流水线”而LangGraph是“调度中心”。对于聊天机器人、游戏NPC、自动化工作流等需要根据上下文动态决定下一步行动的场景LangGraph是更自然的选择。1.2 LangGraph的核心组件理解以下几个核心组件是使用LangGraph的基础State状态这是一个贯穿整个图执行过程的共享数据容器。通常是一个Pydantic模型或TypedDict定义了工作流中需要传递和修改的所有数据。例如在智能体应用中状态可能包含用户消息、对话历史、已执行工具的结果等。Node节点图中的一个执行单元。它是一个函数接收当前State作为输入并返回一个更新后的State。节点可以执行任何操作调用LLM、运行计算、调用API等。Edge边定义节点之间的流转逻辑。分为两种普通边Edge无条件地从上一个节点指向下一个节点。条件边Conditional Edge根据State中的某个值动态决定下一个节点是哪个。这是实现分支逻辑的关键。Graph图将节点和边组装起来的容器。LangGraph提供了StateGraph类来帮助构建图。CompiledGraph编译图将定义好的Graph编译成一个可执行对象。编译过程会进行验证并可以添加中断Interruption、检查点Checkpointing等高级功能。这种基于状态和图的模型使得LangGraph能够优雅地处理智能体中常见的“思考-行动-观察”循环。2. 环境准备与安装我们将在一个干净的环境中进行确保依赖的兼容性。本文示例基于Python 3.9这是大多数AI框架的推荐版本。2.1 创建虚拟环境与安装核心库首先创建一个新的项目目录并设置虚拟环境这是管理Python项目依赖的最佳实践。# 创建项目目录并进入 mkdir langgraph-tutorial cd langgraph-tutorial # 创建并激活虚拟环境以conda为例也可使用venv conda create -n langgraph-env python3.10 -y conda activate langgraph-env接下来安装LangGraph及其常用配套库。我们一并安装LangChain因为LangGraph通常与LangChain的组件如LLM、工具协同工作。# 安装LangGraph和LangChain核心包 pip install langgraph langchain langchain-core # 安装一个LLM提供商这里以OpenAI为例需自有API Key pip install langchain-openai # 安装Langfuse用于可观测性可选但强烈推荐 pip install langfuse # 安装开发常用工具 pip install jupyter python-dotenv2.2 配置环境变量为了安全地管理API密钥我们使用.env文件。在项目根目录创建.env文件。# 创建.env文件 touch .env将你的API密钥填入.env文件。切记不要将此文件提交到版本控制系统如Git。# .env 文件内容 OPENAI_API_KEYsk-your-openai-api-key-here LANGFUSE_SECRET_KEYsk-lf-your-langfuse-secret-key-here LANGFUSE_PUBLIC_KEYpk-lf-your-langfuse-public-key-here LANGFUSE_HOSThttps://cloud.langfuse.com # 或你的自托管地址在Python代码中使用dotenv加载这些变量。# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的所有变量 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) LANGFUSE_SECRET_KEY os.getenv(LANGFUSE_SECRET_KEY) LANGFUSE_PUBLIC_KEY os.getenv(LANGFUSE_PUBLIC_KEY) LANGFUSE_HOST os.getenv(LANGFUSE_HOST)3. LangGraph基础实战构建第一个智能体工作流让我们通过一个经典的“ReAct智能体”示例来上手。这个智能体能够根据用户问题决定是调用工具如计算器、搜索引擎还是直接回答。3.1 定义状态State状态是工作流的“记忆”。我们使用TypedDict来定义。# agent_state.py from typing import TypedDict, List, Annotated import operator from typing_extensions import TypedDict # 使用注解来简化状态更新LangGraph的推荐方式 class AgentState(TypedDict): # 用户输入的问题 input: str # 智能体生成的思考步骤 thoughts: Annotated[List[str], operator.add] # 从工具调用获得的结果 tool_results: Annotated[List[str], operator.add] # 最终给用户的回答 answer: str关键解释Annotated[List[str], operator.add]这是一个神奇的注解。它告诉LangGraph当多个节点修改thoughts或tool_results字段时应该使用operator.add即列表的extend操作来合并更新而不是覆盖。这是实现状态累积的关键。3.2 创建工具Tools智能体需要工具来与世界交互。我们定义两个简单的工具。# tools.py from langchain.tools import tool from datetime import datetime tool def get_current_time(placeholder: str None) - str: 获取当前的日期和时间。当用户询问时间、日期、今天星期几时使用此工具。 now datetime.now() return now.strftime(%Y-%m-%d %H:%M:%S %A) tool def calculator(expression: str) - str: 计算一个数学表达式。只支持基本运算, -, *, /, (), **。 # 警告在生产环境中直接eval是危险的这里仅作演示。 # 实际应用应使用安全计算库如ast.literal_eval或专用计算引擎。 try: # 限制可用的命名空间增加一点安全性 allowed_names {} result eval(expression, {__builtins__: None}, allowed_names) return f计算结果: {expression} {result} except Exception as e: return f计算错误: {e} # 将工具包装成列表供智能体使用 tools [get_current_time, calculator]3.3 构建LangGraph现在是核心部分将节点和边组装成图。# graph_builder.py from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate from agent_state import AgentState from tools import tools import config # 1. 初始化LLM llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, api_keyconfig.OPENAI_API_KEY) # 2. 创建ReAct智能体这是LangChain的Agent prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的助手可以回答问题和使用工具。请一步一步思考。), (human, {input}), (placeholder, {agent_scratchpad}) # 代理会在此处填充其思考和行动 ]) agent create_react_agent(llm, tools, prompt) # 3. 创建代理执行器 agent_executor AgentExecutor(agentagent, toolstools, verboseFalse, handle_parsing_errorsTrue) # 4. 定义图节点 def call_agent(state: AgentState): 调用智能体决定下一步是执行工具还是结束。 # 调用LangChain的AgentExecutor result agent_executor.invoke({input: state[input]}) # 更新状态 state[thoughts].append(result.get(intermediate_steps, [])) state[answer] result[output] return state def process_tool_result(state: AgentState): 处理工具调用的结果本例中已集成在call_agent中此处为展示节点定义 # 在实际更复杂的图中这里可能包含结果后处理逻辑 print(f工具执行结果: {state[tool_results]}) return state # 5. 构建图 graph_builder StateGraph(AgentState) # 添加节点 graph_builder.add_node(agent, call_agent) graph_builder.add_node(process_result, process_tool_result) # 设置入口点 graph_builder.set_entry_point(agent) # 添加边从agent节点出来后直接结束因为ReAct循环在agent内部已完成 graph_builder.add_edge(agent, END) # 如果需要更复杂的分支可以添加条件边 # graph_builder.add_conditional_edges( # agent, # lambda state: needs_tool if some_condition else final_answer, # {needs_tool: process_result, final_answer: END} # ) # 编译图 graph graph_builder.compile() # 6. 运行图 if __name__ __main__: # 初始化状态 initial_state: AgentState { input: 现在几点了如果现在是下午2点请计算 (15 7) * 3 等于多少, thoughts: [], tool_results: [], answer: } # 执行图 final_state graph.invoke(initial_state) print(\n 最终回答 ) print(final_state[answer]) print(\n 思考过程 ) for thought in final_state[thoughts]: print(f- {thought})运行上述代码你将看到智能体首先调用get_current_time工具获取时间然后可能直接计算或再次调用calculator工具最后生成包含时间和计算结果的回答。这个流程完全由LLM驱动LangGraph负责状态流转。4. 集成Langfuse实现可观测性当智能体工作流变得复杂调试和优化就变得困难。Langfuse是一个开源的LLM应用可观测性平台可以跟踪每次调用、记录输入输出、分析延迟和成本。4.1 初始化Langfuse并包装LLM首先确保你已在 Langfuse Cloud 创建项目并获取了密钥。# langfuse_integration.py from langfuse import Langfuse from langfuse.callback import CallbackHandler from langchain_openai import ChatOpenAI import config # 1. 初始化Langfuse客户端 langfuse Langfuse( secret_keyconfig.LANGFUSE_SECRET_KEY, public_keyconfig.LANGFUSE_PUBLIC_KEY, hostconfig.LANGFUSE_HOST ) # 2. 创建Langfuse回调处理器 langfuse_handler CallbackHandler() # 3. 创建集成了Langfuse的LLM observed_llm ChatOpenAI( modelgpt-3.5-turbo, temperature0, api_keyconfig.OPENAI_API_KEY, callbacks[langfuse_handler] # 关键注入回调 ) # 4. 在原有的graph_builder.py中将llm替换为observed_llm # from langfuse_integration import observed_llm as llm4.2 在LangGraph中集成追踪为了追踪整个图的执行而不仅仅是LLM调用我们需要使用Langfuse的trace上下文管理器。# graph_with_tracing.py from langfuse.decorators import observe, langfuse_context # 假设graph是之前编译好的图 from graph_builder import graph import asyncio observe() # 使用装饰器自动创建Trace def run_agent_with_trace(user_input: str): 运行智能体并自动记录Trace到Langfuse initial_state { input: user_input, thoughts: [], tool_results: [], answer: } # 可以手动设置Trace的元数据 langfuse_context.update_current_trace( nameReAct Agent Workflow, user_iddemo_user_123, metadata{model: gpt-3.5-turbo, environment: testing} ) result graph.invoke(initial_state) return result[answer] if __name__ __main__: answer run_agent_with_trace(请计算圆周率小数点后5位是多少) print(f回答: {answer}) # 注意Langfuse是异步上报数据可能需要稍等片刻才能在Dashboard看到运行后登录Langfuse Dashboard你将看到一个清晰的Trace记录包含了整个图执行的时序、每个节点的耗时、LLM调用的输入/输出和Token使用量。这对于性能分析、调试和成本核算至关重要。5. 进阶主题与模型微调SFT和量化QAT结合LangGraph负责工作流编排而模型的“能力”则来源于底层LLM。为了获得更专业、更高效或更私有的模型我们常常需要对基础模型进行微调SFT和量化QAT。5.1 使用SFTTrainer进行监督微调假设我们想微调一个模型使其更擅长执行我们定义的工具调用逻辑。我们可以使用Hugging Face的transformers库中的SFTTrainer。环境准备pip install transformers datasets accelerate peft trl torch微调示例脚本# sft_finetune.py from datasets import load_dataset from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments from trl import SFTTrainer import torch # 1. 加载模型和分词器以小型模型为例如Qwen1.5-1.8B model_name Qwen/Qwen1.5-1.8B model AutoModelForCausalLM.from_pretrained( model_name, torch_dtypetorch.bfloat16, device_mapauto ) tokenizer AutoTokenizer.from_pretrained(model_name) tokenizer.pad_token tokenizer.eos_token # 设置填充token # 2. 准备数据集这里需要你准备工具调用格式的对话数据 # 示例数据格式每条数据是一个包含“工具调用”和“正确响应”的对话 def format_tool_data(example): # 假设example包含query和tool_call_sequence text fHuman: {example[query]}\nAssistant: {example[tool_call_sequence]} return {text: text} # 加载或创建你的数据集 # dataset load_dataset(your_dataset_path) # 这里用一个虚拟示例 from datasets import Dataset data [ {query: 现在时间是多少, tool_call_sequence: 我需要调用get_current_time工具。当前时间是2023-10-27 14:30:00 Friday。}, {query: 计算35*2, tool_call_sequence: 我需要调用calculator工具。计算过程35*213。} ] dataset Dataset.from_list(data) dataset dataset.map(format_tool_data) # 3. 配置训练参数 training_args TrainingArguments( output_dir./sft_finetuned_model, num_train_epochs3, per_device_train_batch_size4, gradient_accumulation_steps4, warmup_steps100, logging_steps10, save_steps500, evaluation_strategyno, learning_rate2e-5, fp16True, # 根据硬件调整 push_to_hubFalse, # 设为True可上传到Hugging Face Hub ) # 4. 创建SFTTrainer trainer SFTTrainer( modelmodel, tokenizertokenizer, argstraining_args, train_datasetdataset, dataset_text_fieldtext, max_seq_length512, ) # 5. 开始训练 trainer.train() trainer.save_model(./sft_finetuned_model_final) print(微调完成模型已保存。)微调后的模型可以替换掉原来LangGraph中使用的通用LLM如ChatOpenAI使你的智能体在特定工具调用任务上表现更精准、更符合你的业务逻辑。5.2 量化感知训练QAT简介与集成量化是将模型权重从高精度如FP32转换为低精度如INT8的过程以大幅减少模型大小和推理延迟。量化感知训练QAT在训练过程中模拟量化效果让模型提前适应精度损失从而在真正量化后保持更高的精度。为什么在边缘设备如Jetson上需要QAT直接在训练好的模型上做后训练量化PTQ可能会导致精度显著下降。QAT通过在训练前向传播中插入“伪量化”节点让模型在训练阶段就学习到量化噪声从而获得更鲁棒的量化模型。使用PyTorch进行QAT的基本步骤# qat_example.py (概念性代码) import torch import torch.nn as nn from torch.quantization import QuantStub, DeQuantStub, prepare_qat, convert class SimpleModelForQAT(nn.Module): def __init__(self): super().__init__() self.quant QuantStub() # 量化入口 self.linear nn.Linear(10, 5) self.dequant DeQuantStub() # 反量化出口 def forward(self, x): x self.quant(x) x self.linear(x) x self.dequant(x) return x # 1. 创建模型并设置为训练模式 model SimpleModelForQAT() model.train() # 2. 准备QAT模型 model.qconfig torch.quantization.get_default_qat_qconfig(fbgemm) # 后端配置 model_prepared prepare_qat(model) # 3. 进行训练在你的训练循环中使用model_prepared # ... 你的训练代码 ... # 4. 训练完成后转换为真正的量化模型 model_quantized convert(model_prepared.eval()) # 5. 保存和加载量化模型 torch.jit.save(torch.jit.script(model_quantized), quantized_model.pt)将量化模型集成到LangGraph 量化后的模型通常通过torch.jit或onnxruntime加载。你可以创建一个自定义的LangChain LLM封装类来调用它。# custom_quantized_llm.py from langchain_core.language_models.llms import BaseLLM from langchain_core.callbacks import CallbackManagerForLLMRun from langchain_core.outputs import LLMResult, Generation from typing import Any, List, Optional, Dict import torch class QuantizedLLMWrapper(BaseLLM): 封装量化PyTorch模型的LangChain LLM接口 model_path: str tokenizer: Any # Hugging Face Tokenizer device: str cpu def __init__(self, model_path: str, tokenizer, **kwargs): super().__init__(**kwargs) self.model_path model_path self.tokenizer tokenizer self.model torch.jit.load(model_path, map_locationself.device) self.model.eval() def _call( self, prompt: str, stop: Optional[List[str]] None, run_manager: Optional[CallbackManagerForLLMRun] None, **kwargs: Any, ) - str: # 编码输入 inputs self.tokenizer(prompt, return_tensorspt) with torch.no_grad(): # 注意实际推理逻辑需根据你的模型结构调整 outputs self.model(inputs.input_ids) generated_ids outputs.argmax(dim-1) response self.tokenizer.decode(generated_ids[0], skip_special_tokensTrue) return response property def _llm_type(self) - str: return quantized_llm # 在LangGraph中使用 # from custom_quantized_llm import QuantizedLLMWrapper # llm QuantizedLLMWrapper(model_path./quantized_model.pt, tokenizertokenizer)通过结合SFT和QAT你可以在LangGraph中部署一个既专精于你的任务又能在资源受限环境下高效运行的定制化AI智能体。6. 常见问题与排查思路在实际开发中你可能会遇到以下典型问题。问题现象可能原因排查步骤与解决方案Graph编译或执行时报State字段错误1. State的TypedDict定义与节点中实际访问的键名不匹配。2. 使用了未在State中定义的键。3.Annotated注解的更新操作符使用错误。1. 仔细检查TypedDict的字段名和类型。2. 确保所有节点只读取和写入已定义的State字段。3. 对于列表累积使用Annotated[List, operator.add]对于字符串追加可使用operator.concat。智能体陷入无限循环或重复调用工具1. 图中存在未正确终止的循环。2. LLM的提示词未能引导其做出结束决策。3. 条件边的判断逻辑有误。1. 使用langgraph的checkpointer或设置max_iterations参数限制循环次数。2. 在系统提示词中明确告知模型“在获得最终答案后请输出最终答案并结束”。3. 打印条件边判断函数的输入State检查逻辑。Langfuse Dashboard上看不到Trace数据1. API密钥或主机地址配置错误。2. 回调处理器未正确附加到LLM或函数。3. 数据上报是异步的有延迟。4. 代码执行环境无网络连接。1. 检查.env文件和环境变量。2. 确保callbacks[langfuse_handler]已传入LLM初始化。3. 等待10-30秒刷新页面或调用langfuse.flush()强制同步。4. 检查网络对于生产环境考虑使用Langfuse SDK的离线模式或队列。集成自定义模型如SFT/QAT后响应慢或错误1. 模型未正确加载到指定设备如GPU。2. 自定义LLM封装类的_call方法逻辑错误。3. 量化模型与推理框架不兼容。1. 检查torch.cuda.is_available()和model.to(device)。2. 单独测试你的自定义LLM类确保输入输出格式符合LangChain预期。3. 确保量化模型如TorchScript, ONNX的导出和加载版本一致。多线程/异步环境下State混乱1. Graph实例被多个线程共享State未隔离。2. 节点函数中修改了可变对象如字典、列表产生了副作用。1. 为每个会话或请求创建新的Graph实例或使用graph.with_config(configurable{...})。2. 遵循函数式编程思想节点函数应返回新的State字典而不是修改传入的State。LangGraph的Annotated机制已处理了部分问题。7. 最佳实践与工程建议将LangGraph应用于生产环境需要关注以下几点状态设计精简而明确State应只包含工作流必需的数据。避免将整个会话历史或大量中间结果塞入State这会影响序列化效率和内存使用。考虑将大数据存储在外部缓存如Redis只在State中保留索引或关键标识。节点功能单一且可测试每个节点应只负责一个明确的职责。这使得单元测试变得容易。确保节点函数是纯函数或接近纯函数便于调试和复现问题。利用检查点Checkpointing实现持久化对于长时间运行的工作流如需要人工审核的流程使用Checkpointer将图的状态持久化到数据库。这样即使服务重启工作流也能从断点恢复。为图配置超时和重试在编译图时使用interrupt_before或interrupt_after配置超时中断。对于可能失败的节点如调用外部API在节点逻辑内实现重试机制或使用tenacity等重试库。实现全面的可观测性不仅集成Langfuse还要在关键节点添加业务日志和指标如执行时间、成功率。这有助于监控系统健康度和分析性能瓶颈。版本化你的图定义当工作流逻辑更新时图的定义也会变化。对定义图的代码进行版本控制并考虑如何平滑迁移正在运行中的旧版本图实例的状态。安全第一谨慎处理用户输入和工具调用。对传入LLM的提示词进行清理防止提示注入。像我们之前示例中的calculator工具在生产中必须用安全的数学表达式解析库如ast.literal_eval或numexpr替代eval。LangGraph为构建复杂、有状态的AI应用提供了一个强大而优雅的范式。从简单的ReAct智能体到包含人工审核、外部系统集成的多步骤业务流程它都能胜任。结合Langfuse的可观测性、SFT的领域定制和QAT的推理优化你可以打造出高效、可靠且易于维护的下一代AI应用。建议从本文的示例出发逐步尝试设计自己的状态模型和图结构在实践中深入掌握这一强大工具。