LangChain+LangGraph企业级Agent实战：手写TextToSQL工作流-北京尧图网络

LangChain 和 LangGraph 放在一起聊现在已经不是“要不要学”的问题而是“怎么按企业级标准落地”的问题。很多团队还在用 LangChain 早期的 Chain 链式写法做 Agent遇到分支路由、循环重试、多工具协同、SQL 生成后校验这些真实需求时会发现代码越来越乱流程越来越难控制。LangGraph 的价值就是把 Agent 变成一张可编排、可回放、可观测的图节点负责干活边负责控制流转状态在节点之间显式传递。配合 LangSmith 这类可观测平台你甚至能清楚看到每一次 LLM 调用、每一个工具返回、每一轮状态变更这对生产环境排查问题太重要了。本文会围绕“LangChain LangGraph 企业级 Agent 全栈开发”这条主线先讲清楚 V1 链式代码到 LangGraph 工作流之间的差距再带你手写一个真实可运行的 TextToSQL Agent覆盖 schema 导入、SQL 生成、SQL 执行校验、错误重试、条件路由、子图分支、API 服务部署和可观测性配置。TextToSQL 不是玩具场景它天然需要工具调用、权限控制、错误恢复和多轮迭代非常适合作为 LangGraph 的实战载体。如果你正在做 Agent 开发准备把 LangChain 项目升级成 LangGraph 工作流或者打算在内部搭建一个 TextToSQL 服务这篇文章可以直接收藏。所有代码都围绕 LangGraph 的 StateGraph 写法展开你在本地把依赖装好、Python 环境配好就能跑。1. 核心能力速览能力项说明项目类型Agent 编排框架实战基于 LangChain LangGraph核心功能StateGraph 工作流、条件路由、循环重试、子图、并行分支TextToSQL支持 schema 读取、SQL 生成、SQL 执行校验、错误反馈重试API 服务可通过 LangGraph Server 暴露 HTTP 接口支持流式输出批量任务可通过循环请求或任务队列支持批量 TextToSQL 查询可观测性支持 LangSmith / Langfuse 等 Trace 平台接入模型接入支持 OpenAI 兼容 API、Ollama 本地模型、各类国产大模型数据库支持SQLite 最适合快速验证生产建议使用只读账号连接真实数据库启动方式命令行运行脚本 / langgraph dev 启动服务适合场景企业报表问答、数据库查询助手、复杂 Agent 工作流改造这里要提前说明LangGraph 本身不限制你必须用什么模型、什么数据库。你只需要把图中各节点内部的 LLM 调用换成自己的模型端点把 SQLDatabase 连接地址换成自己的库整个框架依然成立。2. 适用场景与使用边界LangGraph 适合的团队画像很明确已经在用 LangChain 做 Agent但发现线性 Chain 写不了复杂流程或者想从零搭建一套带人工审核、带失败重试、带完整 Trace 的智能体系统。它能解决的问题包括需要把 Agent 拆成多个阶段每个阶段单独调试和测试需要根据中间结果动态决定下一步走哪个分支需要让 Agent 在调用工具失败后自动重试或切换策略需要给 SQL 生成这类高风险任务加人工确认节点需要把 Agent 暴露为 HTTP API供前端或第三方系统调用不适用或需要谨慎的场景也要说清楚。TextToSQL 本质上是让大模型生成数据库查询语句这意味着权限边界必须提前设计。如果你把数据库的写权限直接暴露给 Agent可能因为一句错误 SQL 导致数据被修改或删除。更稳妥的做法是生产环境使用只读账号SQLite 先做功能验证MySQL / PostgreSQL 使用最小权限账号。涉及核心业务库、用户隐私数据时必须先做脱敏和权限收敛不能因为“模型能生成 SQL”就直接放开。另外LangGraph 虽然擅长工作流编排但它不解决模型能力问题。基础模型的 SQL 生成准确率不高时你需要在提示词、schema 压缩、few-shot 示例、后置校验上投入更多精力这也是本文把 TextToSQL 拆成多个节点的原因。3. 环境准备与前置条件开始写代码之前先把环境确认好。下面给出一套通用检查清单具体版本请以本机实际安装为准。3.1 Python 环境建议使用 Python 3.10 或更高版本。Python 3.8 也能跑 LangChain但部分新版本依赖已经不再支持低版本新项目直接上 3.10 更省事。python --version如果机器上有多个 Python 版本建议为该项目单独建虚拟环境避免和系统环境冲突。3.2 安装 LangChain 与 LangGraph创建一个项目目录然后安装核心依赖mkdir langgraph-text2sql cd langgraph-text2sql pip install langgraph langchain langchain-openai langchain-community pip install pandas pip install langgraph-cli如果后续接入 LangSmith还需要安装pip install langsmith关于langgraph-cli它用于启动 LangGraph 开发服务和构建部署镜像。如果你只是本地跑 Python 脚本不启动 API 服务可以不装但本文后面要演示 API 部署建议装上。3.3 配置 LLM 模型LangGraph 的图中节点可以使用任意 LangChain 支持的 LLM。最通用的是 OpenAI 兼容接口from langchain_openai import ChatOpenAI llm ChatOpenAI( modelgpt-4o-mini, temperature0, base_urlhttp://your-endpoint/v1, api_keyyour-key )如果你使用本地 Ollamafrom langchain_openai import ChatOpenAI llm ChatOpenAI( modelqwen2.5:7b, temperature0, base_urlhttp://localhost:11434/v1, api_keyollama )注意具体模型名是否可用以你的模型服务为准。建议先在独立 Python 脚本里验证一次 LLM 调用能返回内容再开始搭图。3.4 准备示例数据库本文用 SQLite 作为 TextToSQL 的演示数据库零配置、无权限问题。创建一张销售订单表并插入少量示例数据import sqlite3 conn sqlite3.connect(sales.db) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY, region TEXT, product TEXT, amount REAL, order_date TEXT ) ) cursor.executemany( INSERT INTO orders (region, product, amount, order_date) VALUES (?, ?, ?, ?) , [ (华东, 笔记本, 12000, 2024-01-10), (华南, 手机, 8000, 2024-02-12), (华东, 显示器, 5000, 2024-03-05), (华北, 笔记本, 15000, 2024-03-18), (华东, 手机, 9000, 2024-04-02), ]) conn.commit() conn.close()这里只是演示数据真实项目中请使用业务表结构并严格控制数据库账号权限。4. 从 V1 Chain 到 LangGraph 工作流很多 LangChain 老项目是典型的 V1 写法用LCEL把 Prompt、LLM、输出解析器串成一条链或者在AgentExecutor里塞tools和prompt。这种写法对“问答一句话、调用一次工具”的场景完全够用但一旦你需要在 SQL 执行失败后回到上一个节点重新生成在多个工具之间做条件路由或者想人工确认某一步再继续链式结构的控制力就不够了。LangGraph 的核心抽象是StateGraph。你先把一个 Agent 拆成若干节点每个节点是一个函数输入整个状态对象输出状态更新。节点之间用边连接边可以是普通边也可以是条件边。图编译之后通过invoke启动状态会在节点之间流动。看一个最简例子from typing import TypedDict from langgraph.graph import StateGraph, START, END class SimpleState(TypedDict): value: str def node_a(state: SimpleState): return {value: state[value] - A} def node_b(state: SimpleState): return {value: state[value] - B} graph StateGraph(SimpleState) graph.add_node(a, node_a) graph.add_node(b, node_b) graph.add_edge(START, a) graph.add_edge(a, b) graph.add_edge(b, END) app graph.compile() result app.invoke({value: start}) print(result)这个例子虽然简单但它说明了 LangGraph 和 LangChain Chain 的本质区别节点函数的输入是整个状态字典输出是增量更新流程路径由边决定而不是写死在链式调用里。你可以在任意节点之间跳转也可以让一条边回到前面的节点形成循环。实际迁移的时候不需要把原来的 Agent 一次性推翻。更推荐的做法是把原来的AgentExecutor拆成“意图识别节点”“工具调用节点”“结果汇总节点”先用 LangGraph 把骨架搭起来再逐步把原来的工具和提示词迁移进节点里。5. 手写 TextToSQL Agent 工作流TextToSQL 是 LangGraph 工作流一个非常好的练手项目因为它包含完整的 Agent 生命周期读表结构、生成 SQL、执行校验、失败重试、输出答案。5.1 节点设计这里设计四个节点extract_schema从数据库提取表结构摘要generate_sql让 LLM 根据 schema 生成 SQLexecute_sql执行 SQL捕获异常should_retry条件路由节点判断是否重试整体流程是先提取 schema再生成 SQL执行后如果出现错误就带着错误信息重试最多重试三次成功后结束。5.2 完整代码下面是可以直接运行的示例你只需要确保本地有sales.db和langchain-openai相关依赖。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI from langchain_community.utilities import SQLDatabase class Text2SQLState(TypedDict): question: str schema: str sql: str result: str error: str turns: int db SQLDatabase.from_uri(sqlite:///sales.db) def extract_schema(state: Text2SQLState): schema db.get_table_info() return {schema: schema, turns: 0} def generate_sql(state: Text2SQLState): llm ChatOpenAI( modelgpt-4o-mini, temperature0, base_urlhttp://your-endpoint/v1, api_keyyour-key ) prompt f你是一个 TextToSQL 专家。数据库表结构如下 {state[schema]} 用户问题{state[question]} 如果之前生成失败错误信息如下 {state.get(error, )} 请只输出一条可直接执行的 SQL不要输出任何解释。如果无法生成请输出 ERROR: 原因。 sql llm.invoke(prompt).content.strip() return {sql: sql, turns: state.get(turns, 0) 1} def execute_sql(state: Text2SQLState): try: result db.run(state[sql]) return {result: str(result), error: } except Exception as e: return {error: str(e)} def route_after_execute(state: Text2SQLState): if state.get(error) and state.get(turns, 0) 3: return generate_sql return end builder StateGraph(Text2SQLState) builder.add_node(extract_schema, extract_schema) builder.add_node(generate_sql, generate_sql) builder.add_node(execute_sql, execute_sql) builder.add_edge(START, extract_schema) builder.add_edge(extract_schema, generate_sql) builder.add_edge(generate_sql, execute_sql) builder.add_conditional_edges( execute_sql, route_after_execute, {generate_sql: generate_sql, end: END} ) app builder.compile() result app.invoke({ question: 华东地区各产品的销售总额分别是多少 }) if result.get(error): print(最终失败错误, result[error]) else: print(生成的 SQL, result[sql]) print(查询结果, result[result])这段代码的关键点在于条件路由add_conditional_edges。execute_sql执行完之后route_after_execute根据状态里的error和turns决定是回到generate_sql重新生成还是走向END。这就是 LangGraph 循环控制的基本写法。5.3 测试步骤与预期结果测试上面这段代码重点关注三个点第一次生成 SQL 正确时流程应该是extract_schema - generate_sql - execute_sql - END。第一次生成 SQL 报错时流程应该能看到重新进入generate_sql节点并携带上一步的error信息。最多重试三次超过后携带最后一次错误信息结束。判断成功的标准是结果里出现sql和result没有error。如果出现error先检查数据库连接、表名、字段名是否与 schema 一致再看模型返回的 SQL 是否包含多余文本。这里特别提醒上面示例只考虑了“SQL 能否执行”这一层校验。真实项目中你还应该在execute_sql前增加“SQL 是否只读、是否包含危险操作”的检查避免模型生成DELETE、UPDATE、DROP等语句造成数据风险。6. 条件路由、循环、子图与并行分支上面 TextToSQL 里已经用到了条件路由和循环。这一节再展开 LangGraph 更常见的几种写法方便你迁移到自己的 Agent。6.1 条件路由条件路由通常用于意图分流。比如用户问题进来后先判断是查数据库还是查文档再进入对应分支def route_intent(state): if state[intent] sql: return text2sql return rag builder.add_conditional_edges( intent_node, route_intent, { text2sql: text2sql_node, rag: rag_node } )6.2 循环检测LangGraph 的图天然支持循环但循环次数如果没有上限模型反复失败时可能一直不终止。除了在路由函数里维护turns还可以在调用时加recursion_limitresult app.invoke( {question: 华东地区各产品的销售总额分别是多少}, config{recursion_limit: 15} )如果图内循环次数超过这个值LangGraph 会抛出GraphRecursionError。生产环境建议显式捕获from langgraph.errors import GraphRecursionError try: result app.invoke(input_data, config{recursion_limit: 15}) except GraphRecursionError: print(Agent 进入循环提前终止)6.3 子图子图适合把一段可复用的逻辑抽出来。比如你已经写好了一个report_graph在更大的 Agent 图里可以把它作为一个节点调用report_subgraph report_graph.compile() def run_report_subgraph(state): sub_result report_subgraph.invoke({ question: state[question] }) return {report: sub_result[result]}子图的好处是内部节点可以单独测试外部图只需要关心子图的输入和输出。6.4 并行分支LangGraph 支持一个节点之后并行执行多个节点。这里的“并行”指的是图结构上的并行分支实际是否真正并发执行取决于运行时线程池配置。看一个简单示例from typing import TypedDict, Annotated class ParallelState(TypedDict): results: Annotated[list, lambda a, b: a b] def node_summary(state: ParallelState): return {results: [summary_done]} def node_keyword(state: ParallelState): return {results: [keyword_done]} builder StateGraph(ParallelState) builder.add_node(summary, node_summary) builder.add_node(keyword, node_keyword) builder.add_edge(START, summary) builder.add_edge(START, keyword)当结果字段使用Annotated[list, reducer]时多个分支的更新会被合并不会互相覆盖。这种写法可以用于一个 Agent 同时生成 SQL 和生成查询说明。7. 接口 API 与批量任务Agent 写好之后下一步通常是把它暴露成服务让前端、报表系统或者别的后端服务调用。7.1 使用 LangGraph Server 启动 APILangGraph CLI 提供了langgraph dev命令可以在本地启动一个带有 API 的开发服务器。它会读取你项目里的图结构并生成可调用的 HTTP 接口。langgraph dev启动后默认服务地址通常是http://127.0.0.1:8123。访问/docs或/redoc可以查看接口文档。具体路径以你本机版本显示为准。如果langgraph dev提示无法识别图请检查项目目录下是否有可正确导入的编译后图对象。更稳妥的做法是先写一个入口模块例如agent.py在里面编译好app# agent.py app builder.compile()然后直接运行脚本验证python agent.py这样可以先把图逻辑跑通再切换到 API 服务。7.2 调用 API 的通用示例LangGraph Server 通常提供/runs/stream接口使用 POST 请求触发一次运行支持流式返回。下面是一个通用调用模板curl -N -X POST http://127.0.0.1:8123/runs/stream \ -H Content-Type: application/json \ -d { assistant_id: text2sql_agent, input: { question: 华东地区各产品的销售总额分别是多少 }, stream_mode: values }如果使用 Python 调用import requests url http://127.0.0.1:8123/runs/stream payload { assistant_id: text2sql_agent, input: {question: 2024年华东地区销售额是多少}, stream_mode: values } response requests.post(url, jsonpayload, timeout60) print(response.text)注意assistant_id需要按你实际部署的服务配置调整不同版本的 LangGraph Server 对字段要求可能不同先通过/docs确认当前接口格式。7.3 批量任务设计批量 TextToSQL 任务的难点不只是循环调用还要考虑限流、重试和结果归档。下面是一个通用批量处理模板import requests import time from pathlib import Path questions [ 华东区域销售额, 2024年4月各区域订单数, 笔记本品类总销售额, ] output_dir Path(./batch_output) output_dir.mkdir(exist_okTrue) for idx, q in enumerate(questions): payload { assistant_id: text2sql_agent, input: {question: q}, stream_mode: values } try: response requests.post( http://127.0.0.1:8123/runs/stream, jsonpayload, timeout60 ) output_dir.joinpath(fresult_{idx}.json).write_text( response.text, encodingutf-8 ) except Exception as e: output_dir.joinpath(ferror_{idx}.json).write_text( str(e), encodingutf-8 ) time.sleep(0.5)这个模板适合小规模验证。生产环境建议用 Celery、Arq 或内部任务队列管理任务同时记录每个任务的 trace_id方便失败后回溯。8. 可观测部署与性能观察Agent 应用和普通 Web 服务最大的区别是一次用户请求可能触发多次 LLM 调用、多次工具调用、多轮状态更新。没有可观测性排错会非常痛苦。8.1 接入 LangSmithLangChain 生态最直接的可观测方案是 LangSmith。配置方式是在环境变量里开启 tracingexport LANGCHAIN_TRACING_V2true export LANGCHAIN_API_KEYyour-langsmith-api-key export LANGCHAIN_PROJECTtext2sql-agent然后在代码里正常调用即可。启动后每次运行都会生成一条 trace包含每个节点的输入输出每次 LLM 调用的 token 消耗每次数据库工具调用的耗时条件路由走向异常信息如果你不想接入 LangSmith也可以考虑 Langfuse。它是开源的 LLM 可观测平台LangChain 官方有对应的 callback。两者的接入思想类似在请求开始时创建 trace请求结束后把完整事件列表上报你就能在 Web 界面里看到每一步的执行链路。8.2 关键性能指标Agent 场景不建议只看“启动多少毫秒”这种单体指标更值得关注的是指标说明节点耗时每个节点分别消耗多少时间重点看 LLM 调用耗时Token 消耗每次运行消耗多少输入/输出 token决定成本重试次数SQL 生成节点最多重试几次重试率越高说明提示词或模型越不稳定图运行深度是否接近 recursion_limit判断是否可能死循环接口排队情况高并发下请求是否长时间等待评估是否需要限流在本地观察时可以在每个节点打印日志def execute_sql(state: Text2SQLState): print(f[execute_sql] 当前 SQL: {state[sql]}) ...也可以在invoke后打印result和config查看turns字段变化。8.3 资源占用与性能优化对于 LangGraph 这类编排框架真正的资源瓶颈通常不在框架本身而在底层模型推理和数据库查询。模型响应慢整个 Agent 就慢数据库返回大量数据状态对象就会膨胀。常见的优化手段有模型层使用更快的小模型做路由只在关键节点使用强模型缓存层对数据库 schema 做缓存避免每次请求都读取全量表结构提示词层把表结构压缩成精简摘要减少输入 token数据库层给查询设置超时时间避免慢 SQL 拖垮整个流程图结构层尽可能减少不必要的串行节点把能并行的分支并行化9. 常见问题与排查方法问题现象可能原因排查方式解决方案pip 安装失败网络源不稳定、依赖冲突检查错误日志确认 Python 版本使用国内 pip 镜像新建虚拟环境调用模型报 401/403API Key 错误或模型服务未授权用 curl 单独请求模型服务测试检查 base_url、api_key、模型名称数据库连接失败路径错误、缺少数据库文件检查SQLDatabase.from_uri路径先单独连接并执行任意查询生成的 SQL 语法错误模型对 schema 理解不足查看 trace 中 generate_sql 节点的完整输入增加 schema 描述、补充 few-shot 示例Agent 重复执行同一节点路由函数逻辑写错打印路由函数返回值检查条件边映射 key 是否匹配运行时报 GraphRecursionError循环次数超过限制查看当前 turns 计数器增加 recursion_limit或提前终止循环langgraph dev找不到图入口模块未正确导出编译对象检查项目目录和模块导入确保存在可导入的app builder.compile()API 调用超时图运行时间过长查看 trace 各节点耗时减少重试次数、优化模型、设置数据库超时状态字段被覆盖未使用 reducer 合并检查 TypedDict 字段定义对需要合并的字段使用Annotated[list, reducer]SQL 执行结果过大查询返回大量行查看数据库返回行数增加 LIMIT 限制或让模型生成聚合查询排查 LangGraph 问题时先看 trace再看日志最后才改代码。没有 trace 时可以先用print在每个节点输入输出处打点定位是哪一步没按预期执行。10. 最佳实践与使用建议经过上面这些步骤你已经能跑通一个带重试机制的 TextToSQL LangGraph Agent。接下来把它从 Demo 变成可维护、可上线的东西建议按下面这套思路来做。第一维护一套最小可运行配置。项目里固定一套requirements.txt或pyproject.toml记录 LangChain、LangGraph、LangSmith 等核心依赖版本避免换机器后依赖不一致。模型 endpoint、数据库地址用环境变量管理不要写死在代码里。第二数据库权限要最小化。TextToSQL 服务建议只给数据库只读账号URL 里直接禁用写权限。如果数据库不支持账号级只读可以在 SQL 执行前加一个检查节点用简单规则匹配拒绝 DELETE、UPDATE、DROP、ALTER 等语句。更好的方案是让 SQL 查询走独立的只读副本或备库。第三模型生成结果必须人工可回溯。每个请求都要记录问题、schema、生成的 SQL、执行结果、错误信息、重试次数和 trace_id。这样用户反馈“结果不对”时你可以快速定位是模型生成错、SQL 执行错还是数据库本身数据有问题。第四批量任务要加日志和失败重试。批量处理大量问题时不要让单个失败中断整个队列。每个任务独立捕获异常记录失败原因设置重试上限最终生成一份处理报告。第五涉及版权、隐私、敏感数据的应用要谨慎。TextToSQL 虽然只是查询数据库但如果数据库里有用户隐私字段模型调用和日志记录都会增加数据暴露风险。部署时做好访问控制API 服务不要直接暴露公网必要时加鉴权。第六发布前做效果复核。不要只看“能跑通”要准备一组固定的测试问题集对比 SQL 正确率、执行成功率和响应耗时。模型版本升级、提示词调整后重新跑一遍回归测试。11. 总结与下一步LangGraph 对 Agent 开发最大的价值是把“一段链式调用”变成了“一张可控制的图”。你可以在任意节点之间跳转可以循环可以分支可以嵌子图可以在线替换模型和工具。TextToSQL 只是其中一个例子同样的模式完全可以用在客服工单分类、数据分析助手、多工具协同的自动化任务上。如果你现在刚开始接触 LangGraph最先应该验证的不是复杂架构而是把本文第二节的 StateGraph 最小示例跑起来理解状态如何在节点之间流动。然后再把 TextToSQL 的四节点流程加进去观察条件路由和重试效果。这套东西跑通之后再考虑 API 服务和 LangSmith 接入。最容易踩的坑有两个一个是忘记给循环设置次数上限导致 Agent 在节点之间反复执行另一个是没有控制数据库权限让模型生成的语句直接操作生产库。这两点一定要在项目初期就解决。下一步可以继续扩展的方向包括在流程中加入人工审核节点让 SQL 生成后先发给用户确认再执行用 Ollama 部署本地模型把 TextToSQL 完全放到内网把 LangGraph Server 接入到企业内部应用通过统一的接口给报表系统提供自然语言查询能力。建议把这份代码保存成你自己的 Agent 脚手架以后每个新项目都可以基于它改。

LangChain+LangGraph企业级Agent实战：手写TextToSQL工作流

相关推荐

C语言赋值运算符全解析：从=到复合赋值与常见误区

2026 私有云 ITSM 行业趋势与 FAQ 实战问答

自媒体矩阵发布：账号日发条数、发布间隔，蚁小二实操管控方案

更多资讯

高并发淘客返利系统缓存架构设计与优化实践

边缘计算与AI协同：腾讯云EdgeOne Pages实战解析

谷歌AI工业机器人技术解析与应用实践

Rust推理库编译失败：从Cargo到输入法集成的排障指南

Map与Set数据结构解析：从二叉搜索树到哈希表

高防护项目外设选型实战：从误区规避到点位分级落地

今日更新

PLFM_RADAR：从被动告警到主动感知的微服务健康监控系统

SPI隔离设计实战：MAX14431BAEE+与STM32F407工业级应用

数字隔离器如何根治SPI通信地环路干扰：MAX14431与TM4C129实战

本周精选

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？