从Text-to-SQL到Agent：问数系统基础设施搭建实战

做数据平台的人大概都经历过这种场景业务方来提数你打开工单系统一看又是几十条排队等你好不容易把数取出来对方又问能不能换个口径、加个维度、再看下上周。说实话重复劳动倒还好真正的问题是“取数链路”完全靠人肉驱动典型问题没人愿意干、又不得不干。这也是我做LCODER这套AI Agent开发实战系列的初衷——把问数这件事交给智能体。前面那篇已经把方案评估和目标边界拆完了这篇进入实战第二站专门讲基础设施搭建也就是把一个问数Agent跑起来之前最底层那套东西该怎么落地。这篇文章适合谁看想做AI Agent开发但还没跑通全流程的工程师尤其后端和数据方向的同学。整篇不会只给结论核心是讲清楚每一个选型背后的理由以及我在实际搭建时踩过的坑。内容包括架构选型、工程环境、模型接入、数据底座、最小闭环验证读完你可以直接照着搭出一版能跑的问数Agent。1. 为什么问数项目必须走Agent路线而不是封装一个Text-to-SQL先抛个观点市面上很多“自然语言查数”产品本质上就是把Text-to-SQL包了一层壳子用户输入问题模型直接生成一条SQL拿去数据库执行返回结果。这样的方案demo很好做一到真实业务环境就露馅。1.1 真实取数需求长什么样我在落地问数项目之前统计过团队近一个季度的取数工单发现三个典型特征。第一表非常多不是三五张表而是几十上百张。用户说“看一下各渠道本周GMV对比上周”这一句话背后涉及订单表、渠道映射表、日期维度表甚至还牵扯到不同业务线的指标口径。第二口径极其复杂。同一个“GMV”运营说包含退款前财务说扣除退款老板说看净成交。Text-to-SQL生成的SQL再漂亮口径错了就是数字错了后果很严重。第三问题会连环追问。业务方不会只问一句就结束他看完本周数据大概率会追问“哪个渠道跌得最狠”“那个渠道的头部商品是什么”。这时候单次生成的模式就完全失效了。1.2 单次生成模式为什么扛不住把Text-to-SQL直接当产品用问题出在四个层面。一是不可修正。模型生成的SQL一旦有误用户只能重新组织语言再问一次很多不熟悉技术的人根本不知道问题出在哪。二是缺乏工具意识。模型不会主动去查表结构不会确认“你说的这个字段在库里到底存不存在”它只能凭训练时学到的常识猜猜错的概率可想而知。三是没有多轮记忆。上一句的上下文根本无法保留用户说“那再看一下那个渠道”模型完全不知道“那个渠道”是哪个。四是没有自我纠错的闭环。SQL执行报错系统直接甩给用户一句“查询失败”而没有把报错信息回传给模型让它自己修。1.3 Agent化带来的本质变化用Agent来重构问数项目底层逻辑完全不同。不是“问题→SQL→结果”的直线而是“问题→理解→规划→调用工具→观察结果→再规划→输出”的循环。我把改成Agent前后的差异梳理成一张表能力维度Text-to-SQL直连Agent化问数生成方式单次生成多步推理循环上下文管理无状态有状态支持多轮追问工具调用不支持可查元数据、查样例、执行SQL错误处理直接失败报错回填模型自动修正口径对齐靠模型猜可主动检索指标定义可扩展性加需求要改代码加工具即可扩展所以基础设施搭建的第一条原则很明确从第一行代码开始就要按Agent的架构来设计而不是先做个Text-to-SQL再想着往Agent靠。否则后面改造成本会大得让你怀疑人生。2. 技术栈与项目工程化为什么我用LangGraphPythonuv搭这套底座要把Agent落地第一件事就是选技术栈。这个环节很关键因为一旦代码写多了再换框架代价极高。2.1 编排框架LangGraph凭什么排第一Agent编排框架我对比过LangGraph、Spring AI Multi Agent、还有裸写LangChain。最终选LangGraph核心原因有四个。状态管理是LangGraph的一等公民。问数这个场景天然是“状态流”用户提问→模型决策→调工具→结果回填→再决策→输出。LangGraph用StateGraph管理状态流转节点和节点之间怎么走、怎么回退都非常直观。图结构特别适合问数流程。可以很明确地定义先走模型节点模型决定调工具就跳到工具节点工具返回结果再回到模型节点直到模型认为可以回答用户了才跳到输出节点。这种条件路由在Spring AI Multi Agent里写起来就绕很多。Python生态对Agent的支持更成熟。Function Calling、工具注册、各种数据库连接器、数据处理的库都在Python侧。Java生态当然也在快速追赶但对一个要快速跑通验证的项目来说Python的试错效率高太多。可观测性工具链完善。LangSmith、Langfuse这些追踪工具对LangGraph都有一等支持Agent排错本来就难没有可视化的调用链追踪排查问题会非常痛苦。Spring AI Multi Agent也不是不能用团队如果都是Java背景想统一技术栈它可以作为备选。但我的建议是第一版先Python侧跑通验证业务价值后面如果需要Java化再让Java团队照着接口协议平移。2.2 Python环境用uv替代pip和venvPython环境管理这块我强烈建议直接用uv不要再手动建venv再装pip包了。uv是Rust写的新一代包管理工具创建虚拟环境、解析依赖、装包速度都比传统方式快一个数量级适合工程化项目。初始化项目就三步uv init lcoder_qa_agent cd lcoder_qa_agent uv add langgraph langchain-openai pydantic-settings sqlalchemy pymysql pandas这几行跑完pyproject.toml就自动生成好了还会自动创建虚拟环境。后续加依赖直接uv add会顺带把传递依赖解析好不用手动处理版本冲突。提示项目里还应该早一点把ruff加进来做代码规范检查uv add --dev ruff mypy。Agent项目逻辑复杂代码一乱后面想维护都难。2.3 项目目录结构怎么划分基础设施阶段就要确立好目录职责这关系到后续所有代码的组织方式。我采用的目录结构比较简单清晰lcoder_qa_agent/ ├── config/ # 配置读取环境变量管理 │ └── settings.py ├── core/ │ ├── llm.py # 模型接入封装 │ ├── tools.py # 工具注册与统一执行器 │ └── graph.py # LangGraph状态图定义 ├── datasource/ │ ├── connection.py # 数据库连接管理 │ ├── metadata.py # 元数据抓取与缓存 │ └── dialect.py # SQL方言适配 ├── schemas/ │ └── models.py # Pydantic模型定义 ├── prompts/ # 提示词模板文件 ├── logs/ # 日志输出目录 ├── .env # 本地环境变量 └── main.py # 入口每个目录职责单一core下面放Agent核心逻辑datasource管数据层schemas统一数据结构。快速迭代的项目最怕没有边界目录不清晰写一个月之后连自己都找不到代码在哪。2.4 配置管理敏感信息一律走环境变量配置这块不能偷懒数据库密码、API Key这种信息绝对不允许硬编码进代码仓库。我用pydantic-settings统一管理from pydantic_settings import BaseSettings class Settings(BaseSettings): openai_api_key: str openai_base_url: str https://api.openai.com/v1 model_name: str gpt-4o-mini temperature: float 0.1 db_url: str db_readonly_url: str db_max_rows: int 200 metadata_cache_path: str ./cache/metadata_cache.json class Config: env_file .env env_file_encoding utf-8 settings Settings().env文件长这样OPENAI_API_KEYsk-xxx OPENAI_BASE_URLhttps://api.openai.com/v1 DB_READONLY_URLmysqlpymysql://readonly_user:passwordhost:3306/warehouse特别提醒一个容易被忽略的点代码里用的必须始终是只读账号的连接串写库权限绝对不能给Agent。这个后面数据安全那节还会再展开。3. 模型接入层与Function Calling的封装Agent的“手”从哪来Agent和普通的对话机器人最本质的区别就是能调用工具。而工具调用能力在当前主流模型上靠的是Function Calling机制。3.1 统一模型接入层模型接入我用的是LangChain的ChatOpenAI它可以兼容任何OpenAI接口标准的服务。这么做的好处是后面想从GPT切到国产模型、或者接本地模型只需要改配置不用改业务代码。from langchain_openai import ChatOpenAI from config.settings import settings def get_llm() - ChatOpenAI: return ChatOpenAI( modelsettings.model_name, temperaturesettings.temperature, api_keysettings.openai_api_key, base_urlsettings.openai_base_url, )3.2 Function Calling的底层逻辑很多第一次接触Function Calling的人对“模型会执行函数”有误解。实际上模型根本不执行任何代码。流程是这样的调用模型时传入tools参数里面描述了每个工具的名字、用途、参数结构。模型看到用户问题后如果认为需要某个工具会返回一个tool_calls对象里面包含工具名和参数字典。真正执行工具的是我们自己的代码执行完再把结果作为一条消息回传给模型模型根据结果继续推理。打个比方模型是老板工具是下属。老板本人不会干活只会下达指令——“你去查一下用户表的行数”下属干完把结果汇报上来老板再决定下一步安排。整个循环里出力的始终是下属老板只负责决策。3.3 用Pydantic定义工具参数Schema工具参数的定义直接决定模型能不能正确调用。我习惯用Pydantic写数据模型再自动转成模型期望的JSON Schema。from pydantic import BaseModel, Field from typing import Literal class QueryDatabaseInput(BaseModel): sql: str Field(description要执行的SQL查询语句必须是SELECT语句) db_type: Literal[mysql, postgresql, clickhouse] Field( defaultmysql, description目标数据库类型 ) limit: int Field(default50, ge1, le500, description查询返回的最大行数)定义这个模型后转成工具的JSON Schemadef get_query_tool_schema(): return { type: function, function: { name: query_database, description: 执行只读SQL查询获取业务数据, parameters: QueryDatabaseInput.model_json_schema(), } }注意这里描述的字段信息要尽可能具体ge1、le500这种约束条件也不要省略模型会读取这些信息来生成合法的参数。3.4 LangGraph节点里的模型调用封装模型调用不能只调一次而是在Agent循环里被反复调用。所以在LangGraph节点里我封装了一个统一方法from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from core.tools import get_all_tool_schemas def llm_node(state): messages state[messages] llm get_llm().bind_tools(get_all_tool_schemas()) response llm.invoke(messages) return { messages: [response], need_tool: len(response.tool_calls) 0 }关键点在need_tool这个状态字段。LangGraph的下一步路由就是依据它来判断如果模型返回了tool_calls就跳转到工具执行节点如果没有说明模型已经可以直接回答就进入输出节点。整个Agent的手就是从这层封装开始的后面每增加一个新能力比如查指标口径、查表结构只需要加一个工具函数并注册进来不需要改动图结构。4. 问数项目的真正底座数据库连接与Schema元数据管理如果说模型是Agent的大脑那元数据就是问数项目的记忆。这一层做不好后面所有查询的准确率都上不去。4.1 为什么元数据如此重要模型并不知道你的数据库里有哪些表、每个字段是什么含义。不喂元数据他只能凭“常识”编SQL字段名编错了就是灾难。举个例子用户问“查一下累计注册用户数”库里那张表可能叫t_user_register_record字段叫user_id。模型凭训练记忆猜测可能写成查users表的id字段。你把它接上真实库一执行就是“Table not found”。所以问数Agent必须有一个机制能在模型生成SQL之前把相关的表结构信息喂给它。这其实就是一种程序化RAG——从元数据里检索最相关的表结构拼进提示词。4.2 元数据抓取与缓存实现抓取表结构最简单的方式是用SQLAlchemy的inspectorfrom sqlalchemy import create_engine, inspect def fetch_table_metadata(db_url: str, table_name: str) - dict: engine create_engine(db_url) insp inspect(engine) columns insp.get_columns(table_name) return { table: table_name, columns: [ { name: col[name], type: str(col[type]), nullable: col[nullable], primary_key: col.get(primary_key, False) } for col in columns ] }也可以用information_schema直接查对表多的情况更灵活SELECT column_name, data_type, is_nullable, column_comment FROM information_schema.columns WHERE table_schema your_schema AND table_name your_table ORDER BY ordinal_position;但有一个很现实的性能问题如果每次对话都实时抓取所有表的元数据数据库压力非常大。所以缓存是必须的。我采用本地文件缓存抓取一次后写入JSON设置合理的过期时间import json, os from config.settings import settings def load_metadata_with_cache(): path settings.metadata_cache_path if os.path.exists(path): with open(path, r, encodingutf-8) as f: return json.load(f) metadata fetch_all_tables_metadata() os.makedirs(os.path.dirname(path), exist_okTrue) with open(path, w, encodingutf-8) as f: json.dump(metadata, f, ensure_asciiFalse, indent2) return metadata注意线上环境建议把缓存放到Redis并用表结构变更事件来做主动失效否则用户改了一张表你还给他看旧结构生成的SQL照样错。4.3 方言适配别让模型把SQL写串了不同的数据库SQL语法差异非常大。模型容易出问题的地方集中在时间函数、分页语法、字符串拼接这几个点上。我用过MySQL、PostgreSQL、ClickHouse踩过的坑整理如下功能MySQLPostgreSQLClickHouse分页LIMIT 10 OFFSET 20LIMIT 10 OFFSET 20LIMIT 20, 10时间差DATEDIFF(d1, d2)d1 - d2dateDiff(day, d2, d1)字符串拼接CONCAT(a, b)a || bconcat(a, b)当前时间NOW()NOW()now()类型转换CAST(x AS CHAR)x::texttoString(x)方言问题不能在生成SQL之后才去管要在Prompt层面就做好强约束。我是在系统提示词里明确写明“当前目标数据库是MySQL所有SQL必须使用MySQL语法禁止使用其他数据库的语法”同时在工具执行层做一层校验发现分页写法不对就拦截并提示模型重新生成。4.4 查询安全边界必须前置基础设施阶段就要把数据安全的钩子埋好否则后面接生产库就是定时炸弹。我做了四件事只读账号Agent连接的任何数据源都是只读权限从连接串层面杜绝写操作。强制LIMIT在SQL执行器里对传入SQL做解析没有LIMIT就自动追加防止拖垮线上库。超时控制单次查询超过20秒直接杀掉返回超时错误。禁止多语句校验SQL中不能存在分号分隔的多条语句防止注入式的恶意构造。用SQLAlchemy配置只读连接from sqlalchemy import create_engine def get_readonly_engine(): engine create_engine( settings.db_readonly_url, pool_size5, pool_recycle1800, connect_args{ read_timeout: 20, charset: utf8mb4, init_command: SET SESSION TRANSACTION READ ONLY, } ) return engine这一层属于那种“不出事你觉得多余一出事你就知道值多少钱”的工程防护。5. 跑通最小闭环从自然语言问题到SQL查询结果基础设施搭好不等于能用必须把一条最简链路完整跑通。我直接给出这段核心验证代码你可以在自己的环境里跑起来看效果。5.1 定义状态与图结构LangGraph要求先定义全局状态from typing import TypedDict, List from langchain_core.messages import BaseMessage class AgentState(TypedDict): messages: List[BaseMessage] need_tool: bool然后构建图from langgraph.graph import StateGraph, END from core.nodes import llm_node, tool_executor_node def build_agent_graph(): graph StateGraph(AgentState) graph.add_node(llm, llm_node) graph.add_node(tools, tool_executor_node) graph.add_edge(llm, tools, conditionlambda state: state.get(need_tool, False)) graph.add_edge(tools, llm) graph.add_edge(llm, END, conditionlambda state: not state.get(need_tool, False)) graph.set_entry_point(llm) return graph.compile()这里的路由逻辑是进入llm节点如果模型说要调用工具就跳去tools节点执行执行完带着工具结果回到llm再循环直到模型认为不需要工具了就走到END节点返回最终答案。5.2 工具执行节点工具执行节点做的事很简单取模型返回的tool_calls逐个执行把结果包成ToolMessage塞回状态。核心代码from langchain_core.messages import ToolMessage from core.tools import tool_executor def tool_executor_node(state): messages state[messages] last_message messages[-1] tool_messages [] for tool_call in last_message.tool_calls: result tool_executor.invoke(tool_call) tool_messages.append(ToolMessage( contentresult, tool_call_idtool_call[id], )) return {messages: messages tool_messages, need_tool: True}执行工具时发现SQL报错怎么办这里就是Agent和普通程序的核心区别——把错误信息当成普通文本原样丢回去给模型。模型看到报错结合表结构大概率能自己写出修正后的SQL。5.3 跑一个完整例子配置好环境后在main.py里写个最小入口from langchain_core.messages import HumanMessage from core.graph import build_agent_graph if __name__ __main__: agent build_agent_graph() result agent.invoke({ messages: [HumanMessage(content帮我查一下用户表的总行数)], need_tool: False, }) print(result[messages][-1].content)跑完这个之后你会看到链路一层层执行下来模型识别到要查用户表调用了query_database工具传了SELECT COUNT(*) FROM users这样的SQL工具执行返回数字模型把这个数字包装成一句自然语言。这个最小闭环的意义在于它证明了基础设施是通的之后所有上层的能力多轮对话、指标口径检索、自动生成报表都是在加厚这条路而不是另起炉灶。5.4 结果格式化与边界情况返回给用户的不能是一张干巴巴的表。我在工具执行后加了一层格式化import pandas as pd def format_query_result(rows, columns): df pd.DataFrame(rows, columnscolumns) return df.to_markdown(indexFalse)把查询结果转成markdown表格模型读取后再转述成自然语言用户感知就非常好。同时设置最大返回行数比如200行超出部分截断并提示模型“结果集过大是否需要聚合统计”。三种高频边界情况也要处理空结果模型需要给出“没有查到相关数据”的结论而不是尴尬地交出空表。SQL报错见前面说的把报错丢回去让模型修正最多重试两次再失败就明确告知用户当前无法回答。字段不存在模型可能生成了不存在的字段名我会把“当前表结构中包含哪些字段”作为工具结果一并提供帮助模型自检。6. 基础设施层我踩过的三个坑以及下一步怎么走最后说点实在的下面这三个问题都是我实际构建过程中踩进去过的分享出来希望后来者少走弯路。6.1 坑一连接池不配置一次元数据全量抓取拖垮业务库刚开始做元数据抓取时我没有加缓存也没有加连接池限制直接一个insp.get_table_names()全库扫描结果高峰期把业务库的CPU打满了。后来改成从库抓取、本地文件缓存、每天定时刷新一次彻底解决了。问数Agent的一切访问都要按“为生产库减负”来设计。6.2 坑二LLM生成SQL的方言混用在MySQL库上跑模型偶尔会生成ILIKE这种PostgreSQL语法。起初我没在工具层做校验直到线上有人问了个复杂问题生成的SQL执行直接报错。后面我在Prompt里强调了方言并加了一层语法规则校验器时间函数、分页写法不对就直接拦截并回传模型修正准确率一下就上来了。6.3 坑三工具定义太复杂模型干脆不调用了一开始我设计工具时在参数里放了很多可选字段总想着一个工具能覆盖所有场景。结果模型面对复杂工具更倾向于“猜答案”而不是“调工具”。解决办法是拆分一个工具只干一件事参数尽量精简必填项用Field标清楚。工具质量比工具数量重要得多。6.4 关于MCP、多Agent架构和本地模型的取舍基础设施达到稳定之后接下来无非是几个方向接入MCP协议让Agent能统一调度外部知识库、API和更多数据源把单Agent拆成多Agent让“语义理解”“取数执行”“结果解释”各司其职以及私有化部署本地模型解决数据合规问题。我的实际判断是第一阶段先单Agent跑通业务闭环不要一上来就多Agent问题复杂度不够时拆了只会增加系统开销和排查难度。MCP可以先把适配层设计好等外部数据源多起来再正式接入。本地模型则看数据合规要求如果延迟和效果可以接受尽早切本地模型会省很多后续麻烦。6.5 基础设施的搭建节奏最后再分享一点个人体会。基础设施最容易犯的错不是建得不够多而是建得太全。我见过很多团队在Agent还没跑通的时候花了一周纠结用Postgres还是Milvus存记忆、用Redis还是Mongo做缓存——这些对第一版毫无意义。比较靠谱的节奏是先跑通最小闭环验证Agent能答对业务问题再回头加固数据层、安全层、可观测性之后再扩展工具、接入更多数据源、拆多Agent。基础设施的价值只有在业务跑起来之后才真正体现它是用来支撑业务迭代的不是用来展示技术美感的。这一期关于基础设施先聊到这下一期进入Prompt工程和Agent节点设计的细节我会把问数场景的系统提示词模板、工具调用的约束策略、以及多轮对话状态管理全部拆一遍。