去年接手一个 WMS 仓储系统的升级项目业务方提的需求越来越离谱——运营想要直接问系统华东仓A001库位还有多少可发库存仓管想用一句话触发盘点流程财务甚至问上周出库单里有没有异常的整单未扣减。这些需求放在传统模式下每一句都对应一张提数工单排 IT 排两三天。后来我把 AI Agent 接到了 WMS 的数据库上让大模型通过一组统一的工具函数直接摸到那 15 张核心业务表配合 19 个工具组件组成的数据访问链路这些自然语言提问才真正变成了可落地的功能。这篇就是整个实战过程的第 4 讲。我会完整拆解 15 张表的业务含义和表间关系讲清楚 19 个 AI 工具在链路里各自扮演什么角色并把让大模型摸数据库的搭建步骤、核心代码、以及生产化改造要点一次说透。适合已经在做 WMS、WCS、ERP 这类系统的开发同学也适合准备用 AI Agent 改造企业内部系统的技术负责人参考。1. 为什么大模型摸数据库比堆 RAG 知识库更实在1.1 仓储数据的真实分布15 张表背后是散落的业务事实WMS 系统不像互联网应用那样只有一张用户表和一张订单表它的业务事实分布在十几张表里而且互相咬得很紧。库存余额在 stock 表里但能不能发要去看订单占用、批次效期、库位状态入库单上有收货数量但真实入库数可能分布在多张明细里一个波次下挂着几十个任务任务状态决定拣货进度。这些业务如果靠人工去写关联查询光是搞清楚表结构就要花不少时间。更重要的是业务人员想知道的问题往往跨多张表。比如最简单的查某商品可用库存就需要把库存表、库位状态、批次效期、订单占用量、锁定量全部关联起来。这种多表 join 的需求正好是大模型擅长的——前提是它知道有哪些表、表里有什么字段、表之间怎么关联。这就引出了本讲的核心思路与其费劲把数据同步到向量库里去喂给大模型不如让大模型直接对数据库发起查询。前者是隔了一层纱的 RAG 方案后者是直接建立数据事实的连接。1.2 RAG 与大模型直连数据库的本质差异很多团队一提到让大模型懂业务第一反应就是做 RAG——把文档、报表、甚至数据库导出结果切成 chunk存进向量库回答问题时先去检索再生成。但放在 WMS 这种强实时、强状态、强逻辑的系统里RAG 有个致命问题它回答不了随时变化的库存数字。你可以把 RAG 理解成让大模型看一张昨天拍的照片而直连数据库是让大模型直接看实时监控画面。库存是动态变化的照片再高清也会过期。Agent 直连数据库则不同大模型通过工具函数生成 SQL、执行查询、拿到最新结果再组织语言回复整个过程拿到的都是当前时刻的真实数据。当然RAG 不是完全没用。我在项目里把 RAG 用来管理业务规则文本——比如波次策略说明、异常处理流程、库位编码规则这些相对静态的知识更适合走向量检索。而所有需要实时数字的问题全部走数据库直连。两条腿走路效果远好于只押注一种方案。2. 15 张表拆解WMS 数据模型其实没有想象中复杂2.1 从主数据到单据15 张表怎么分真正开始设计 Agent 要访问的数据集时我没有直接拿 WMS 线上库的全量几百张表去喂大模型而是先梳理出最核心的 15 张业务表。这个梳理过程本身就是一次业务收敛删掉日志表、配置表、中间临时表只保留能让 Agent 回答库存、出入库、波次任务、盘点这四类高频问题的表。15 张表可以分成四组基础主数据组仓库表wms_warehouse、库区表wms_area、库位表wms_location、货品表wms_product。库存与批次组批次表wms_batch、库存表wms_stock、库存流水表wms_stock_log。出入库单据组入库单wms_inbound_order、入库单明细wms_inbound_item、出库单wms_outbound_order、出库单明细wms_outbound_item。作业与盘点组波次表wms_wave、波次任务表wms_wave_task、盘点单头表wms_stocktake_header、盘点明细表wms_stocktake_item。下面是每张表的核心职责和关键字段清单。表名业务含义关键字段wms_warehouse仓库主数据id, warehouse_code, warehouse_name, statuswms_area库区主数据id, warehouse_id, area_code, area_type(存储/拣货/暂存)wms_location库位主数据id, area_id, location_code, location_type, statuswms_product货品主数据id, sku_code, sku_name, spec, unit, default_location_idwms_batch批次/效期id, product_id, batch_no, production_date, expiry_date, statuswms_stock实时库存id, product_id, location_id, batch_id, qty, available_qty, locked_qtywms_stock_log库存流水id, stock_id, change_type(入库/出库/调整), before_qty, after_qty, created_atwms_inbound_order入库单头id, order_no, warehouse_id, supplier_id, status, expected_timewms_inbound_item入库单明细id, inbound_order_id, product_id, expected_qty, actual_qtywms_outbound_order出库单头id, order_no, warehouse_id, customer_id, status, order_timewms_outbound_item出库单明细id, outbound_order_id, product_id, qty, allocated_qty, picked_qtywms_wave波次id, wave_no, outbound_order_ids, status, created_atwms_wave_task波次任务id, wave_id, location_id, product_id, qty, status(待拣/拣选中/完成)wms_stocktake_header盘点单头id, stocktake_no, warehouse_id, status, created_bywms_stocktake_item盘点明细id, stocktake_header_id, product_id, location_id, system_qty, actual_qty, diff_qty这张表过一遍基本就能回答 WMS 里 80% 的日常问题。比如某货品在不同库位的分布某出库单分配了多少量某个批次还有多少可用库存。2.2 表与表之间的主外键关系大模型要正确生成 SQL光知道有哪些表还不够还得知道表之间怎么关联。我在实际给 Agent 的上下文里给每一对关键关系都写成了名词解释而不是让模型去看外键约束。最核心的关系库存表是事实中心它同时关联货品、库位、批次一条库存记录指代某个 SKU 在某个库位上的某个批次目前有多少数量。单据体系是围绕订单的扩展。入库单头对明细是 1:N明细里记录预期收货数和实际收货数实际数回写库存表。出库单和波次是多对多关系。一个波次可以包含多个出库单一个出库单也可以被拆到多个波次里波次任务表是执行层记录每个库位每个 SKU 要拣多少。盘点表是独立的作业流盘点明细记录系统账面数和实际盘点数差异数直接驱动库存调整。这些关系在人工写 SQL 时靠经验就能自动带上但大模型不是靠经验是靠上下文描述。所以我在后面的代码实践里会把这些关联关系做成注释文本和表结构描述一起交给 Agent。2.3 给 Agent 用的 Schema 设计和给人用的 ER 图不一样这里有个容易忽略的关键点给大模型看的表结构不能直接丢一张 ER 图或者 information_schema 里导出的建表 SQL。原因有两个。第一字段太多会浪费 token。线上表的字段动辄二十个但 Agent 高频使用的可能只有其中六七个。每次请求都把全字段塞进上下文既贵又慢。第二大模型对字段注释的依赖远高于人。人看到 qty 能猜到是数量但模型不一定能分清 qty 和 available_qty 哪个是可发数哪个是物理数。给字段写清晰的中文注释比给字段取一个聪明的英文名更有效。因此我实际采用的是视图 精简元数据描述的组合方案在数据库层建了若干面向查询的视图比如 v_stock_overview 就已经把库存、货品、库位、批次的关键信息 join 好了Agent 只需对这个视图做条件过滤而不是在 15 张表里做复杂关联。这个设计一方面降低了 SQL 生成难度另一方面也天然做了权限收敛——视图里没有的敏感字段Agent 永远看不到。3. 19 个工具的分工表一条 Agent 查询背后到底站着哪些组件3.1 为什么是 19 个工具让大模型摸到数据库这句话听起来很轻巧实际拆开落地涉及的不只是调用一个大模型的 API 那么简单。从用户提问到 SQL 执行再回到自然语言回答中间要经过模型层、Agent 框架层、协议接入层、数据访问层、数据库层、运维监控层、前端展示层。每一层都需要至少一个趁手的工具。我在项目里整理出一条稳定的工具链总共 19 个。它们不是 AI 工具列表的堆砌而是沿着一次查询请求的完整生命周期倒推出来的。下表就是这 19 个工具的完整分工。编号工具/组件定位在项目里负责什么1DeepSeek大模型推理Agent 的大脑负责把自然语言理解成工具调用2通义千问 Qwen大模型备选本地化部署、低成本高频查询时的兜底模型3MCP模型上下文协议统一 Agent 与数据库工具之间的调用标准和安全边界4Function Calling模型能力让模型按结构化参数调用 query_stock 这类函数5LangChainAgent 编排框架管理 Agent 循环、工具注册、历史记忆6LlamaIndex元数据索引管理表结构描述、业务规则等 long-tail 上下文7Dify低代码平台快速搭建给业务人员试用的对话界面和工作流8SQLAlchemyORM 连接层统一管理数据库连接、会话、事务9pymysqlMySQL 驱动SQLAlchemy 底层实际执行 SQL 的驱动10VannaText2SQL 专用引擎通过训练专用模型提高 SQL 生成准确率11MySQL 8.0数据库承载 15 张核心业务表12Navicat数据库客户端人工核数、写复杂校验脚本、排查脏数据13DataGrip数据库 IDE查看执行计划、调优慢 SQL14Docker容器化一键拉起 MySQL、Redis、Agent 服务环境15Redis缓存与限流缓存表结构元数据、限制 Agent 并发查询16Grafana监控观测 Agent 调用量、SQL 执行耗时、错误率17Gradio快速 UI给内部用户做 Demo 用的对话界面18Git版本管理管理 Prompt 模板、工具函数代码变更19Postman接口调试调试 Agent HTTP 接口、验证输出格式这张表基本反映了我当时实际工作的全部工具面。需要说明的是这里面的AI 工具不只是传统意义上的一两个 AI 应用而是一整套能让大模型在真实生产环境里干活的基础设施。3.2 一条查询请求在工具链里的实际流转只看表格还是不够直观。我从实际运行中抽一条请求完整走一遍工具链。运营在 Gradio 界面上输入华东仓的 A001 库位批号为 20250101 的商品还有多少可用库存Gradio 拿到文本转成 HTTP 请求发给 Dify 构建的 Agent 服务。LangChain 的 Agent 循环接管任务从 LlamaIndex 管理的业务元数据中检索出库存相关的表结构和库位编码规则。模型层DeepSeek根据上下文决定调用 query_stock 这个工具参数是 warehouse_code华东仓、location_codeA001、batch_no20250101。Function Calling 把结构化参数传给工具执行器工具函数内部通过 SQLAlchemy 打开 MySQL 连接。SQL 在视图 v_stock_overview 上执行结果集经 pymysql 返回工具把结果整理成 JSON。Redis 缓存表结构元数据避免每次请求都去 information_schema 现查元数据。模型把 JSON 结果转成自然语言Gradio 刷新对话界面。Grafana 记录这轮请求的耗时和 SQL 执行时间。这一整条链路每一环掉链子都会导致摸不到数据库或者摸错了数据。我自己在第一次联调时就踩过工具之间参数格式不一致的坑后面第 5 节会详细展开。3.3 用 MCP 还是裸 Function Calling这里必须展开聊一下 MCP 和 Function Calling 的取舍这是很多第一次做 Agent 接数据库的人会纠结的点。Function Calling 是模型能力大模型在输出时如果判断需要查数据会输出一段结构化的 JSON声明要调用哪个函数、传什么参数。它的优点是简单框架里注册一下就行缺点是每个系统各写各的工具分散在各个工程里安全边界和权限控制完全靠开发自觉。MCP 则是一个统一协议。你可以把它理解成给外部工具装上了统一的USB-C 接口。数据库、文件系统、第三方 API只要实现了 MCP ServerAgent 就能用标准方式去访问而不需要针对每个系统写一遍集成代码。我的实际选择是两者都用。MCP 负责连接底层数据资源比如把 15 张表的查询封装成一个 MCP ServerFunction Calling 负责在 Agent 循环里做即时调用决策。MCP 解决了连接标准问题Function Calling 解决了调用效率问题。项目跑起来之后这套组合的可维护性明显比纯 Function Calling 高。4. 实战链路从查库存到第一个可跑的 Agent4.1 环境准备与初始数据实践的前提是先有一套能跑的 WMS 表结构。我建议用 Docker 起一个 MySQL 8.0 容器把 15 张表和少量种子数据装进去。这样后面无论怎么折腾一条命令就能恢复环境。docker run -d --name wms-mysql \ -p 3306:3306 \ -e MYSQL_ROOT_PASSWORDwms123456 \ -e MYSQL_DATABASEwms_ai \ mysql:8.0建表时注意三点每张表都加上 status 状态字段方便 Agent 查询时过滤已作废数据所有时间字段统一用 datetime并且强制 UTC 存储前端展示时再转本地时区关键业务表都建立联合索引尤其是库存表上 (product_id, location_id, batch_id) 的组合索引这是高频查询路径。Python 依赖安装pip install langchain langchain-openai mcp sqlalchemy pymysql gradio openai4.2 暴露给 Agent 的不是裸表而是业务视图我在第 2 节里强调过视图的价值这里直接给出实践。与其让 Agent 在 15 张表里自行 join不如把最高频的查询场景直接做成视图。以下是我在项目里最先落地的一个核心视图。CREATE VIEW v_stock_overview AS SELECT w.warehouse_code, w.warehouse_name, a.area_code, l.location_code, p.sku_code, p.sku_name, p.unit, b.batch_no, b.expiry_date, s.qty AS physical_qty, s.available_qty, s.locked_qty FROM wms_stock s JOIN wms_product p ON s.product_id p.id JOIN wms_location l ON s.location_id l.id JOIN wms_area a ON l.area_id a.id JOIN wms_warehouse w ON a.warehouse_id w.id LEFT JOIN wms_batch b ON s.batch_id b.id WHERE s.qty 0;这个视图做了一件事把哪个仓、哪个库区、哪个库位、哪个 SKU、哪个批次、多少库存完整拉平。Agent 以后只需要对这个视图过滤而不用关心 15 张表之间怎么关联。视图相比裸表的好处我在使用中体会非常明显大幅降低 SQL 生成难度模型只需要写 WHERE 条件不需要写 join。天然收敛字段视图里没有的敏感数据模型想查也查不到。查询计划可控避免模型生成跨多表的大 join 导致性能问题。4.3 核心代码Agent 查询库存的完整实现现在进入整个项目的核心代码段。我写了一个 query_stock 工具它接收仓库、库位、SKU 等条件在 v_stock_overview 视图上查询返回结构化结果。import os import json from sqlalchemy import create_engine, text from langchain_openai import ChatOpenAI from langchain.agents import initialize_agent, AgentType DB_URI os.getenv(DB_URI, mysqlpymysql://root:wms123456127.0.0.1:3306/wms_ai) engine create_engine(DB_URI, pool_size10, pool_recycle3600) def query_stock(warehouse_code: str None, location_code: str None, sku_code: str None, batch_no: str None) - str: 查询实时库存。参数均可选至少提供一个过滤条件。 conditions [] params {} if warehouse_code: conditions.append(warehouse_code :warehouse_code) params[warehouse_code] warehouse_code if location_code: conditions.append(location_code :location_code) params[location_code] location_code if sku_code: conditions.append(sku_code :sku_code) params[sku_code] sku_code if batch_no: conditions.append(batch_no :batch_no) params[batch_no] batch_no if not conditions: return 至少需要一个查询条件 sql SELECT * FROM v_stock_overview WHERE AND .join(conditions) LIMIT 50 with engine.connect() as conn: rows conn.execute(text(sql), params).mappings().all() if not rows: return 未查询到符合条件的库存记录 return json.dumps([dict(r) for r in rows], ensure_asciiFalse, defaultstr)这里有两个关键设计决策所有参数用绑定变量而不是把用户输入直接拼进 SQL从源头防止注入问题。返回结果用 json.dumps 的 defaultstr 处理避免 Decimal、datetime 这类类型序列化失败。工具函数写好之后把它注册到 LangChain 的 Agent 里。tools [ { name: query_stock, func: query_stock, description: ( 查询 WMS 实时库存。支持按仓库、库位、SKU、批次筛选。 当用户问‘某仓某库位某商品还有多少库存’时使用。 ) } ] llm ChatOpenAI( modeldeepseek-chat, api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com/v1, temperature0 ) agent initialize_agent( toolstools, llmllm, agentAgentType.ZERO_SHOT_REACT_DESCRIPTION, verboseTrue )注册工具时description 的写法很有讲究。大模型是靠这段文字来判断什么时候该调这个工具所以描述里要写清触发场景而不是罗列参数。我原来写的是查询库存表模型经常不知道该不该调改成当用户问某仓某库位某商品还有多少库存时使用之后命中率明显提升。跑一个真实问题result agent.invoke(华东仓 A001 库位的 SKU1001 可用库存是多少) print(result[output])模型内部会先判断这是一个库存查询问题然后调用 query_stock(warehouse_code华东仓, location_codeA001, sku_codeSKU1001)拿到 JSON 后整理成自然语言回答。整个过程不需要任何人去手工查表。4.4 让 Agent 执行写操作的正确姿势先确认再执行只读查询跑通之后很多人会想更进一步让 Agent 直接做库存调整、单据修改。这个想法很危险因为 WMS 的写操作往往牵一发而动全身一个错误的 UPDATE 可能让账面库存和实物库存差出十万八千里。我的方案是写操作永远走审批。Agent 负责生成变更建议和待执行的 SQL但不直接执行。比如运营说把 A001 库位 SKU1001 的库存调整到 100Agent 会做两件事先查当前实际库存再生成一条 UPDATE 语句然后把原值、新值、变更原因、执行人拼成一条审核消息推送到企业微信或者管理后台等仓库主管点确认后由独立的写接口执行。这个设计等于把 Agent 的写权限降级成了建议权限。真实生产环境里机器的判断力再强也要给人留一道最后的确认闸门。5. 最容易翻车的 5 个环节踩坑记录与修复方案5.1 Token 爆炸15 张表的元数据把上下文塞满了第一个坑来得很快。刚开始我把 15 张表的建表 SQL 全部塞进系统提示词里让大模型充分理解表结构结果第一轮请求就把上下文塞到了接近窗口上限。查询本身很简单但模型的注意力被几万个 token 的字段描述占满反而频繁生成语法错误的 SQL。后来我做了三件事表结构描述只保留字段名、类型、注释去掉默认值、字符集、索引等与查询无关的信息高频查询全部走视图让元数据量降低一半以上表结构元数据存进 Redis按天刷新而不是每次请求都现场拼装。这一套组合下来单次查询的 token 消耗下降了大约一半。5.2 幻觉字段名模型把 product_id 写成 productID大模型对字段名的记忆是语义级别的不是精确的字符串级别。它可能记住商品对应字段是 product_id生成 SQL 时却写成 productID 或者 productid这在 MySQL 里直接报错。一开始我想靠给模型更完整的字段说明来解决发现治标不治本。后来在工具函数里加了一层字段白名单校验所有查询参数在进入 SQL 之前先与视图的实际字段列表比对一次不在白名单里的字段直接拦截并提示模型参考字段规范重新生成。这样就杜绝了绝大部分字段名幻觉问题。5.3 权限失控Agent 生成了 DELETE 语句这是我在测试中真实遇到过的情况。运营在对话里问能不能把这个库位的库存删掉我要重新导入模型理解成了删除库存记录直接生成了一条 DELETE 语句。虽然我当时立刻发现了问题但这足以给整个项目敲响警钟。从那以后Agent 连接数据库的账号一律改成只读权限除非明确执行写操作审批流程否则任何 INSERT、UPDATE、DELETE、DDL 都会被数据库账号权限挡在门外。这个底线必须尽早建立不能指望模型每一次都理解对。5.4 超时与连接池耗尽业务高峰期一查就卡死WMS 的高峰期很典型——波次释放、盘点录入、出库分配全挤在同一个时间段。此时 Agent 如果同时响应多个查询请求连接池很容易被打满数据库 RT 飙升甚至拖垮正常业务。我的处理手段分成两个层面数据库层面为 Agent 查询走单独的只读副本主库不承担这些分析型查询的压力应用层面给查询设置 statement timeout超过 3 秒直接中断再通过 Redis 对 Agent 的并发请求做限流。宁可让用户等一下也不能让 Agent 把 WMS 主业务拖死。5.5 数据精度和时区问题库存数量对不上WMS 里的库存数量在很多场景下是 Decimal 类型大模型工具返回 JSON 时如果直接做 float 转换可能出现 0.29 变成 0.29000000000000004 这种精度问题时间字段如果不统一时区同一个订单的创建时间在数据库里显示 08:00前端显示 16:00就会产生严重误导。这个问题相对容易修但容易漏。我的经验是统一所有时间字段按 UTC 存储展示层再按用户时区转换所有数值型字段在工具层一律转成字符串返回不做隐式类型转换。这两条规范看起来不起眼恰恰是让大模型摸数据库能够长期稳定运转的地基。6. 生产化改造权限、审计与回退机制6.1 数据库账号分级Agent 永远拿不到管理员权限从 Demo 到生产第一件事不是优化性能而是改权限模型。我在数据库里建了三个账号等级Agent 只读账号、Agent 写操作账号、管理账号。Agent 日常运行只持有只读账号连接串放在独立的配置中心里不写死在代码中审批通过的写操作由独立服务使用写操作账号执行管理账号只存在运维手里Agent 和业务系统都接触不到。这套分级模型配合视图使用等于给 Agent 加了两把锁视图决定了能看哪些列账号权限决定了能执行哪些操作。即使模型被恶意诱导生成了超范围的 SQL数据库层的权限也会把它拦下来。6.2 审计每一次摸数据库都要留下痕迹企业内部系统接入 AI Agent合规审计是绕不开的一环。我的做法是做一个独立的审计表记录每一次 Agent 请求的完整链路信息。审计表字段包括user_id谁在提问、user_prompt原始问题、generated_sql模型生成的 SQL、tool_name调用了哪个工具、execute_result_status执行成功还是失败、latency_ms执行耗时、created_at请求时间。这份审计日志的价值在前期看不出来等真正出了数据问题比如某笔库存对不上账、某张出库单被异常锁单它就是排查的第一手证据。我还会定时把审计日志导入数仓做分析反过来优化 Agent 的 Prompt 和工具描述。6.3 熔断与回退慢 SQL 不能拖垮核心业务最后一个生产化要素是熔断机制。Agent 的查询再智能也难免会出现索引没命中导致的慢 SQL。我配置了三个维度的兜底策略单条 SQL 执行超过 3 秒自动终止Agent 连续失败超过 5 次时自动进入只读降级模式不再响应任何请求每日高峰时段对 Agent 的查询并发量按阈值限流。这套机制上线后最明显的效果是再也没有出现过 Agent 查询拖垮 WMS 主服务的事故。回退策略也要提前想好。我的方案是给 Agent 服务保留一键下线开关一旦出现异常运维可以立刻把流量切回人工查询模式让业务方用原来的界面继续工作。AI 能力的加入应该是锦上添花而不是不可替代的唯一通道。写在最后的一些实际体会项目上线到现在我最深的感受是让大模型摸数据库真正的难点从来不在连上数据库这一步而在于如何划定边界。表结构要收敛成它该知道的视图权限要收敛成它最高只读的角色写操作要收敛成必须人工确认的审批流程每一步本质上都是在回答同一个问题——我们到底允许这个 AI Agent 在多大范围内替人做判断。目前这个架构已经在往两个方向扩展一个是智能补货让 Agent 实时盯库存水位和效期批次出现低于安全库存时主动生成补货建议另一个是波次策略推荐让 Agent 根据订单结构、库位热力、人工任务量推荐最优的波次组合方式。这些功能的共同点都是建立在Agent 能实时摸到 15 张表、能读懂业务字段含义这个基础之上。如果你正准备在内部系统里接入 AI Agent我建议从一个小切口开始——比如先做库存查询这一个工具把模型、框架、数据库连接、权限、审计这一整套链路跑通再逐步扩展工具数量和业务覆盖范围。一口吃不成胖子AI Agent 落地也是这样。