最近在企业级 AI 应用落地中团队从“模型对话”迈向真实的业务自动化时最常卡住的不是 Prompt 怎么写而是多个工具串联后流程乱、状态管理靠全局变量硬撑、模型调用偶发超时、SQL 生成方向失控、上线之后出了错没法追踪。市面上关于 LangChain、LangGraph 的教程不少但大多停留在 Demo 阶段真正能拿进业务项目的工作流编排、TextToSQL、可观测部署反而讲得很少。这篇文章就围绕一条完整的 Agent 开发链路来展开从 LangChain 和 LangGraph 的概念区别出发手写一个基于工作流的智能体落地自然语言查询数据库的 TextToSQL 能力最后把可观测性和部署方案补上。适合刚接触 Agent 的开发者也适合准备把智能体放进生产环境的工程同学。1. 背景与核心概念1.1 什么是 AI Agent为什么需要工作流在传统开发里程序逻辑是确定性的输入 A走函数 B返回结果 C。但在大模型时代开发者往往希望应用能够自主决定“下一步该调用哪个工具、生成什么内容、什么时候结束”。这种具备感知、决策、行动能力的程序就是 AI Agent智能体。不过如果只是让模型不停循环调用工具很容易出现两个问题一是流程不受控模型可能反复调用同一个工具导致费用飙升二是状态不清晰多轮交互中的“当前任务进度”难以维护。企业级应用尤其不能接受这种失控。于是我们引入工作流Workflow的概念把 Agent 的任务拆成节点节点之间有方向、有条件、有分支模型并不是完全自由的而是在我们设定的图结构里做决策。这样既保留了模型的灵活性又给系统加上了工程边界。1.2 LangChain 和 LangGraph 的区别很多初学者会把 LangChain 和 LangGraph 混在一起。简单来说LangChain 是一个组件库提供模型封装、Prompt 模板、输出解析、向量存储、工具调用等基础能力LangGraph 是一个基于图的状态编排框架把 LLM 应用描述成一张有向图支持条件路由、循环、子图、并行分支等高级流程。可以这样理解LangChain 解决“怎么和大模型交互、怎么写 Prompt、怎么接工具”。LangGraph 解决“交互完成后下一步该走哪条路、状态怎么流转”。两者不是替代关系。LangChain 为 LangGraph 提供节点内部的能力LangGraph 则在更高维度上编排这些能力。实际项目里经常是 LangChain 负责封装模型和工具LangGraph 负责搭建流程骨架。1.3 TextToSQL 在企业场景中的价值TextToSQL 是 Agent 最常见的落地场景之一用户用自然语言提问系统生成 SQL 并执行最后把结果转化为自然语言回答。例如业务人员问“上个月华东区域销售额排名前五的商品是什么”系统需要拆解出筛选条件、时间范围、排序逻辑生成可执行的 SQL。TextToSQL 的难点不在于让模型输出一段 SQL而在于保证生成结果可靠、可解释、可审计。生产环境中数据库权限、敏感字段、超时控制、错误重试都必须纳入工作流。这也正是把 TextToSQL 放在 LangGraph 里实现的原因校验失败可以走重写分支执行失败可以走兜底分支整个过程可以追踪。1.4 可观测部署为什么必须提前做Agent 和普通接口的最大区别是它的执行路径不固定。同一个问题有时模型直接回答有时要调用工具有时要重试几次。如果只记录最终结果很难定位失败发生在哪一步。企业级开发需要链路追踪、日志、指标监控并且最好从开发第一天就接入而不是上线之后再补。2. 环境准备与版本说明2.1 运行环境本文的示例以 Python 为主建议使用 Python 3.10 或更高版本。操作系统不限Windows、macOS、Linux 均可。使用虚拟环境管理依赖推荐uv或venv避免污染全局环境。示例项目会用到以下核心依赖langchain langgraph langchain-openai openai sqlalchemy pymysql python-dotenv版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路。如果使用 OpenAI 之外的模型可以通过 LangChain 的模型封装替换。2.2 项目结构规划在写代码之前先把项目结构设计好。一个简单的 Agent 项目可以这样组织agent_project/ ├── app.py # 核心入口编译图并暴露调用接口 ├── state.py # 定义 Agent 的状态类型 ├── nodes/ # 节点函数每个节点做一件事 │ ├── __init__.py │ ├── planner.py # 任务规划 │ ├── text2sql.py # 自然语言转 SQL │ ├── execute_sql.py # 执行 SQL │ └── answer.py # 生成最终回答 ├── tools/ # LangChain 工具封装 │ └── db_tool.py ├── config.py # 配置读取 ├── requirements.txt └── .env # 环境变量这种结构不是为了好看而是为了让每个节点职责单一便于测试和观测。3. LangGraph 核心原理解析3.1 StateGraph把 Agent 变成一张图LangGraph 的核心是StateGraph。它定义了一张有状态的图节点Node是处理逻辑的单元边Edge是节点之间的连接状态State则在节点之间传递。一个最小示例是这样的from typing import TypedDict from langgraph.graph import StateGraph, END class MyState(TypedDict): value: int def node_a(state: MyState): return {value: state[value] 1} def node_b(state: MyState): return {value: state[value] * 2} graph_builder StateGraph(MyState) graph_builder.add_node(a, node_a) graph_builder.add_node(b, node_b) graph_builder.set_entry_point(a) graph_builder.add_edge(a, b) graph_builder.add_edge(b, END) graph graph_builder.compile() result graph.invoke({value: 1}) print(result) # {\value\: 4}每个节点函数接收当前状态返回一个字典字典中的键会更新到状态里。这里的关键是“状态”相比普通函数调用LangGraph 把每个步骤的中间结果保存在状态对象中方便整个生命周期内访问和调试。3.2 conditional_edge条件路由和分支控制真实业务中流程往往不是一条直线。比如 TextToSQL 流程中SQL 校验通过就走执行节点校验失败就要回到生成节点重新写。LangGraph 使用add_conditional_edges来实现这种分支控制。条件路由的核心是一个判断函数根据当前状态决定下一步走向哪个节点。下面是一个简化示例def route_by_validation(state: MyState): if state[value] 10: return b return END graph_builder.add_conditional_edges(a, route_by_validation, {b: b, end: END})条件路由让 Agent 具备了“自主决策”的骨架模型决定下一步做什么但所有可能的分支都是预先定义好的不会出现不可控的路径。3.3 循环、子图和并行分支LangGraph 除了支持普通边和条件边还支持循环、子图、并行分支。循环可以让 Agent 在未达到目标时反复执行某个节点比如“生成 SQL - 校验失败 - 重新生成”。LangGraph 中有循环检测机制避免无限循环消耗资源开发者也可以设置recursion_limit来控制最大步数。子图适合把复杂功能拆成独立模块。例如一个“数据分析 Agent”里面可以包含“TextToSQL 子图”和“报表生成子图”主图只负责调度子图内部细节不影响外部。并行分支则适合多个独立任务并发执行例如同时查询多个数据源。4. 完整实战基于 LangGraph 的 TextToSQL Agent4.1 业务需求假设我们有一个订单数据库表结构包含orders订单表和products商品表。业务人员希望通过 Agent 直接提问例如“本月销售额前 5 的商品有哪些”。Agent 需要完成以下步骤判断用户意图是否需要查询数据库。根据数据库元信息生成 SQL。校验 SQL 是否符合安全规则。执行 SQL 并获取结果。把结果转换为自然语言回答。4.2 定义状态我们先用TypedDict定义全局状态。这里的messages使用 LangChain 的add_messages注解可以在每次节点返回时自动追加消息。# state.py from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_question: str generated_sql: str sql_valid: bool query_result: str retry_count: intuser_question保存用户原始问题generated_sql保存模型生成的 SQLquery_result保存查询结果retry_count用于控制重试次数。4.3 规划节点规划节点负责分析用户问题决定是否需要走 TextToSQL 流程。这里可以用一个简单规则如果问题中包含“查询、销售额、排名、数量”等关键词就进入 TextToSQL否则直接回答。更复杂的场景可以用模型判断但规则在早期更可控。# nodes/planner.py def planner_node(state: AgentState): question state[user_question] # 简化判断实际项目可用大模型做意图识别 need_db any(kw in question for kw in [查询, 销售额, 排名, 数量, 订单]) return {planner_result: need_db}为了让流程更清晰我们可以在状态中增加planner_result字段。如果不需要查询数据库直接走到最终回答节点。4.4 TextToSQL 节点TextToSQL 节点的职责是让模型根据表结构元信息生成 SQL。元信息可以通过sqlalchemy从数据库读取也可以写死在 Prompt 中。示例如下# nodes/text2sql.py from langchain_openai import ChatOpenAI SCHEMA_PROMPT 数据库包含以下表 表名orders 字段order_id, product_id, sales_amount, order_date, region 表名products 字段product_id, product_name, category 请根据用户问题生成 SQL 查询只输出 SQL 语句本身。 用户问题{question} def text2sql_node(state: AgentState): llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt SCHEMA_PROMPT.format(questionstate[user_question]) sql llm.invoke(prompt).content.strip() # 简单清洗去掉可能的 markdown 代码块 if sql.startswith(): sql sql.strip() if sql.startswith(sql): sql sql[2:] return {generated_sql: sql, retry_count: state.get(retry_count, 0) 1}这里的核心思路是让模型只负责生成 SQL不直接执行。这样可以单独校验也可以单独调试。4.5 SQL 校验节点在真正执行 SQL 前必须做安全校验。生产环境更应该依赖数据库账号的最小权限而不是只靠字符串过滤。下面是一个基础校验函数# nodes/validate_sql.py import re def validate_sql_node(state: AgentState): sql state[generated_sql] # 只允许查询语句 if not re.match(r^\s*SELECT, sql, re.IGNORECASE): return {sql_valid: False} # 禁止多条语句或危险关键字 if ; in sql.replace(;, , 1): return {sql_valid: False} if re.search(r\b(DELETE|UPDATE|INSERT|DROP|ALTER|GRANT)\b, sql, re.IGNORECASE): return {sql_valid: False} return {sql_valid: True}校验失败时我们可以通过条件路由让节点回到 TextToSQL 节点重新生成但必须设置最大重试次数。4.6 执行 SQL 节点执行 SQL 时使用 SQLAlchemy 连接数据库。为了保障安全连接账号建议只开通只读权限同时设置查询超时和最大返回行数。# nodes/execute_sql.py from sqlalchemy import create_engine, text DATABASE_URL mysqlpymysql://readonly_user:passwordlocalhost:3306/orders_db def execute_sql_node(state: AgentState): sql state[generated_sql] engine create_engine(DATABASE_URL) try: with engine.connect() as conn: result conn.execute(text(sql)) rows [dict(row._mapping) for row in result.fetchmany(20)] return {query_result: str(rows)} except Exception as e: return {query_result: f执行失败: {str(e)}}注意这里只是演示思路实际项目中不要把数据库连接字符串直接放在代码里而应该从环境变量读取。4.7 回答节点回答节点负责把查询结果转成自然语言回复。可以简单拼接也可以让模型润色。# nodes/answer.py from langchain_openai import ChatOpenAI def answer_node(state: AgentState): llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt f 用户问题{state[user_question]} 查询结果{state[query_result]} 请用通俗的语言回答用户如数据为空则说明没有查询到结果。 answer llm.invoke(prompt).content return {messages: [{role: assistant, content: answer}]}4.8 组装图最后把这些节点组装成 LangGraph 图# app.py from langgraph.graph import StateGraph, END from state import AgentState from nodes.planner import planner_node from nodes.text2sql import text2sql_node from nodes.validate_sql import validate_sql_node from nodes.execute_sql import execute_sql_node from nodes.answer import answer_node def route_after_planner(state: AgentState): if state.get(planner_result): return text2sql return answer def route_after_validate(state: AgentState): if state[sql_valid] and state.get(retry_count, 0) 3: return execute if not state[sql_valid] and state.get(retry_count, 0) 3: return text2sql return answer builder StateGraph(AgentState) builder.add_node(planner, planner_node) builder.add_node(text2sql, text2sql_node) builder.add_node(validate, validate_sql_node) builder.add_node(execute, execute_sql_node) builder.add_node(answer, answer_node) builder.set_entry_point(planner) builder.add_conditional_edges(planner, route_after_planner, {text2sql: text2sql, answer: answer}) builder.add_edge(text2sql, validate) builder.add_conditional_edges(validate, route_after_validate, {execute: execute, text2sql: text2sql, answer: answer}) builder.add_edge(execute, answer) builder.add_edge(answer, END) graph builder.compile() if __name__ __main__: result graph.invoke({ user_question: 本月销售额前5的商品有哪些, messages: [{role: user, content: 本月销售额前5的商品有哪些}], }) print(result[messages][-1].content)这个图的核心是条件路由意图不明确就绕过 SQL 流程SQL 校验失败就重试重试超过限定次数直接返回错误回答。这样一来Agent 的行为变得可控。4.9 运行与验证在.env中配置模型 API Key 和数据库连接信息后运行以下命令python app.py预期输出是一段自然语言回答例如“本月销售额前5的商品分别是 A 商品 10 万元、B 商品 8 万元……”。如果模型生成的 SQL 校验失败流程会回到text2sql节点重新生成最多重试 3 次。5. 可观测部署让 Agent 能被追踪5.1 日志与链路追踪Agent 的一个关键问题是“黑盒”。因此可观测性第一个要解决的就是日志每个节点进入和退出时都打印状态。在 LangGraph 中可以在节点函数内部加日志import logging logger logging.getLogger(__name__) def text2sql_node(state: AgentState): logger.info(start text2sql, question%s, state[user_question]) # ... logger.info(finish text2sql, sql%s, sql) return {generated_sql: sql}更重要的是链路追踪。LangGraph 官方提供了 LangSmith 集成通过环境变量开启export LANGCHAIN_TRACING_V2true export LANGCHAIN_API_KEYyour_langsmith_api_key export LANGCHAIN_PROJECTagent-project开启后每次图执行都会生成一个完整的追踪链路哪个节点先执行、模型调用耗时、SQL 校验结果、重试了几次都一清二楚。对于生产环境排查这类追踪比普通日志高效得多。5.2 指标监控除了请求粒度的追踪还需要服务粒度的指标。可以记录这些信息Agent 执行总次数。TextToSQL 节点调用次数。SQL 校验失败率。数据库查询耗时。模型调用 Token 消耗。如果使用 Prometheus可以用prometheus_client定义自定义指标from prometheus_client import Counter, Histogram, generate_latest TEXT2SQL_COUNTER Counter(text2sql_calls_total, Number of TextToSQL calls) SQL_VALID_FAIL_COUNTER Counter(sql_valid_fail_total, Number of SQL validation failures) AGENT_DURATION Histogram(agent_execution_duration_seconds, Agent execution duration)然后在节点中更新指标通过一个/metrics接口暴露给 Prometheus 抓取。5.3 部署方式开发模式 vs 生产模式本地开发时LangGraph 推荐使用开发服务器可以实时查看图结构、测试状态。启动命令通常如下uv run langgraph dev这种模式适合调试包含热重载和可视化界面但不太适合承载生产流量。生产环境更常见的做法是把图编译后包装成 FastAPI 接口通过uvicorn启动uv run uvicorn app:app --host 0.0.0.0 --port 8000这里的区别在于开发模式的目的是观察迭代生产模式的目的是稳定服务。如果使用 LangGraph 平台部署流程会更完整但即便不使用平台也可以通过 FastAPI 暴露一个 POST 接口把graph.invoke包进去。5.4 接口封装示例# app_fastapi.py from fastapi import FastAPI from pydantic import BaseModel from app import graph app FastAPI() class QueryRequest(BaseModel): question: str class QueryResponse(BaseModel): answer: str app.post(/agent/query, response_modelQueryResponse) def query_agent(req: QueryRequest): result graph.invoke({user_question: req.question, messages: [{role: user, content: req.question}]}) return QueryResponse(answerresult[messages][-1].content)上线时再配合 Nginx 或网关做负载均衡、限流、鉴权。6. 常见问题与排查思路问题现象常见原因解决思路Agent 流程无法结束一直重复执行条件路由缺少终止条件或判断不严谨检查所有条件分支是否都能到达 END设置recursion_limit模型生成的 SQL 总是校验失败Prompt 没有给出清晰的表结构和规则增加示例、限制必须输出 SELECT加入 SQL 校验重试机制数据库查询报权限错误数据库账号权限不足或授权未生效确认账号只有只读权限并检查连接串生产环境使用最小权限账号模型调用超时网络不稳定或模型接口响应慢增加超时配置、重试策略、熔断机制可切换到更快的模型日志太多但没有链路信息只打了普通日志没有开启追踪接入 LangSmith 或 OpenTelemetry按 trace_id 聚合日志生产环境接口偶发 500图的状态对象中包含不可序列化对象确保状态中只放基础类型大对象放入文件或存储服务6.1 关于“the agent execution provider did not respond in time”部分团队在部署 Agent 时会遇到执行提供方未及时响应的错误。这通常意味着模型调用或工具调用超出了等待时间可能原因包括模型服务负载高、网络代理延迟、单次生成 Token 过多、工具执行过慢。排查时先看链路追踪中的耗时段是模型耗时还是工具耗时再决定是调整超时、增大重试还是优化工具执行速度。7. 最佳实践与工程建议7.1 节点职责要单一在 LangGraph 中每个节点最好只做一件事。比如“生成 SQL”不要顺便执行“执行 SQL”不要顺便生成回答。这样做的原因是单人开发还能忍多人协作时职责混乱的节点会变成无法维护的泥潭。测试也会更好写每个节点可以单独输入输出调试。7.2 状态中的字段越少越好状态是图的核心但不是垃圾桶。不要把临时变量、大段文本、数据库连接都塞进状态。状态应尽量只包含可序列化的数据字符串、数字、列表、字典。数据库连接应该由节点内部管理避免跨节点传递。7.3 数据库安全是底线TextToSQL 最大的风险就是 SQL 注入和越权。生产环境必须做到数据库账号只授予 SELECT 权限。禁止使用;拼接多条 SQL。通过数据库视图或字段权限控制敏感字段。所有查询都加上LIMIT。对模型生成的 SQL 进行二次校验。记录每次查询的用户、SQL、结果摘要方便审计。不要依赖提示词来保证安全模型生成的 SQL 不一定可靠必须靠工程手段兜底。7.4 重视可观测性预算可观测性不是越多越好。日志太全会增加成本指标太杂会增加维护负担。建议从这三个角度开始每个节点的一行摘要日志。每次 Agent 执行的总耗时和结果。TextToSQL 相关指标单独统计。等系统稳定后再逐步增加自定义事件和告警规则。可观测性是伴随系统演进的不是一次性做完。7.5 流程要可中断、可恢复企业级 Agent 流程可能会执行很久。如果中间某一步失败整个图是否要从头重跑建议把流程设计成小步快跑每个节点幂等允许失败重试。状态持久化后甚至可以从中断点继续执行但这需要额外的基础设施支持。7.6 从第一版开始就留好升级空间我在项目中会尽量避免把业务逻辑全塞进一个巨大的 Prompt。第一版可能只有一个简单的planner - text2sql - answer流程但后面一定会扩展出更多的节点和分支。设计图的时候就要想清楚后续如果要加“查询前的数据权限校验”“查询后的结果解释”应该插入哪个位置。状态字段提前预留retry_count、error这些通用字段能省下很多重构成本。8. 继续深入的方向如果你已经理解了上面的流程下一步可以往这些方向拓展在 LangGraph 中加入子图把 TextToSQL 模块独立成一个可复用的子图供其他 Agent 调用。使用并行分支同时查询多个数据源并合并结果。引入评估机制准备一组典型问题每次修改 Prompt 后自动回归测试 TextToSQL 生成结果。把图结构打包成 Docker 镜像规划好资源限制用 Prometheus 采集指标。尝试接入本地模型如 vLLM Ollama替换外部模型降低调用延迟和成本。线上环境的 Agent 不是“模型调用一次”就结束而是一个稳定的流程系统。从状态设计到条件路由从 SQL 安全到链路追踪每一环都值得用工程标准去打磨。希望这篇文章能帮你把 LangChain 和 LangGraph 从概念走向可运行、可部署、可观测的真实项目。如果你正在搭建自己的 Agent建议从最小的图开始先跑通流程再做复杂分支这样排查问题会轻松很多。