1. 项目概述从链式编排到图式编排的Agent进化如果你在过去一年里接触过基于大语言模型的应用开发那么“LangChain”这个名字大概率不会陌生。它几乎成了快速搭建一个具备检索增强生成RAG或简单工具调用能力应用的“脚手架”代名词。但当你真正试图用它构建一个逻辑复杂、状态持久、且需要根据中间结果动态调整执行路径的智能体Agent时可能会感到有些力不从心。那种感觉就像是用一堆乐高积木搭一个静态模型很顺手但要让它动起来还得自己手动去拧发条、调齿轮过程变得异常繁琐。这正是“LangGraph”出现的背景也是我们这次工程实践要解决的核心问题。简单来说LangChain提供的是“链”Chain——一种线性的、预定义顺序的执行流。而LangGraph提供的是“图”Graph——一种可以包含循环、分支和状态管理的执行流。从链到图的跃迁意味着我们的智能体从“按剧本走”升级到了“能自己看地图找路”。我最近在一个需要处理多轮、多步骤、且后续步骤严重依赖前序步骤结果的项目中就深刻体会到了这种升级的必要性。项目要求智能体能够理解用户一个模糊的初始请求比如“帮我分析一下上季度的销售数据并给出下季度的预测建议”然后自主决定先去数据库拉取数据接着清洗数据再选择分析模型生成报告最后甚至能根据初步报告发现的问题发起新一轮的细化查询。这种带有“循环”和“状态记忆”的任务用传统的LangChain Agent架构来堆砌代码会迅速变得难以维护和调试。于是我把技术栈切换到了LangGraph。这不仅仅是一个库的更换更是一种构建复杂AI应用范式的转变。本次实践我将分享如何利用LangGraph的核心概念——StateGraph和create_agent来构建一个真正可控、可观测、可扩展的智能体。我们会从为什么需要Graph开始一步步拆解其核心组件并最终落地一个具备循环、工具调用和状态管理能力的实用Agent。无论你是正在LangChain中挣扎于复杂逻辑的开发者还是刚刚对AI Agent产生兴趣的探索者相信这篇从实战中总结的指南都能给你带来直接的参考价值。2. 核心理念解析为什么是状态图StateGraph在深入代码之前我们必须先理解LangGraph赖以立足的核心理念。这能帮助我们在设计阶段就做出更合理的决策而不是盲目地堆砌节点和边。2.1 链Chain的局限与图Graph的优势LangChain的Chain很好用它将一系列对大模型的调用、工具的使用、数据的处理封装成一个可执行的单元。对于“输入-处理-输出”这种单一流程的任务Chain是最高效的抽象。例如一个典型的RAG链查询转换 - 向量检索 - 上下文组装 - 答案生成。这条路径是确定的一次执行单向流动。然而智能体的本质是“自主决策”它需要根据环境反馈工具执行结果、用户输入、中间状态来决定下一步做什么。这就引入了几个Chain难以优雅处理的概念循环Looping智能体可能需要反复尝试同一个工具如调整参数后重新查询或者在一个“思考-行动-观察”的循环中持续运行直到满足终止条件。分支Branching根据工具执行的结果或模型的分析智能体可能需要选择不同的执行路径。比如如果数据库查询返回为空则转向调用网络搜索工具如果数据量很大则先调用总结工具否则直接分析。状态State的持久与共享在整个执行过程中会产生大量的中间信息用户原始问题、历史对话、工具执行结果、模型生成的思考过程等。这些信息需要在不同的节点Node之间传递和更新并且其结构可能是动态变化的。用Chain来模拟这些行为通常意味着你要在单个Chain里写大量的if-else逻辑或者将多个Chain嵌套调用手动传递状态字典。代码很快就会变成“面条代码”可读性和可维护性急剧下降。更重要的是这种结构的执行流程是“黑盒”的你很难清晰地观测到智能体在每一步做了什么决策状态是如何演变的。2.2 StateGraph将智能体流程可视化与结构化LangGraph的StateGraph正是为了解决上述问题而生。它将整个智能体的工作流定义为一个有向图。这个图由两部分组成节点Nodes代表一个原子操作单元。它可以是一个调用大模型生成回复的函数也可以是一个执行工具的函数甚至可以是一个简单的逻辑判断函数。每个节点接收当前的“状态”进行处理然后返回更新后的“状态”。边Edges定义了节点之间的执行流向。分为两种普通边Edges无条件地从上一个节点指向下一个节点。条件边Conditional Edges根据当前状态中的某个条件通常由一个大模型或一个判断函数决定来决定下一步走向哪个节点。这实现了分支逻辑。StateGraph的强大之处在于它显式地将工作流、状态和决策逻辑分离开了。工作流就是这张图的结构一目了然。你可以像画流程图一样设计你的智能体。状态是一个定义好的Pydantic模型或TypedDict规定了在整个流程中流转的数据结构。每个节点都读写这个统一的状态对象保证了数据传递的一致性和类型安全。决策逻辑被封装在条件边和节点函数内部。节点负责“做什么”条件边负责“接下来去哪”。这种架构带来了几个工程上的巨大优势可观测性你可以轻松地记录下每个节点的输入输出甚至将整个图的结构和运行轨迹可视化出来调试效率倍增。可维护性修改流程只需增删节点或调整边而不会影响其他部分的代码。状态结构的变更也集中在状态定义处。可组合性复杂的图可以由多个子图组合而成支持模块化开发。2.3 与其他Agent框架的横向对比在决定使用LangGraph之前我也评估过其他流行的Agent框架这里分享一些直观的对比感受帮助你在技术选型时更有依据。AutoGen微软AutoGen的核心概念是“代理”Agent和“对话”它更侧重于多智能体之间的协作对话。对于构建一个单一但内部逻辑复杂的智能体AutoGen的抽象层次有时会显得过重你需要定义多个代理并通过它们之间的聊天来推进任务。而LangGraph更侧重于对单个智能体内部执行流程的精细控制感觉更像是在编排一个“工作流引擎”。Semantic Kernel微软SK提供了强大的插件Plugins规划和函数调用能力。它与LangGraph在理念上有相似之处都强调规划和编排。但LangGraph与LangChain生态的无缝集成是其巨大优势特别是如果你已经在使用LangChain的众多工具、向量库和文档加载器那么LangGraph几乎是平滑升级的最优路径。SK则更偏向于与.NET生态深度集成。原生OpenAI Assistant APIOpenAI提供了官方的Assistant API内置了代码解释器、文件搜索等功能。它的优点是开箱即用、无需管理状态和线程。但对于需要深度定制工作流、集成内部工具如公司数据库API、或对执行逻辑有严格把控的场景它就显得不够灵活。LangGraph让你拥有100%的控制权。选择建议如果你的需求是快速搭建一个标准化的、以对话和文件处理为主的助手OpenAI Assistant API可能更省心。如果你的智能体需要复杂的业务逻辑、循环判断、与大量异构系统集成并且你希望拥有极高的可观测性和可维护性那么LangGraph是目前更强大的工程化选择。3. 工程实践构建一个数据分析智能体理论说得再多不如一行代码。接下来我将以一个“数据分析智能体”为例展示从零开始用LangGraph构建一个具备多步执行和循环能力的智能体的全过程。这个智能体的目标是接收用户关于数据比如销售数据的自然语言提问自主决定调用查询、清洗、分析、可视化等工具并可以基于初步分析结果进行追问或深入挖掘。3.1 环境准备与状态定义首先安装必要的库。除了langgraph我们通常还需要langchain的核心包以及对应大模型的SDK这里以OpenAI为例。pip install langgraph langchain-openai一切的核心始于状态定义。我们需要仔细规划智能体在整个生命周期中需要记住什么。这里我们定义一个相对全面的状态from typing import TypedDict, List, Annotated import operator from langgraph.graph.message import add_messages class AgentState(TypedDict): # 对话消息历史LangGraph内置了add_messages缩减器来处理列表追加 messages: Annotated[List, add_messages] # 用户的原始目标 user_objective: str # 智能体计划执行的步骤列表 plan: List[str] # 已执行步骤的结果记录 executed_steps: List[dict] # 从数据库或工具获取的原始数据 raw_data: str # 经过清洗或处理后的数据 processed_data: str # 最终的分析结果或答案 final_answer: str # 一个标志位用于控制循环是否继续 should_continue: bool关键点解析Annotated和add_messages这是LangGraph的一个精妙设计。add_messages是一个“缩减器”reducer它定义了当多个节点并行修改messages字段时如何合并这些修改这里是追加到列表。这为未来支持更复杂的图结构如分支合并打下了基础。状态字段设计字段并非越多越好。plan和executed_steps对于实现可解释的智能体至关重要它们记录了智能体的“思考过程”。should_continue是控制循环的关键开关。类型提示使用TypedDict和明确的类型提示不仅能利用IDE的自动补全和错误检查也让图的结构和节点函数的输入输出更加清晰。3.2 工具Tools的封装与集成智能体的能力边界由其工具集决定。我们将封装几个模拟工具。在实际项目中这里替换成真实的数据库查询、API调用或内部函数即可。from langchain.tools import tool from typing import Optional tool def query_database(query: str) - str: 根据自然语言描述查询数据库返回数据。 # 模拟这里应替换为真实的数据库连接和查询逻辑 print(f[工具调用] query_database: {query}) # 返回模拟数据 return 日期,产品,销售额,区域\n2024-01-01,产品A,10000,华北\n2024-01-01,产品B,15000,华东\n2024-01-02,产品A,12000,华北 tool def clean_data(raw_data: str, instructions: Optional[str] None) - str: 清洗数据例如处理缺失值、格式标准化。 print(f[工具调用] clean_data: {instructions}) # 模拟清洗过程 cleaned raw_data.replace(产品, Product) # 简单示例 return fCleaned Data:\n{cleaned} tool def analyze_data(data: str, analysis_type: str) - str: 对数据进行分析如统计汇总、趋势判断。 print(f[工具调用] analyze_data: {analysis_type}) lines data.strip().split(\n)[1:] # 跳过标题 total_sales sum(float(line.split(,)[2]) for line in lines if line) return f分析类型『{analysis_type}』完成。总销售额为{total_sales} tool def generate_report(analysis_result: str, user_objective: str) - str: 根据分析结果和用户目标生成一份总结报告。 print(f[工具调用] generate_report for objective: {user_objective}) return f# 数据分析报告\n**目标**{user_objective}\n\n**核心发现**{analysis_result}\n\n建议重点关注高销售额区域和产品。工具封装心得清晰的文档字符串Docstring大模型依赖这些描述来决定何时以及如何使用工具。描述应简洁、准确说明输入输出。工具粒度工具应该足够“原子”。一个工具最好只做一件事。例如不要把“查询并清洗数据”做成一个工具而应拆分成query_database和clean_data。这给了智能体更大的灵活性。错误处理在实际工具中务必加入健壮的错误处理try-catch并返回结构化的错误信息以便智能体能理解并采取补救措施如重试或换一种方式。3.3 构建节点Nodes智能体的功能单元节点是图的执行单元。每个节点都是一个函数它接收当前AgentState返回一个更新后的AgentState或一个包含更新字段的字典。我们首先构建最核心的节点——规划节点和执行节点。from langchain_openai import ChatOpenAI from langchain.agents import create_openai_functions_agent from langchain.agents import AgentExecutor from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder # 初始化大模型 llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0) # 1. 规划节点将用户目标分解为步骤 def plan_node(state: AgentState) - AgentState: 根据用户目标生成一个初步的执行计划。 print(\n--- 进入规划节点 ---) user_goal state[user_objective] planner_prompt ChatPromptTemplate.from_messages([ (system, 你是一个资深数据分析师。请将用户的数据分析目标分解为具体的、可执行的步骤。步骤应清晰且后续步骤可以基于前序步骤的结果进行。输出格式为纯文本每行一个步骤。), (human, 用户目标{goal}) ]) planner_chain planner_prompt | llm plan_response planner_chain.invoke({goal: user_goal}) # 解析模型输出假设每行是一个步骤 plan_steps [step.strip() for step in plan_response.content.split(\n) if step.strip()] print(f生成的计划{plan_steps}) return {plan: plan_steps, should_continue: True} # 2. 执行节点利用LangChain Agent执行单个步骤 # 首先为执行节点创建一个专用的、功能更聚焦的Agent tools [query_database, clean_data, analyze_data, generate_report] executor_prompt ChatPromptTemplate.from_messages([ (system, 你是一个专注的执行者。根据当前步骤和已有信息选择合适的工具并执行。只做当前步骤要求的事情不要提前做后续步骤。如果步骤是分析性的且没有对应工具请用你的知识直接回答。), MessagesPlaceholder(variable_namemessages), (human, 当前需要执行的步骤是{current_step}。已获取的数据或信息{context}), ]) executor_agent create_openai_functions_agent(llm, tools, executor_prompt) agent_executor AgentExecutor(agentexecutor_agent, toolstools, verboseFalse) def execute_step_node(state: AgentState) - AgentState: 执行计划中的下一个步骤。 print(f\n--- 进入执行节点 ---) if not state[plan]: return {should_continue: False, final_answer: 计划已全部执行完毕。} current_step state[plan].pop(0) # 取出并移除第一个步骤 context f用户目标{state[user_objective]}\n已执行步骤结果{state[executed_steps][-2:] if state[executed_steps] else 无}\n原始数据{state.get(raw_data, 无)}\n处理后的数据{state.get(processed_data, 无)} # 准备给Agent的消息 from langchain_core.messages import HumanMessage human_message HumanMessage(contentf步骤{current_step}\n上下文{context}) # 调用Agent执行 response agent_executor.invoke({ messages: [human_message], current_step: current_step, context: context }) output response[output] # 记录执行结果 executed_step_record {step: current_step, result: output, tool_calls: response.get(intermediate_steps, [])} new_executed_steps state[executed_steps] [executed_step_record] # 关键根据工具调用结果更新状态中的特定字段这是一个简化逻辑实际应根据工具类型精细更新 # 例如如果调用了query_database则更新raw_data # 这里我们做一个简单的模式匹配 if any(query_database in str(tc[0]) for tc in response.get(intermediate_steps, [])): state[raw_data] output elif any(clean_data in str(tc[0]) for tc in response.get(intermediate_steps, [])): state[processed_data] output elif any(analyze_data in str(tc[0]) for tc in response.get(intermediate_steps, [])): state[final_answer] output # 临时存放分析结果 print(f执行步骤{current_step}) print(f执行结果{output[:200]}...) # 打印前200字符 return { executed_steps: new_executed_steps, plan: state[plan], # 更新后的计划已移除当前步骤 raw_data: state.get(raw_data, ), processed_data: state.get(processed_data, ), final_answer: state.get(final_answer, ), should_continue: len(state[plan]) 0 # 如果计划不为空则继续 }节点设计注意事项单一职责每个节点只做一件事。plan_node只负责生成计划execute_step_node只负责执行一个步骤。这使调试和测试变得简单。状态更新节点返回的字典中只需包含需要修改的状态字段。LangGraph会自动将其与旧状态合并。未返回的字段保持不变。执行节点的复杂性execute_step_node内部嵌套了一个完整的LangChain Agent。这体现了LangGraph的包容性你可以将现有的、复杂的LangChain Chain或Agent作为一个“超级节点”嵌入到图中。这极大地保护了已有投资。3.4 编排图Graph连接节点与定义流程现在我们将节点组装起来定义它们之间的执行顺序和条件。from langgraph.graph import StateGraph, END # 创建图构建器 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(planner, plan_node) workflow.add_node(executor, execute_step_node) # 可以添加更多节点例如一个专门的“判断”节点来决定是否深入分析 workflow.add_node(refine_question, lambda state: state) # 这里先占位后面实现 # 设置入口点 workflow.set_entry_point(planner) # 添加边从规划节点无条件指向执行节点 workflow.add_edge(planner, executor) # 关键定义从执行节点出发的条件边 def should_continue(state: AgentState) - str: 根据状态决定下一步是继续执行还是结束或是进入细化追问。 # 情况1计划已空结束 if not state[plan]: return end # 情况2可以在这里添加更复杂的逻辑例如根据final_answer的质量决定是否循环 # 比如如果分析结果太笼统可以进入“refine_question”节点生成一个新问题加入计划 # 这里我们先实现简单的“继续执行下一个步骤” return continue_execution # 添加条件边 workflow.add_conditional_edges( executor, should_continue, { continue_execution: executor, # 继续执行下一个步骤循环 end: END, # 结束 # refine: refine_question, // 可以指向细化节点 } ) # 编译图得到可执行对象 app workflow.compile()图编排的核心set_entry_point定义了智能体启动时的第一个节点。add_edge定义了固定的执行顺序。add_conditional_edges这是实现智能“循环”和“分支”的魔法所在。should_continue函数是一个“路由函数”它检查当前状态并返回下一个要执行的节点的名字。在这里我们实现了一个简单的循环只要计划列表不为空执行完一个步骤后就路由回executor节点自己处理下一个步骤直到计划清空才路由到END结束。compile()将图定义编译成一个可调用的、高性能的执行器。编译过程会进行优化和验证。3.5 运行与调试观察智能体的思考过程现在让我们运行这个智能体并观察其内部状态的变化。# 定义初始状态 initial_state { messages: [], # 初始对话为空 user_objective: 分析一下产品A和产品B在今年1月初的销售表现并告诉我哪个更好, plan: [], executed_steps: [], raw_data: , processed_data: , final_answer: , should_continue: True, } # 运行图 print( 开始执行智能体 ) final_state None # 我们可以使用流的模式来逐步观察这里为了清晰直接运行到底 for step in app.stream(initial_state, stream_modevalues): node_name list(step.keys())[0] node_state step[node_name] print(f\n 节点 [{node_name}] 执行完毕当前状态片段:) print(f 剩余计划: {node_state.get(plan, [])}) print(f 已执行步骤数: {len(node_state.get(executed_steps, []))}) if node_state.get(final_answer): print(f 最新分析结果: {node_state[final_answer][:100]}...) print(- * 50) final_state node_state print(\n 执行结束 ) print(最终答案, final_state.get(final_answer, 无)) print(\n完整执行记录) for i, record in enumerate(final_state.get(executed_steps, [])): print(f{i1}. {record[step]}) print(f 结果: {record[result][:150]}...)运行上述代码你将在控制台看到一个清晰的执行轨迹。智能体会先进入planner节点生成一个类似[“1. 查询产品A和产品B在2024-01-01至2024-01-07的销售数据”, “2. 清洗和整理查询到的数据”, “3. 对比分析产品A和产品B的销售额”, “4. 生成分析报告”]的计划。然后进入executor节点依次执行每个步骤并在步骤间循环。状态如plan,executed_steps,raw_data在整个过程中被有序地更新和传递。4. 高级模式与生产级考量上面的例子展示了LangGraph的核心工作模式。但要将其用于生产环境还需要考虑更多。4.1 使用create_agent函数快速构建对于许多标准场景LangGraph提供了更上层的create_react_agent基于ReAct范式等函数。但为了极致控制我们也可以围绕create_agent思路进行定制。本质上create_agent帮你封装了“模型调用 - 工具选择 - 执行”这个循环。在我们的架构中execute_step_node节点内的AgentExecutor就扮演了这个角色。你可以选择使用LangGraph预置的Agent节点也可以像我们这样自己封装以获得更高的灵活性。4.2 实现长期记忆与持久化我们的状态在单次运行中是存在的但一旦程序结束就消失了。生产环境中的智能体往往需要“记住”跨会话的信息。LangGraph通过检查点Checkpointing机制来支持这一点。from langgraph.checkpoint.sqlite import SqliteSaver import tempfile # 创建一个临时的SQLite数据库来存储检查点 with tempfile.NamedTemporaryFile(suffix.db, deleteFalse) as f: db_path f.name memory SqliteSaver.from_conn_string(fsqlite:///{db_path}) # 在编译图时传入检查点存储器 app_persistent workflow.compile(checkpointermemory) # 现在运行图时需要传入一个config其中包含线程ID用于标识不同的会话 config {configurable: {thread_id: user_session_123}} initial_state {...} # 同前 # 第一次运行 for step in app_persistent.stream(initial_state, configconfig, stream_modevalues): ... # 即使程序重启只要使用相同的thread_id就可以从上次中断的地方恢复状态 # 例如用户问了后续问题 follow_up_state {messages: [HumanMessage(content那么产品A在华北区的具体趋势呢)], ...} # 再次stream时LangGraph会自动加载之前保存的状态在其基础上继续运行 for step in app_persistent.stream(follow_up_state, configconfig, stream_modevalues): ...检查点的价值这对于构建多轮对话助手、长流程任务如订票、客服至关重要。它使得智能体有了“会话记忆”。4.3 错误处理与鲁棒性增强生产环境必须考虑失败。节点级错误处理在每个节点函数内部使用try...except捕获异常并更新状态例如设置一个error字段然后通过条件边路由到一个“错误处理”节点。图级超时设置在app.invoke()或流式调用时设置超时防止智能体陷入死循环。工具调用降级当某个工具调用失败时节点可以尝试备用工具或生成一个友好的错误信息放入状态让后续节点决定如何应对。4.4 可观测性与监控LangGraph的图结构天生适合可观测性。日志记录在每个节点的开始和结束记录状态快照。你可以轻松地将这些日志发送到像LangSmith这样的平台。可视化使用workflow.get_graph().draw_mermaid_png()可以生成图的Mermaid图表直观展示你的智能体架构。追踪Tracing集成LangChain的Callbacks可以详细追踪每一次大模型调用和工具调用的输入、输出、延迟和token消耗这对于性能优化和成本控制必不可少。5. 常见陷阱与性能优化实战记录在项目落地过程中我踩过不少坑也总结了一些优化经验。5.1 状态设计过载与更新冲突问题初期设计状态时恨不得把所有信息都塞进去导致状态对象非常庞大。更糟糕的是多个节点可能同时修改同一个字段虽然我们的例子是顺序的但复杂图可能并行引发更新冲突或意外覆盖。解决方案状态最小化只存储真正需要在节点间流转的信息。中间计算结果如果只在下个节点用一次可以考虑不存入全局状态而是通过节点返回值直接传递。善用缩减器Reducer对于像messages这样的列表使用add_messages这类内置缩减器。对于自定义的复杂合并逻辑如合并两个字典你需要定义自己的缩减器函数。清晰的更新策略在节点函数文档中明确说明它会修改哪些状态字段。对于可能被多个节点修改的字段设计好更新顺序或使用锁机制在复杂并行图中。5.2 条件边路由函数的复杂性失控问题should_continue这类路由函数最初可能很简单但随着业务逻辑复杂它可能变成一个充斥着if-elif-else的庞然大物难以维护和测试。解决方案路由节点化不要把所有判断逻辑都塞进一个路由函数。可以创建一个专门的“决策节点”Decision Node。这个节点接收状态运行一个专门的LLM Chain或规则引擎输出下一个要执行的节点名称。然后将这个决策节点插入图中用普通边连接它。这样决策逻辑就变成了一个可独立测试和优化的模块。分层判断先进行简单的布尔判断如plan是否为空如果为真再进入更复杂的LLM判断如“分析结果是否足够好”。5.3 工具调用中的幻觉与无效循环问题LLM有时会“幻觉”出不存在工具的参数或固执地反复调用一个失败的工具导致死循环。解决方案严格的工具验证在工具函数内部对输入参数进行严格的类型和值验证并返回明确的错误信息。循环中断机制在状态中设置计数器如query_retry_count。在路由函数中检查如果同一工具失败超过N次则强制路由到“人工接管”或“失败处理”节点。丰富的系统提示词在规划节点和执行节点的系统提示词中明确约束工具的用途和调用条件。例如“不要尝试调用query_database工具来执行数据清洗操作”。5.4 性能瓶颈分析与优化问题当图变得复杂、节点增多时执行速度可能变慢。优化方向节点并行化如果两个节点间没有数据依赖关系可以使用workflow.add_edge()的并行模式或者设计子图来实现并行执行。LangGraph支持在单个步骤中并行调用多个节点。缓存对于昂贵的操作如查询某些静态数据库可以在节点中引入缓存机制将结果暂存在状态或外部缓存中避免重复计算。LLM调用优化这是最大的开销。考虑使用更快的模型在非关键路径上使用gpt-3.5-turbo。精简提示词去除不必要的上下文。流式响应对于需要长时间运行的智能体使用流式输出可以提升用户体验感觉更快。编译优化compile()方法会进行一些静态优化。确保你的图结构尽可能清晰避免动态添加节点等运行时操作。从LangChain到LangGraph的迁移不是一个简单的库替换而是一次构建AI智能体心智模型的升级。它迫使你从“线性脚本”的思维转向“状态机”和“工作流”的思维。这种思维下构建的智能体其行为更可预测、更易调试、也更强大。
)
