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. 核心能力速览

这里要提前说明：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( model="gpt-4o-mini", temperature=0, base_url="http://your-endpoint/v1", api_key="your-key" )

如果你使用本地 Ollama：

from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="qwen2.5:7b", temperature=0, base_url="http://localhost:11434/v1", api_key="ollama" )

注意，具体模型名是否可用以你的模型服务为准。建议先在独立 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 生成 SQL
• execute_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( model="gpt-4o-mini", temperature=0, base_url="http://your-endpoint/v1", api_key="your-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 测试步骤与预期结果

测试上面这段代码，重点关注三个点：

1. 第一次生成 SQL 正确时，流程应该是extract_schema -> generate_sql -> execute_sql -> END。
2. 第一次生成 SQL 报错时，流程应该能看到重新进入generate_sql节点，并携带上一步的error信息。
3. 最多重试三次，超过后携带最后一次错误信息结束。

判断成功的标准是：结果里出现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_limit：

result = 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 启动 API

LangGraph 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, json=payload, timeout=60) 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_ok=True) 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", json=payload, timeout=60 ) output_dir.joinpath(f"result_{idx}.json").write_text( response.text, encoding="utf-8" ) except Exception as e: output_dir.joinpath(f"error_{idx}.json").write_text( str(e), encoding="utf-8" ) time.sleep(0.5)

这个模板适合小规模验证。生产环境建议用 Celery、Arq 或内部任务队列管理任务，同时记录每个任务的 trace_id，方便失败后回溯。

8. 可观测部署与性能观察

Agent 应用和普通 Web 服务最大的区别是：一次用户请求可能触发多次 LLM 调用、多次工具调用、多轮状态更新。没有可观测性，排错会非常痛苦。

8.1 接入 LangSmith

LangChain 生态最直接的可观测方案是 LangSmith。配置方式是在环境变量里开启 tracing：

export LANGCHAIN_TRACING_V2=true export LANGCHAIN_API_KEY=your-langsmith-api-key export LANGCHAIN_PROJECT=text2sql-agent

然后在代码里正常调用即可。启动后每次运行都会生成一条 trace，包含：

• 每个节点的输入输出
• 每次 LLM 调用的 token 消耗
• 每次数据库工具调用的耗时
• 条件路由走向
• 异常信息

如果你不想接入 LangSmith，也可以考虑 Langfuse。它是开源的 LLM 可观测平台，LangChain 官方有对应的 callback。两者的接入思想类似：在请求开始时创建 trace，请求结束后把完整事件列表上报，你就能在 Web 界面里看到每一步的执行链路。

8.2 关键性能指标

Agent 场景不建议只看“启动多少毫秒”这种单体指标，更值得关注的是：

在本地观察时，可以在每个节点打印日志：

def execute_sql(state: Text2SQLState): print(f"[execute_sql] 当前 SQL: {state['sql']}") ...

也可以在invoke后打印result和config，查看turns字段变化。

8.3 资源占用与性能优化

对于 LangGraph 这类编排框架，真正的资源瓶颈通常不在框架本身，而在底层模型推理和数据库查询。模型响应慢，整个 Agent 就慢；数据库返回大量数据，状态对象就会膨胀。

常见的优化手段有：

• 模型层：使用更快的小模型做路由，只在关键节点使用强模型
• 缓存层：对数据库 schema 做缓存，避免每次请求都读取全量表结构
• 提示词层：把表结构压缩成精简摘要，减少输入 token
• 数据库层：给查询设置超时时间，避免慢 SQL 拖垮整个流程
• 图结构层：尽可能减少不必要的串行节点，把能并行的分支并行化

9. 常见问题与排查方法

排查 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 脚手架，以后每个新项目都可以基于它改。