之前做 Agent 项目时最让我头疼的不是模型效果而是数据接入。每次换一个业务场景Agent 就要重新问数据库里有哪些表、字段是什么含义、哪些是关键字段同一个数据库上一轮刚告诉它订单表在哪下一轮它又把表名记错了。说白了模型本身不笨但它每次开聊都像第一天入职什么都要从头问一遍。这篇文章想聊的就是怎么给 AI Agent 建一个“数据接入层”让查询工具、Schema 说明书、结果格式化、安全边界都沉淀下来而不是让 Agent 每次从零开始摸数据。我准备按实际落地顺序来写先分析问题本质再实现一个最小可用的数据接入层然后通过 Function Calling 让 Agent 自己查库最后补充数据治理和工程化建议。示例使用 Python MySQL OpenAI 兼容接口核心代码可以直接复制改造成自己的项目。1. 为什么 AI Agent 总是在“摸数据”1.1 最典型的五个“摸数据”现象如果你做过数据库问答或者数据查询类 Agent应该对下面这些场景不陌生。第一个现象是重复获取表结构。用户问“这个月的订单量是多少”Agent 先查一遍有哪些表再查订单表有哪些字段下一轮用户问“按城市统计订单”它又要重新查一遍完全没记住之前已经拿到的结构信息。第二个现象是靠猜字段含义。实际业务表里经常有status、type、flag这类字段存储的值是 0、1、2 这种枚举值。没有字段注释没有数据字典Agent 只能根据列名猜测status1代表什么。猜对了皆大欢喜猜错了整个回答都是错的。第三个现象是结果集过大导致上下文爆炸。有些表有几十个字段其中还有remark、content这种大文本字段。Agent 执行SELECT *后工具把几千行数据原样塞回上下文一次回答就把 Token 预算烧掉大半。第四个现象是权限边界模糊。很多内部 Demo 直接把高权限账号配置给 AgentAgent 生成的 SQL 没有做任何校验一旦模型被诱导生成DELETE或UPDATE语句后果会很严重。开发环境还好生产环境这是底线问题。第五个现象是指标口径不统一。同一个“订单金额”有的表叫amount有的表叫total_price有的甚至不区分币种。Agent 没有指标口径说明每次只能根据问题临时猜不同时间的回答结果不一致业务方根本不敢用。1.2 问题的本质Agent 缺的不是聪明而是数据接入层上面这些现象看起来是模型能力问题但本质上是一个工程问题Agent 缺少一个稳定的数据接入层。大模型擅长的是文本理解和推理它本身不具备连接数据库的能力。它需要工具去查库需要上下文知道表结构需要明确的安全边界需要把结果格式化后再理解。这些能力如果放在 prompt 里临时写每次都是“从零开始”如果沉淀成一层代码或者一套工具Agent 每次只需要调用不需要重新摸索。所以“别再让 Agent 从零开始摸数据”的关键不是换一个更强的模型而是把数据接入能力组件化。连接器负责连库Schema 读取器负责生成说明书查询执行器负责安全查询格式化器负责压缩结果Agent 只负责调用工具和生成回答。1.3 目标架构从“临时问”到“开箱即用”我们先看落地后的整体流程。用户问题 ↓ Agent 决策 ├─ 调用 get_database_schema 读取数据库说明书 ├─ 调用 execute_readonly_query 执行安全的 SELECT 查询 └─ 拿到格式化后的 JSON 结果 生成最终回答这里最关键的一步是 Agent 不要直接对接数据库而是对接我们提供的工具。数据库连接信息、Schema 缓存、安全校验、结果截断这些逻辑全部封装在工具内部。用户问数据时Agent 先看说明书再写 SQL再执行查询整个过程不再依赖模型“记住”上一次对话的内容而是每次都能拿到最新的、结构化的数据资产信息。下面我们开始搭建这套数据接入层。2. 先搭一套可复用的数据接入层2.1 连接器需要做哪些事数据接入层的第一块地基是一个统一的数据库连接器。它的职责有三个。第一是管理连接。封装pymysql连接参数统一字符集统一游标类型避免每个工具函数重复写一遍连接逻辑。第二是读取元数据。通过SHOW TABLES和SHOW FULL COLUMNS获取表名、字段名、字段类型、注释、主键等结构信息并把这些信息缓存在内存里。第三是配合查询执行器做安全约束。连接器本身不直接暴露给 AgentAgent 能调用的只有我们定义好的工具函数。我建议把连接器单独一个文件不要和业务逻辑混在一起。这样以后换数据库类型只需要替换连接器内部实现Agent 核心代码不用动。2.2 最小连接器代码我们以 MySQL 为例使用pymysql实现一个最小连接器。# 文件路径agent_data_demo/database.py import pymysql class DatabaseConnector: 统一封装数据库连接、Schema 读取与只读查询入口。 def __init__(self, config: dict): self.config config self._schema_cache: dict | None None def _connect(self): return pymysql.connect( hostself.config[host], portself.config[port], userself.config[user], passwordself.config[password], databaseself.config[database], charsetutf8mb4, cursorclasspymysql.cursors.DictCursor, ) def get_schema(self, force_refresh: bool False) - dict: 返回 {表名: [字段信息列表]}默认使用内存缓存。 if self._schema_cache is not None and not force_refresh: return self._schema_cache with self._connect() as conn: with conn.cursor() as cursor: cursor.execute(SHOW TABLES) raw_tables cursor.fetchall() tables [] for row in raw_tables: # SHOW TABLES 的结果列名随数据库名变化统一取第一个值 tables.append(list(row.values())[0]) schema {} for table in tables: # 反引号包裹表名避免特殊字符影响 cursor.execute(fSHOW FULL COLUMNS FROM {table}) schema[table] cursor.fetchall() self._schema_cache schema return schema这段代码有两个细节需要注意。一是SHOW TABLES的结果列名在不同版本 MySQL 里可能不同直接写死列名容易出问题。这里统一用list(row.values())[0]获取第一个值兼容性更好。二是SHOW FULL COLUMNS FROM \表名能拿到非常完整的字段信息包括Field字段名、Type类型、Null、Key主键/索引、Default、Comment注释。这些信息后面会用来生成 Agent 可读的说明书。2.3 连接安全校验连接器的配置信息不要写死在代码里建议放到环境变量或者配置中心。生产环境一定要用只读账号账号的权限只开放给业务库不要给全局权限。下面是配置示例实际使用时要把账号密码改成自己的。config { database: { host: 127.0.0.1, port: 3306, user: demo_user, password: demo_pass, database: demo_shop, } }需要提醒的是pymysql连接本身不能限制 SQL 类型只能靠账号权限和代码层面的校验来兜底。所以数据库账号建议用最小权限原则创建只授予SELECT权限连INSERT、UPDATE、DELETE、DDL都不要给。这一步做好了Agent 就算生成恶意语法也执行不了。3. 把数据库结构变成 Agent 能读懂的“说明书”3.1 为什么不能直接粘贴所有建表语句有人说直接把SHOW CREATE TABLE的结果塞给模型不就行了在小项目里可以但一旦表数量超过几十张建表语句就会非常长远远超出上下文窗口能承受的范围。更重要的是建表语句适合人看不一定适合 Agent 理解。一张表可能有几十个字段其中不少是内部冗余字段全部列出来反而会干扰 Agent 判断该用哪个字段。比如订单表里有created_time和create_ts两个字段含义相同放在一起 Agent 就会纠结。所以我们要做的是结构化的 Schema 提取与压缩只保留 Agent 回答问题需要的信息表名、字段名、字段类型、字段注释、主键标识。控制好这条信息流的长度既能让 Agent 知道有哪些数据又不把上下文塞满。3.2 Schema 缓存避免每次从零开始DatabaseConnector.get_schema()内部已经实现了缓存首次调用读取数据库结构之后直接从内存返回。这样 Agent 在一个会话里多次调用工具时不需要每次都重新访问数据库。缓存的思路可以继续扩展进程级缓存适合单机部署的 Agent 服务。Redis 缓存适合多实例部署Schema 变更后通过版本号或手动刷新。定时刷新数据库结构每天可能变化时设置定时任务刷新缓存。Schema 变更频率通常远低于业务数据变更频率所以缓存策略可以激进一些。唯一要注意的是当 DBA 改了表结构后Agent 拿到的还是旧 Schema可能查询报错。因此工具里要保留force_refreshTrue的参数并提供一个“手动刷新”入口。3.3 把 Schema 转成紧凑文本拿到结构信息后需要把它转成 Agent 更容易阅读的文本格式。这里用format_schema_for_llm函数实现。# 文件路径agent_data_demo/formatter.py import json def format_schema_for_llm(schema: dict, max_tables: int 30) - str: 把 Schema 转成紧凑文本按需截断表数量。 lines [] tables list(schema.items())[:max_tables] for table, columns in tables: lines.append(f表 {table}:) for col in columns: field col.get(Field, ) col_type col.get(Type, ) key PK if col.get(Key) PRI else comment col.get(Comment, ) line f - {field} {col_type} if key: line f [{key}] if comment: line f 说明: {comment} lines.append(line) return \n.join(lines)这个输出格式有几个优点。一是按“表名 字段列表”组织Agent 一眼能看到完整结构。二是保留PK标记方便 Agent 理解关联关系。三是字段注释直接拼进文本Agent 不会靠猜。四是max_tables参数可以控制数据量避免所有表都灌进上下文。实际输出效果类似表 orders: - id BIGINT [PK] 说明: 主键 - order_no VARCHAR(64) 说明: 订单号 - user_id BIGINT 说明: 用户ID - amount DECIMAL(10,2) 说明: 订单金额 - status TINYINT 说明: 状态0待支付 1已支付 2已发货 3已完成 - created_at DATETIME 说明: 下单时间到这里Agent 已经具备“读懂数据库结构”的能力了。接下来就是让它动手查数据。4. 让 Agent 自己动手查数据工具化查询4.1 查询工具设计给 Agent 暴露工具时接口要尽量少、职责要尽量清晰。最少只需要两个工具一个读 Schema一个执行只读查询。这里不要暴露“写操作”工具比如“执行任意 SQL”或者“修改数据”。我们的目标场景是数据分析不是数据运维。Agent 能做的就是查看说明、写 SELECT、拿结果剩下的操作全部在工具内部完成。工具定义使用 Function Calling 的标准格式下面这段代码可以在大多数支持 tools 的模型上直接使用。# 文件路径agent_data_demo/tools.py TOOLS [ { type: function, function: { name: get_database_schema, description: 获取当前数据库的表结构包括表名、字段名、字段类型、字段说明。, parameters: { type: object, properties: {}, required: [], }, }, }, { type: function, function: { name: execute_readonly_query, description: 执行只读 SELECT 查询并返回 JSON 结果。只允许 SELECT 语句不允许修改数据。, parameters: { type: object, properties: { sql: { type: string, description: 完整的 SELECT SQL 语句, } }, required: [sql], }, }, }, ]为什么把get_database_schema也设计成工具而不是直接放进 System Prompt因为 Schema 可以很大放在 System Prompt 里会占用大量空间而且如果放在提示词里Agent 可能会在回答时“引用”一份已经过期的 Schema。设计成工具后Agent 需要时再拉取拉到的永远是当前缓存里的最新版本。4.2 SQL 安全边界查询执行环节是整个系统最容易出问题的地方必须做多层安全限制。这里给出一个相对完整的只读查询执行器。# 文件路径agent_data_demo/query_executor.py import re from database import DatabaseConnector class QueryExecutor: 只读 SQL 查询执行器带基础安全校验。 FORBIDDEN_KEYWORDS [ insert, update, delete, drop, alter, truncate, grant, revoke, create, replace, load_file, into outfile, into dumpfile, sleep(, benchmark(, ] def __init__(self, connector: DatabaseConnector, limit: int 50): self.connector connector self.limit limit def execute_query(self, sql: str) - list[dict]: sql sql.strip().rstrip(;).strip() if not sql: raise ValueError(SQL 为空) # 第一条防线只允许 SELECT first_keyword sql.split(maxsplit1)[0].lower() if first_keyword ! select: raise ValueError(只允许执行 SELECT 查询) # 第二条防线屏蔽危险关键字 lowered sql.lower() for keyword in self.FORBIDDEN_KEYWORDS: if keyword in lowered: raise ValueError(fSQL 包含禁止关键字: {keyword}) # 第三条防线自动追加 LIMIT sql self._add_limit(sql) with self.connector._connect() as conn: with conn.cursor() as cursor: cursor.execute(sql) rows cursor.fetchall() return rows[: self.limit] def _add_limit(self, sql: str) - str: if re.search(r\blimit\s\d, sql, re.IGNORECASE): return sql return f{sql} LIMIT {self.limit}这个实现里有三个需要注意的点。第一INSERT、UPDATE、DELETE等写操作会被第一条防线直接拦截。第二sleep(、benchmark(这类可能导致数据库压力增大的函数被列入黑名单避免 Agent 被诱导执行慢查询。第三查询结果返回前再做一次截断保证单次工具调用不会把几万行数据全部塞回上下文。需要说明的是关键字黑名单只是简单规则生产环境更推荐使用 SQL AST 解析器或者数据库代理来做到更准确的语法级校验。但作为一套可落地的起点上面的代码足够挡住大部分风险。4.3 查询结果格式化数据库返回的结果集不一定适合直接给模型看。比如TEXT字段可能有几千字JSON 字段序列化后很长。所以在把结果回传给模型之前需要做一次格式化处理。# 文件路径agent_data_demo/formatter.py追加内容 def format_result(rows: list[dict], max_rows: int 20, max_chars: int 200) - str: 把查询结果转成 JSON 文本限制行数并截断超长字段。 output_rows [] for row in rows[:max_rows]: new_row {} for key, value in row.items(): text str(value) if len(text) max_chars: text text[:max_chars] ...(截断) new_row[key] text output_rows.append(new_row) if len(rows) max_rows: output_rows.append({note: f查询结果超过 {max_rows} 行仅展示前 {max_rows} 行。}) return json.dumps(output_rows, ensure_asciiFalse, indent2, defaultstr)格式化后的结果是 JSON 字符串说明几点字段名保留原始名超长字段截断并加标记完全超过阈值的内容不会进入上下文。这样既能保证模型拿到数据又不会让上下文被大文本字段占满。4.4 Function Calling 完整循环接下来我们把连接器、查询执行器、格式化器、工具定义串起来封装成一个DataAgent。# 文件路径agent_data_demo/agent.py import json from openai import OpenAI from database import DatabaseConnector from formatter import format_result, format_schema_for_llm from query_executor import QueryExecutor from tools import TOOLS class DataAgent: def __init__(self, config: dict, max_rows: int 50): self.connector DatabaseConnector(config[database]) self.executor QueryExecutor(self.connector, limitmax_rows) self.client OpenAI( base_urlconfig[llm].get(base_url), api_keyconfig[llm][api_key], ) self.model config[llm][model] def invoke_tool(self, tool_name: str, args: dict) - str: if tool_name get_database_schema: schema self.connector.get_schema() return format_schema_for_llm(schema) if tool_name execute_readonly_query: sql args.get(sql, ) rows self.executor.execute_query(sql) return format_result(rows) return f未知工具: {tool_name} def chat(self, user_question: str, max_iterations: int 5) - str: messages [ { role: system, content: ( 你是一名数据分析助手职责是回答用户关于业务数据的查询。 你需要按以下步骤工作\n 1. 如果还不了解表结构先调用 get_database_schema 获取数据库说明\n 2. 根据表结构编写安全的 SELECT 查询调用 execute_readonly_query 获取数据\n 3. 根据查询结果回答问题。请始终引用查询得到的真实数据不要编造字段和数值。 ), }, {role: user, content: user_question}, ] for step in range(max_iterations): response self.client.chat.completions.create( modelself.model, messagesmessages, toolsTOOLS, tool_choiceauto, ) message response.choices[0].message # 没有工具调用说明模型已经可以得到最终答复 if not message.tool_calls: return message.content or messages.append(message) for tool_call in message.tool_calls: fn tool_call.function args json.loads(fn.arguments or {}) result self.invoke_tool(fn.name, args) messages.append( { role: tool, tool_call_id: tool_call.id, content: result, } ) return 已达到最大工具调用次数请尝试换一种问法。这个循环是整个 Agent 的核心逻辑并不复杂模型返回一个tool_calls数组我们逐个执行工具把结果以roletool的消息回填给模型然后继续让模型生成新回复。如果模型不再调用工具说明它已经准备好最终答案直接返回。max_iterations参数用来避免模型陷入无限循环。正常情况下一个问答 3 到 5 次工具调用就足以完成超过这个次数直接终止提示用户换一种问法。到这一步一个最小可用的数据问答 Agent 已经成型了。它具备读 Schema、写 SELECT、查结果、回答问题的完整能力。5. 数据质量与数据治理Agent 的结果上限5.1 脏数据对 Agent 有多大的影响模型的推理能力再强也架不住上游数据是脏的。举个例子。订单表的status字段存储的是TINYINT如果建表时没有注释也没有数据字典说明 0 到 3 分别是什么Agent 只能从列名和上下文猜。它可能猜 0 是“未支付”也可能猜 1 是“未支付”同一个问题换一个会话答案就不一样。再比如时间字段。有的表存DATETIME有的表存BIGINT时间戳有的存VARCHAR格式化字符串。Agent 如果没有上下文提示根本不知道该不该做时区转换。你问“昨天流水是多少”它直接拿BIGINT字段和NOW()比较结果完全错误。所以数据质量直接决定了 Agent 回答质量的上限。Agent 不是数据的生产者只是数据的搬运工和解释器。上游数据本身口径混乱下游 Agent 只能跟着一起乱。5.2 数据加工的三个原则想让 Agent 稳定地理解业务数据在数据加工阶段就要遵守三个原则。原则一字段名可读。create_ts、order_status_cd这种缩写Agent 能推断但可能出错。如果业务允许尽量使用created_at、order_status这种语义清晰的字段名。已经存在的历史字段可以在注释里补充可读说明。原则二枚举值有注释。status字段添加注释后Agent 在这个示例里就能直接读到“0待支付 1已支付 2已发货 3已完成”不用猜。原则三指标口径统一。同一个指标在不同表里的命名、单位、计算方式最好保持一致。比如金额统一用“人民币元”时间统一用DATETIME状态统一用一个枚举定义。口径不一致时需要额外提供一份指标口径说明。5.3 把关系数据加工成模型更容易读