LangChain 和 LangGraph 这套技术栈最近几乎成了企业级 Agent 开发的默认组合。很多人从 LangChain V1 一路迁移到 LangGraph核心原因只有一个真实业务里的 Agent 不是一个对话框而是一套带记忆、带工具、带分支判断、带状态管理的完整工作流。如果你也在做 AI Agent 开发或者正纠结 LangChain 和 LangGraph 到底怎么分工这篇文章会给你一条能照着落地的路径从最小 Agent 搭起到 TextToSQL、工作流编排、可观测部署最后把上线后最容易踩的坑一起说清楚。1. 先想清楚LangChain 和 LangGraph 在企业级 Agent 里到底各管哪一段1.1 V1 的便利和局限LangChain V1 的核心思路是“链式调用”。比如最常见的 RAG 流程先加载文档再做切分然后向量化检索拼 Prompt最后调大模型返回答案。这个流程用 Chain 来表达非常直观因为每一步都是固定的、顺序的中间不需要判断。但真实的企业级 Agent 没这么简单。一个订单查询助手可能要先判断用户意图再决定是查数据库还是查文档查询结果不满意还要重试或者反问用户补充条件。这种场景用 V1 写结果就是到处堆 if-else流程长了以后根本不好调试。更麻烦的是V1 对“状态”的处理很弱会话记忆、分支回退、并行任务都靠手工维护团队协作时很容易乱。所以 LangGraph 的出现本质上不是为了替代 LangChain而是补上 V1 不具备的“图式编排能力”。它不排斥 LangChain 已有的模型封装、Prompt 模板、Retriever 和 Tool 体系而是把整个流程从链改成图让状态流转、分支判断、循环重试都变得显式可见。1.2 从链式调用到图式编排LangGraph 把一次 Agent 任务拆成几个核心概念节点Node实际做事的模块。一个节点可以是调模型、跑工具、做校验也可以是另一个子图。边Edge节点之间的流转关系。普通边表示“做完前一个节点就进下一个”条件边表示“根据当前状态决定走哪个分支”。状态State整个任务共享的数据容器。节点读状态、写状态后面所有节点都能看到更新后的结果。这个设计和 V1 最大的不同在状态。V1 里每步调用是相对独立的参数要显式传递LangGraph 里所有节点围绕同一个 State 工作后续节点可以拿到前面节点产生的中间结果。比如“意图识别节点”输出一个 intent 字段后面的路由节点直接读这个字段做判断不需要额外传参。用 LangGraph 写复杂逻辑你会明显感觉流程更可控。每一步在哪里、下一步有几条可能路径、在什么条件下切换都能在一张图上看清楚。这也是企业级项目里团队更喜欢 LangGraph 的原因代码不只是自己能看懂别人接手时也能沿着图快速定位问题。1.3 分清 Agent、Skill、Tool、记忆聊 Agent 开发时这几个概念经常混在一起但实际设计时边界必须清楚。Agent 是“做决策的执行者”它负责判断当前用户需求需要调用什么能力、调用完之后是否已经满足、是否要结束。这个决策过程可以由大模型完成也可以通过规则或路由配置完成。Tool 是最小的能力单元。比如查询订单接口、执行只读 SQL、读取某个文件、调用某个搜索 API。Tool 本身不做决策只负责执行。Skill 是比 Tool 更大的业务能力包。一个 Skill 可以由多个 Tool 组成也可以自带提示词和固定处理步骤。更好的理解方式是Tool 是“我能做什么”Skill 是“我该怎么把一个复杂场景做完”。记忆分为长期和短期。短期记忆通常指当前会话的上下文消息LangGraph 里用状态里的 messages 累积长期记忆要外挂存储比如把关键信息存到数据库或向量库下次会话再取出来。企业场景里短期记忆不能无限增长长期记忆要设计清晰的存取接口否则会话一长Token 消耗和状态混乱都会冒出来。2. 搭一个最小可运行的 Agent依赖、模型接入和记忆2.1 环境准备和依赖安装先看硬性条件。Python 建议用 3.10 或 3.11不要用太旧的版本。安装核心依赖可以用一条命令pip install -U langgraph langchain-core langchain-openai这里我一般会先确认版本再更新。LangGraph 的 API 演进比较快如果项目里已经有 V1 的 LangChain 代码先固定主版本不要盲目升级否则可能出现接口不兼容。模型接入有两种常见方式。一种是直接接 OpenAI 兼容的接口比如 ChatOpenAI 指向云端模型地址另一种是接本地或私有化部署的模型服务比如 vllm 或 ollama 暴露出来的 OpenAI 兼容端点。企业场景里模型地址往往是内网服务所以建议从第一天就把模型配置抽成环境变量不要写死在代码里。2.2 用 LangGraph 搭一个带记忆的 Agent最小可运行的 Agent 不需要工具也不需要复杂路由只需要接收用户消息、调用大模型、返回回答、保留历史上下文。用 LangGraph 写大概是这样的from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI class AgentState(TypedDict): messages: Annotated[list, add_messages] llm ChatOpenAI(modelgpt-4o-mini, temperature0) def model_node(state: AgentState): response llm.invoke(state[messages]) return {messages: [response]} graph StateGraph(AgentState) graph.add_node(model, model_node) graph.add_edge(START, model) graph.add_edge(model, END) app graph.compile()这里有一个关键机制add_messages。它告诉 LangGraph当节点返回新的消息时不是覆盖原来的 messages而是把新消息追加到历史列表里。这样状态里天然就累积了对话上下文Agent 就有了短期记忆。如果只是做普通聊天这个结构已经够了。但如果要做企业级 Agent还要加一层把用户消息和中间推理过程分开记录。比如状态里不只存 messages还存 current_task、tool_calls、error_count 等业务字段方便后面做路由和排查。2.3 从单条任务到上下文连续对话图编译好之后先跑单条任务result app.invoke({messages: [(user, 你好介绍一下你自己)]}) print(result[messages][-1].content)如果这一步能正常返回再测试连续对话app.invoke({messages: [(user, 我的订单号是 1024)]}, config{configurable: {thread_id: test-1}}) app.invoke({messages: [(user, 我刚才说的订单号是什么)]}, config{configurable: {thread_id: test-1}})注意第二个 invoke 带上了 thread_id。在较新的 LangGraph 版本里thread_id 用来隔离不同会话的记忆。如果不传 thread_id每次调用都像新会话记忆不会生效。我在实测时经常发现一种误用单次调用能回答但用户换一个 session 再问答案串号了。原因是所有请求都用了同一个 thread_id或者反过来完全没用 thread_id。在设计服务时thread_id 必须由前端传来通常对应登录用户 ID 或会话 ID不能全局共用。3. TextToSQL 实战自然语言转 SQL 的真实落地逻辑3.1 别把 TextToSQL 当纯文本生成TextToSQL 是企业 Agent 里出现频率很高的能力用户用自然语言问“上个月华南区订单总金额是多少”系统要生成 SQL、执行查询、把结果转回自然语言回答。如果只是把问题丢给大模型让它生成 SQL多数情况下效果很差。原因是模型不了解你的表结构、字段含义、业务口径。同一个 status 字段status1 在订单表里可能表示“已支付”在售后表里可能表示“处理中”模型不知道。所以 TextToSQL 的落地重点不是生成本身而是上下文组织。你要把数据库的 schema、字段注释、枚举值解释、常见查询示例全部塞给模型模型才知道怎么把自然语言映射到真实表结构上。3.2 Schema 注入和示例怎么写最稳妥的做法是自动读取 information_schema生成一份精简后的表结构说明而不是手写所有字段。表很多的时候还要做一层过滤用户问题里只能关联哪些表就只给哪些表的结构避免把整套数据库 schema 全部注入 Prompt那样又费 Token 又容易让模型分心。给模型的上下文建议包含三类内容表结构表名、字段名、字段类型、是否主键。业务口径每个关键枚举字段的取值含义比如 status0 表示待支付status1 表示已支付。示例每个表给 2 到 3 条有代表性的查询 SQL覆盖统计、时间范围、分组、排序等常见场景。可以先写一个辅助函数专门拼接这些上下文def build_text2sql_prompt(user_question: str, schema_text: str) - str: return f 你是数据库查询助手。 只能生成 SELECT 查询语句禁止修改数据禁止删除数据。 表结构如下 {schema_text} 业务口径说明 - order.status: 0待支付, 1已支付, 2已取消 - 金额字段 unit: 元两位小数 - 日期字段默认格式: YYYY-MM-DD 请根据用户问题生成 SQL不要解释直接输出 SQL。 用户问题{user_question} 注意这个 Prompt 里明确了“只能 SELECT”。企业落地的底线是Agent 不能有写权限数据库账号也要给只读用户不能直接拿生产主库账号去跑。3.3 SQL 验证和上线边界生成了 SQL 不等于可以执行。我在项目里一般会加两道校验第一道是语法校验。可以用 sqlglot 解析生成的 SQL确认里面只有 SELECT没有 UPDATE、DELETE、INSERT、DROP 等危险操作。解析不过就直接让 Agent 重新生成或者返回“无法生成有效 SQL”。第二道是执行保护。在数据库连接层设置超时、最大返回行数、最大执行时间。比如查询超过 5 秒强制中断返回结果最多 100 条。这样即使 SQL 逻辑有性能问题也不会拖垮数据库。还要做结果验证。建议准备一组标准问答对比如“华北区上月订单数”“最近一天退款金额”把预期 SQL 和预期结果提前写死。每次改动 Prompt 或模型后跑一遍这组用例观察生成 SQL 是否稳定、结果是否一致。实测中TextToSQL 最怕的是“这次对、下次错”所以回归测试比单个 Case 调优更重要。上线边界也要想清楚不要一开始就让 Agent 直接查询线上业务库。可以先接影子库、只读副本或者人工审核通过后再放开。如果 Agent 面向内部运营人员可以在前端加一层“展示 SQL、人工确认再执行”这样既保留效率也留出安全闸门。4. 工作流编排条件路由、子图和并行分支4.1 Conditional Edge 的典型场景Agent 跑起来之后下一步就是把不同业务场景串成图。LangGraph 里的条件路由是最高频的用法通常用 add_conditional_edges 实现。一个很常见的需求是意图分流。用户进来之后先由一个意图识别节点判断这个问题是要查数据库还是要查知识库文档还是普通聊天然后根据 intent 字段走不同节点。def router(state): intent state.get(intent) if intent sql: return text2sql return rag workflow.add_conditional_edges( intent_node, router, { text2sql: text2sql, rag: rag, }, )这样写的好处是流程清晰新增一个意图只需要在路由函数里加一个分支不需要动其他节点。实际项目里路由函数不一定只依赖 intent也可以把“是否有足够信息”“置信度是否达标”“用户身份权限”一起纳入判断。比如用户问“帮我删掉昨天的数据”这时即使意图正确也要先走权限校验节点没有权限就直接拒绝。4.2 子图拆分和循环检测流程一旦超过五六个节点就不适合全部堆在一个图里了。比如 TextToSQL 本身就包含意图识别、表选择、SQL 生成、SQL 校验、执行、结果解释等多个步骤。这些步骤单独看是一个完整子流程放到大图里会让主流程变得臃肿。LangGraph 支持把整个子流程封装成子图。父图只需要关心子图的入口状态和最终返回状态子图内部怎么流转父图不关心。这样就能把复杂系统拆成一个个可独立测试的模块。子图拆分的判断标准一组节点是否围绕同一个业务目标且内部状态不一定要暴露给外部。如果一组节点只服务“生成并校验 SQL”这一个目标就应该拆成子图。循环检测是另一个重点。LangGraph 的图结构允许回边也就是节点可以回到前面的节点重跑AI Agent 里这通常用于“生成结果不满意重新生成”。但如果没有设计终止条件就容易死循环。我一般会给每个循环节点加计数器超过两轮就停止输出默认兜底结果。这里最容易踩的坑是模型觉得结果不够好反复自我纠偏但每次纠偏都一样白白消耗 Token 和时间。所以不要只看循环次数还要结合结果是否变化来判断。4.3 并行分支与失败重试企业级 Agent 里经常需要并行处理。典型场景是用户问了一个需要同时查多个系统的问题比如“这个客户在 CRM 的等级、在订单系统的消费总额、在客服系统的最近工单状态”这时没必要顺序查三个系统可以并行跑三个节点最后汇总结果。LangGraph 支持 fan-out 和 fan-in。实现上可以是一个节点并行调度多个子节点等所有子节点完成后再进入汇总节点。并行这里有两个经验第一个是并发数不要一开始就拉满。外部系统通常有限流比如数据库连接池、API QPS、模型服务并发。可以先跑 2 到 3 个并发观察延迟和成功率再逐步调高。第二个是每个并行分支都要单独考虑超时和失败处理。如果三个查询里有一个挂了整个任务不应该全部失败。更稳妥的做法是单个分支失败时返回一个“查询失败”的占位结果汇总节点看到后告诉用户“订单数据查询失败其他数据正常”而不是整条链路重试。对于可重试的失败场景比如接口返回 5xx可以在分支内部重试两次重试间隔采用递增策略不要疯狂发送请求。5. 可观测部署Trace、日志、指标和服务化5.1 从 LangSmith 到自建 Trace 的取舍Agent 开发完之后最难的不是写代码而是上线后出了问题怎么排查。Agent 可能经过了意图识别、工具调用、模型生成多个节点一次回答的中间步骤几十个没有可观测能力基本无从下手。官方生态里 LangSmith 是比较完整的方案可以记录每一步的输入输出、延迟、Token 消耗和状态快照。开发阶段用 LangSmith 做调试非常方便一个链路从开始到结束哪一步慢、哪一步返回了异常值一眼就能看到。但生产环境有些团队会担心数据隐私和外部依赖问题。这种情况下可以自己做可观测。不需要一开始就上很重的系统先解决三个问题能查日志、能看耗时、能回溯输入输出。具体做法是在每个节点入口和出口记录结构化日志包括节点名、会话 ID、耗时、状态、错误信息、Token 用量。这些日志统一打到标准输出或文件再接入 Prometheus、Grafana 或 ELK 这类工具。最简单的方式是写一个装饰器包在每个节点外面自动记录执行情况。5.2 服务化部署的最小流程Agent 要对外提供服务通常选 FastAPI 包装一层 HTTP 接口。核心是把 graph.invoke 封装进请求处理函数。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class AgentQuery(BaseModel): session_id: str message: str app.post(/agent) def agent_endpoint(query: AgentQuery): result graph.invoke( {messages: [(user, query.message)]}, config{configurable: {thread_id: query.session_id}} ) return {answer: result[messages][-1].content}部署时要注意一点不要把所有会话的内存状态都放在进程内。LangGraph 的记忆机制支持配置持久化后端生产环境应该把 thread 状态存到 Redis 或数据库这样应用重启后历史会话还能恢复多实例部署时也不会出现会话状态漂移。服务化之后每个请求的耗时可能会很长。业务上需要把接口设计成异步任务创建任务、轮询状态、返回结果而不是让 HTTP 请求一直挂着等模型回答。如果只做内部工具可以考虑 WebSocket 或 SSE 推送体验会好很多。5.3 生产环境要盯住的指标和常见报错核心指标其实不多请求量、平均耗时、错误率、Token 消耗、队列阻塞长度。每次上线新 Prompt 或换模型都要对比这些指标的变化。有几个报错在 Agent 开发里非常典型我列一下第一个是 “the agent execution provider did not respond in time”。这个报错通常不是 Agent 逻辑的问题而是外部模型接口响应超时。优先检查模型服务地址是否可达、模型响应时间是否过长、网络超时配置是不是太短。如果模型处理长上下文本来就慢可以把超时时间调大或者走异步模式。第二个是节点返回值缺少字段。LangGraph 要求节点返回的 dict 里的 key 必须能被状态定义接受。如果删了某个字段导致状态类型不对运行时会直接报错通常看异常堆栈就能定位。第三个是会话记忆串号。症状是用户 A 问的问题用户 B 的回答里出现了 A 的隐私数据。这个几乎都是 thread_id 管理问题。排查时先看网关和前端传的 session_id 是否唯一再看服务端有没有把 session_id 正确传递到 config 里。第四个是并行任务卡住。常见原因是某个分支没有设置超时或者外部依赖的连接池耗尽。排查顺序是先看分支节点日志到了哪一步再看外部服务的连接数和响应情况最后检查并行节点的超时配置。6. 常见坑和排查链路6.1 我最常遇到的四个问题版本兼容问题排在第一位。LangGraph 的 API 更新频繁网上很多教程是基于旧版本写的代码复制过来可能直接报错。建议以官方文档和你实际安装版本的 changelog 为准不要盲信网上的示例代码。如果项目已经稳定就把依赖版本写死至少主版本不能随意变。第二个是记忆和上下文的处理。有些团队在 Agent 里不加限制地累积 messages上下文越来越长Token 消耗越来越高响应越来越慢。解决方式是给历史消息做裁剪或摘要。企业场景里可以按时间或轮数截断也可以用摘要模型压缩早期历史。第三个是 TextToSQL 的稳定性。很多团队上线后才发现模型偶尔会生成一个字段名不存在、或者表关联方式错误的 SQL。这类问题无法完全靠模型提示词解决必须在执行层加校验。如果产品面向外部用户更要谨慎不能开放任意查询。第四个是日志缺失。Agent 本身是有状态、有分支的复杂系统如果代码里没有结构化日志上线后排查问题的成本会远高于开发成本。我个人会建议把日志从第一天就加上每个节点开始和结束都打一条哪怕很简陋也比没有强。6.2 推荐排查顺序当 Agent 运行结果不符合预期时不要急着改模型或调 Prompt先按这个顺序排查看请求参数。会话 ID 是否传了消息内容是否完整权限是否足够。看输入数据。数据库连接是否正常检索文档是否命中工具返回格式是否符合预期。看状态流转。走的是哪个分支哪个节点花费最多时间有没有进入循环。看模型响应。生成结果是否被截断是否违反指令是不是上下文塞得太长。看资源占用。模型服务、数据库连接池、服务进程日志是否有异常。遇到“输出为空”的情况我一般先确认是不是节点返回的字段写错了。一个很常见的错误是节点里计算了 answer但状态定义里没有 answer 字段或者返回时 key 写成了 answers最后下游节点读不到结果自然为空。这种问题看日志非常快不看日志就会在模型和 Prompt 上浪费很长时间。最后再强调一下整体心态LangChain 加 LangGraph 的 Agent 开发真正的复杂度不在“能不能跑通一个 Demo”而在于状态管理、业务边界、权限控制、可观测性和回归验证。如果你正在从 V1 迁移到 LangGraph或者正要做一个企业级的 TextToSQL Agent建议先把最小闭环跑稳再把路由、子图、并行、可观测一步一步加进去。每一项都验证通过之后再扩展整体稳定性会高很多。
LangChain与LangGraph实战：企业级Agent与TextToSQL
LangChain 和 LangGraph 这套技术栈最近几乎成了企业级 Agent 开发的默认组合。很多人从 LangChain V1 一路迁移到 LangGraph核心原因只有一个真实业务里的 Agent 不是一个对话框而是一套带记忆、带工具、带分支判断、带状态管理的完整工作流。如果你也在做 AI Agent 开发或者正纠结 LangChain 和 LangGraph 到底怎么分工这篇文章会给你一条能照着落地的路径从最小 Agent 搭起到 TextToSQL、工作流编排、可观测部署最后把上线后最容易踩的坑一起说清楚。1. 先想清楚LangChain 和 LangGraph 在企业级 Agent 里到底各管哪一段1.1 V1 的便利和局限LangChain V1 的核心思路是“链式调用”。比如最常见的 RAG 流程先加载文档再做切分然后向量化检索拼 Prompt最后调大模型返回答案。这个流程用 Chain 来表达非常直观因为每一步都是固定的、顺序的中间不需要判断。但真实的企业级 Agent 没这么简单。一个订单查询助手可能要先判断用户意图再决定是查数据库还是查文档查询结果不满意还要重试或者反问用户补充条件。这种场景用 V1 写结果就是到处堆 if-else流程长了以后根本不好调试。更麻烦的是V1 对“状态”的处理很弱会话记忆、分支回退、并行任务都靠手工维护团队协作时很容易乱。所以 LangGraph 的出现本质上不是为了替代 LangChain而是补上 V1 不具备的“图式编排能力”。它不排斥 LangChain 已有的模型封装、Prompt 模板、Retriever 和 Tool 体系而是把整个流程从链改成图让状态流转、分支判断、循环重试都变得显式可见。1.2 从链式调用到图式编排LangGraph 把一次 Agent 任务拆成几个核心概念节点Node实际做事的模块。一个节点可以是调模型、跑工具、做校验也可以是另一个子图。边Edge节点之间的流转关系。普通边表示“做完前一个节点就进下一个”条件边表示“根据当前状态决定走哪个分支”。状态State整个任务共享的数据容器。节点读状态、写状态后面所有节点都能看到更新后的结果。这个设计和 V1 最大的不同在状态。V1 里每步调用是相对独立的参数要显式传递LangGraph 里所有节点围绕同一个 State 工作后续节点可以拿到前面节点产生的中间结果。比如“意图识别节点”输出一个 intent 字段后面的路由节点直接读这个字段做判断不需要额外传参。用 LangGraph 写复杂逻辑你会明显感觉流程更可控。每一步在哪里、下一步有几条可能路径、在什么条件下切换都能在一张图上看清楚。这也是企业级项目里团队更喜欢 LangGraph 的原因代码不只是自己能看懂别人接手时也能沿着图快速定位问题。1.3 分清 Agent、Skill、Tool、记忆聊 Agent 开发时这几个概念经常混在一起但实际设计时边界必须清楚。Agent 是“做决策的执行者”它负责判断当前用户需求需要调用什么能力、调用完之后是否已经满足、是否要结束。这个决策过程可以由大模型完成也可以通过规则或路由配置完成。Tool 是最小的能力单元。比如查询订单接口、执行只读 SQL、读取某个文件、调用某个搜索 API。Tool 本身不做决策只负责执行。Skill 是比 Tool 更大的业务能力包。一个 Skill 可以由多个 Tool 组成也可以自带提示词和固定处理步骤。更好的理解方式是Tool 是“我能做什么”Skill 是“我该怎么把一个复杂场景做完”。记忆分为长期和短期。短期记忆通常指当前会话的上下文消息LangGraph 里用状态里的 messages 累积长期记忆要外挂存储比如把关键信息存到数据库或向量库下次会话再取出来。企业场景里短期记忆不能无限增长长期记忆要设计清晰的存取接口否则会话一长Token 消耗和状态混乱都会冒出来。2. 搭一个最小可运行的 Agent依赖、模型接入和记忆2.1 环境准备和依赖安装先看硬性条件。Python 建议用 3.10 或 3.11不要用太旧的版本。安装核心依赖可以用一条命令pip install -U langgraph langchain-core langchain-openai这里我一般会先确认版本再更新。LangGraph 的 API 演进比较快如果项目里已经有 V1 的 LangChain 代码先固定主版本不要盲目升级否则可能出现接口不兼容。模型接入有两种常见方式。一种是直接接 OpenAI 兼容的接口比如 ChatOpenAI 指向云端模型地址另一种是接本地或私有化部署的模型服务比如 vllm 或 ollama 暴露出来的 OpenAI 兼容端点。企业场景里模型地址往往是内网服务所以建议从第一天就把模型配置抽成环境变量不要写死在代码里。2.2 用 LangGraph 搭一个带记忆的 Agent最小可运行的 Agent 不需要工具也不需要复杂路由只需要接收用户消息、调用大模型、返回回答、保留历史上下文。用 LangGraph 写大概是这样的from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI class AgentState(TypedDict): messages: Annotated[list, add_messages] llm ChatOpenAI(modelgpt-4o-mini, temperature0) def model_node(state: AgentState): response llm.invoke(state[messages]) return {messages: [response]} graph StateGraph(AgentState) graph.add_node(model, model_node) graph.add_edge(START, model) graph.add_edge(model, END) app graph.compile()这里有一个关键机制add_messages。它告诉 LangGraph当节点返回新的消息时不是覆盖原来的 messages而是把新消息追加到历史列表里。这样状态里天然就累积了对话上下文Agent 就有了短期记忆。如果只是做普通聊天这个结构已经够了。但如果要做企业级 Agent还要加一层把用户消息和中间推理过程分开记录。比如状态里不只存 messages还存 current_task、tool_calls、error_count 等业务字段方便后面做路由和排查。2.3 从单条任务到上下文连续对话图编译好之后先跑单条任务result app.invoke({messages: [(user, 你好介绍一下你自己)]}) print(result[messages][-1].content)如果这一步能正常返回再测试连续对话app.invoke({messages: [(user, 我的订单号是 1024)]}, config{configurable: {thread_id: test-1}}) app.invoke({messages: [(user, 我刚才说的订单号是什么)]}, config{configurable: {thread_id: test-1}})注意第二个 invoke 带上了 thread_id。在较新的 LangGraph 版本里thread_id 用来隔离不同会话的记忆。如果不传 thread_id每次调用都像新会话记忆不会生效。我在实测时经常发现一种误用单次调用能回答但用户换一个 session 再问答案串号了。原因是所有请求都用了同一个 thread_id或者反过来完全没用 thread_id。在设计服务时thread_id 必须由前端传来通常对应登录用户 ID 或会话 ID不能全局共用。3. TextToSQL 实战自然语言转 SQL 的真实落地逻辑3.1 别把 TextToSQL 当纯文本生成TextToSQL 是企业 Agent 里出现频率很高的能力用户用自然语言问“上个月华南区订单总金额是多少”系统要生成 SQL、执行查询、把结果转回自然语言回答。如果只是把问题丢给大模型让它生成 SQL多数情况下效果很差。原因是模型不了解你的表结构、字段含义、业务口径。同一个 status 字段status1 在订单表里可能表示“已支付”在售后表里可能表示“处理中”模型不知道。所以 TextToSQL 的落地重点不是生成本身而是上下文组织。你要把数据库的 schema、字段注释、枚举值解释、常见查询示例全部塞给模型模型才知道怎么把自然语言映射到真实表结构上。3.2 Schema 注入和示例怎么写最稳妥的做法是自动读取 information_schema生成一份精简后的表结构说明而不是手写所有字段。表很多的时候还要做一层过滤用户问题里只能关联哪些表就只给哪些表的结构避免把整套数据库 schema 全部注入 Prompt那样又费 Token 又容易让模型分心。给模型的上下文建议包含三类内容表结构表名、字段名、字段类型、是否主键。业务口径每个关键枚举字段的取值含义比如 status0 表示待支付status1 表示已支付。示例每个表给 2 到 3 条有代表性的查询 SQL覆盖统计、时间范围、分组、排序等常见场景。可以先写一个辅助函数专门拼接这些上下文def build_text2sql_prompt(user_question: str, schema_text: str) - str: return f 你是数据库查询助手。 只能生成 SELECT 查询语句禁止修改数据禁止删除数据。 表结构如下 {schema_text} 业务口径说明 - order.status: 0待支付, 1已支付, 2已取消 - 金额字段 unit: 元两位小数 - 日期字段默认格式: YYYY-MM-DD 请根据用户问题生成 SQL不要解释直接输出 SQL。 用户问题{user_question} 注意这个 Prompt 里明确了“只能 SELECT”。企业落地的底线是Agent 不能有写权限数据库账号也要给只读用户不能直接拿生产主库账号去跑。3.3 SQL 验证和上线边界生成了 SQL 不等于可以执行。我在项目里一般会加两道校验第一道是语法校验。可以用 sqlglot 解析生成的 SQL确认里面只有 SELECT没有 UPDATE、DELETE、INSERT、DROP 等危险操作。解析不过就直接让 Agent 重新生成或者返回“无法生成有效 SQL”。第二道是执行保护。在数据库连接层设置超时、最大返回行数、最大执行时间。比如查询超过 5 秒强制中断返回结果最多 100 条。这样即使 SQL 逻辑有性能问题也不会拖垮数据库。还要做结果验证。建议准备一组标准问答对比如“华北区上月订单数”“最近一天退款金额”把预期 SQL 和预期结果提前写死。每次改动 Prompt 或模型后跑一遍这组用例观察生成 SQL 是否稳定、结果是否一致。实测中TextToSQL 最怕的是“这次对、下次错”所以回归测试比单个 Case 调优更重要。上线边界也要想清楚不要一开始就让 Agent 直接查询线上业务库。可以先接影子库、只读副本或者人工审核通过后再放开。如果 Agent 面向内部运营人员可以在前端加一层“展示 SQL、人工确认再执行”这样既保留效率也留出安全闸门。4. 工作流编排条件路由、子图和并行分支4.1 Conditional Edge 的典型场景Agent 跑起来之后下一步就是把不同业务场景串成图。LangGraph 里的条件路由是最高频的用法通常用 add_conditional_edges 实现。一个很常见的需求是意图分流。用户进来之后先由一个意图识别节点判断这个问题是要查数据库还是要查知识库文档还是普通聊天然后根据 intent 字段走不同节点。def router(state): intent state.get(intent) if intent sql: return text2sql return rag workflow.add_conditional_edges( intent_node, router, { text2sql: text2sql, rag: rag, }, )这样写的好处是流程清晰新增一个意图只需要在路由函数里加一个分支不需要动其他节点。实际项目里路由函数不一定只依赖 intent也可以把“是否有足够信息”“置信度是否达标”“用户身份权限”一起纳入判断。比如用户问“帮我删掉昨天的数据”这时即使意图正确也要先走权限校验节点没有权限就直接拒绝。4.2 子图拆分和循环检测流程一旦超过五六个节点就不适合全部堆在一个图里了。比如 TextToSQL 本身就包含意图识别、表选择、SQL 生成、SQL 校验、执行、结果解释等多个步骤。这些步骤单独看是一个完整子流程放到大图里会让主流程变得臃肿。LangGraph 支持把整个子流程封装成子图。父图只需要关心子图的入口状态和最终返回状态子图内部怎么流转父图不关心。这样就能把复杂系统拆成一个个可独立测试的模块。子图拆分的判断标准一组节点是否围绕同一个业务目标且内部状态不一定要暴露给外部。如果一组节点只服务“生成并校验 SQL”这一个目标就应该拆成子图。循环检测是另一个重点。LangGraph 的图结构允许回边也就是节点可以回到前面的节点重跑AI Agent 里这通常用于“生成结果不满意重新生成”。但如果没有设计终止条件就容易死循环。我一般会给每个循环节点加计数器超过两轮就停止输出默认兜底结果。这里最容易踩的坑是模型觉得结果不够好反复自我纠偏但每次纠偏都一样白白消耗 Token 和时间。所以不要只看循环次数还要结合结果是否变化来判断。4.3 并行分支与失败重试企业级 Agent 里经常需要并行处理。典型场景是用户问了一个需要同时查多个系统的问题比如“这个客户在 CRM 的等级、在订单系统的消费总额、在客服系统的最近工单状态”这时没必要顺序查三个系统可以并行跑三个节点最后汇总结果。LangGraph 支持 fan-out 和 fan-in。实现上可以是一个节点并行调度多个子节点等所有子节点完成后再进入汇总节点。并行这里有两个经验第一个是并发数不要一开始就拉满。外部系统通常有限流比如数据库连接池、API QPS、模型服务并发。可以先跑 2 到 3 个并发观察延迟和成功率再逐步调高。第二个是每个并行分支都要单独考虑超时和失败处理。如果三个查询里有一个挂了整个任务不应该全部失败。更稳妥的做法是单个分支失败时返回一个“查询失败”的占位结果汇总节点看到后告诉用户“订单数据查询失败其他数据正常”而不是整条链路重试。对于可重试的失败场景比如接口返回 5xx可以在分支内部重试两次重试间隔采用递增策略不要疯狂发送请求。5. 可观测部署Trace、日志、指标和服务化5.1 从 LangSmith 到自建 Trace 的取舍Agent 开发完之后最难的不是写代码而是上线后出了问题怎么排查。Agent 可能经过了意图识别、工具调用、模型生成多个节点一次回答的中间步骤几十个没有可观测能力基本无从下手。官方生态里 LangSmith 是比较完整的方案可以记录每一步的输入输出、延迟、Token 消耗和状态快照。开发阶段用 LangSmith 做调试非常方便一个链路从开始到结束哪一步慢、哪一步返回了异常值一眼就能看到。但生产环境有些团队会担心数据隐私和外部依赖问题。这种情况下可以自己做可观测。不需要一开始就上很重的系统先解决三个问题能查日志、能看耗时、能回溯输入输出。具体做法是在每个节点入口和出口记录结构化日志包括节点名、会话 ID、耗时、状态、错误信息、Token 用量。这些日志统一打到标准输出或文件再接入 Prometheus、Grafana 或 ELK 这类工具。最简单的方式是写一个装饰器包在每个节点外面自动记录执行情况。5.2 服务化部署的最小流程Agent 要对外提供服务通常选 FastAPI 包装一层 HTTP 接口。核心是把 graph.invoke 封装进请求处理函数。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class AgentQuery(BaseModel): session_id: str message: str app.post(/agent) def agent_endpoint(query: AgentQuery): result graph.invoke( {messages: [(user, query.message)]}, config{configurable: {thread_id: query.session_id}} ) return {answer: result[messages][-1].content}部署时要注意一点不要把所有会话的内存状态都放在进程内。LangGraph 的记忆机制支持配置持久化后端生产环境应该把 thread 状态存到 Redis 或数据库这样应用重启后历史会话还能恢复多实例部署时也不会出现会话状态漂移。服务化之后每个请求的耗时可能会很长。业务上需要把接口设计成异步任务创建任务、轮询状态、返回结果而不是让 HTTP 请求一直挂着等模型回答。如果只做内部工具可以考虑 WebSocket 或 SSE 推送体验会好很多。5.3 生产环境要盯住的指标和常见报错核心指标其实不多请求量、平均耗时、错误率、Token 消耗、队列阻塞长度。每次上线新 Prompt 或换模型都要对比这些指标的变化。有几个报错在 Agent 开发里非常典型我列一下第一个是 “the agent execution provider did not respond in time”。这个报错通常不是 Agent 逻辑的问题而是外部模型接口响应超时。优先检查模型服务地址是否可达、模型响应时间是否过长、网络超时配置是不是太短。如果模型处理长上下文本来就慢可以把超时时间调大或者走异步模式。第二个是节点返回值缺少字段。LangGraph 要求节点返回的 dict 里的 key 必须能被状态定义接受。如果删了某个字段导致状态类型不对运行时会直接报错通常看异常堆栈就能定位。第三个是会话记忆串号。症状是用户 A 问的问题用户 B 的回答里出现了 A 的隐私数据。这个几乎都是 thread_id 管理问题。排查时先看网关和前端传的 session_id 是否唯一再看服务端有没有把 session_id 正确传递到 config 里。第四个是并行任务卡住。常见原因是某个分支没有设置超时或者外部依赖的连接池耗尽。排查顺序是先看分支节点日志到了哪一步再看外部服务的连接数和响应情况最后检查并行节点的超时配置。6. 常见坑和排查链路6.1 我最常遇到的四个问题版本兼容问题排在第一位。LangGraph 的 API 更新频繁网上很多教程是基于旧版本写的代码复制过来可能直接报错。建议以官方文档和你实际安装版本的 changelog 为准不要盲信网上的示例代码。如果项目已经稳定就把依赖版本写死至少主版本不能随意变。第二个是记忆和上下文的处理。有些团队在 Agent 里不加限制地累积 messages上下文越来越长Token 消耗越来越高响应越来越慢。解决方式是给历史消息做裁剪或摘要。企业场景里可以按时间或轮数截断也可以用摘要模型压缩早期历史。第三个是 TextToSQL 的稳定性。很多团队上线后才发现模型偶尔会生成一个字段名不存在、或者表关联方式错误的 SQL。这类问题无法完全靠模型提示词解决必须在执行层加校验。如果产品面向外部用户更要谨慎不能开放任意查询。第四个是日志缺失。Agent 本身是有状态、有分支的复杂系统如果代码里没有结构化日志上线后排查问题的成本会远高于开发成本。我个人会建议把日志从第一天就加上每个节点开始和结束都打一条哪怕很简陋也比没有强。6.2 推荐排查顺序当 Agent 运行结果不符合预期时不要急着改模型或调 Prompt先按这个顺序排查看请求参数。会话 ID 是否传了消息内容是否完整权限是否足够。看输入数据。数据库连接是否正常检索文档是否命中工具返回格式是否符合预期。看状态流转。走的是哪个分支哪个节点花费最多时间有没有进入循环。看模型响应。生成结果是否被截断是否违反指令是不是上下文塞得太长。看资源占用。模型服务、数据库连接池、服务进程日志是否有异常。遇到“输出为空”的情况我一般先确认是不是节点返回的字段写错了。一个很常见的错误是节点里计算了 answer但状态定义里没有 answer 字段或者返回时 key 写成了 answers最后下游节点读不到结果自然为空。这种问题看日志非常快不看日志就会在模型和 Prompt 上浪费很长时间。最后再强调一下整体心态LangChain 加 LangGraph 的 Agent 开发真正的复杂度不在“能不能跑通一个 Demo”而在于状态管理、业务边界、权限控制、可观测性和回归验证。如果你正在从 V1 迁移到 LangGraph或者正要做一个企业级的 TextToSQL Agent建议先把最小闭环跑稳再把路由、子图、并行、可观测一步一步加进去。每一项都验证通过之后再扩展整体稳定性会高很多。
