1. 项目概述与整体设计思路1.1 这个问数项目智能体到底要解决什么问题先聊聊我为什么要做这个系列。上一篇文章里我讲了LCODER问数项目智能体的整体架构设计当时有不少朋友留言问基础设施部分到底怎么搭、怎么选型、踩了哪些坑。这篇就专门把基础设施搭建这部分拆开揉碎了讲。所谓问数项目智能体说白了就是让用户用自然语言提问Agent自动理解问题、拆解任务、生成查询逻辑、跑到数据库里取数再把结果用人类能看懂的方式返回。比如业务人员输入上个月华东区各产品的销售额排名Agent内部要做的事情远不止查个表这么简单先要识别出时间范围上个月、地区维度华东区、业务指标销售额、分组维度产品然后去元数据里找到对应的表和字段生成正确的聚合逻辑执行查询最后还得把数字组织成一段人话回答。这个过程中基础设施决定了Agent的稳定性、扩展性和可维护性。我见过太多项目死在第一步代码写了一大堆但环境管理混乱、配置写死在代码里、数据库连接靠人肉维护结果模型一换、数据源一加整个系统就崩了。所以基础设施搭建真不是简单的装个环境它直接决定了后面的开发效率和生产稳定性。1.2 基础设施设计的三个核心理念我在搭建这套基础设施时给自己定了三条必须遵守的原则第一模块化隔离。模型接入、数据库连接、Agent编排、工具调用、记忆存储每一块都必须是独立的模块互相之间只通过接口通信。这样做的好处是后续想换模型厂商、加一个新的数据源只需要改一个模块其他部分完全不用动。第二配置与代码分离。所有环境相关的配置包括数据库连接串、模型API地址、密钥、超时时间全部走环境变量和配置文件绝不硬编码在代码里。团队协作时每个人本地一套配置提交代码不冲突。第三可观测性优先。基础设施搭建阶段就要把日志、监控、追踪这三件事纳入设计而不是等功能写完了再补。Agent类应用有个特点它的行为是动态的同一个问题每次走的链路可能不一样。没有完整的日志链路出了问题真的就是大海捞针。这三条原则后面所有的基础设施代码都是围绕它们展开的。我见过太多项目一开始图省事结果后面付出十倍代价返工这个坑大家真的别再踩了。2. 技术选型与核心依赖解析2.1 语言与框架为什么选 Python FastAPI问数Agent的基础设施我最终选了 Python 3.11 FastAPI 这套组合。先说语言Python 在 AI 生态里的统治地位是无可争议的无论是模型调用 SDK、数据处理库、还是各种 Agent 框架都是 Python 支持最好、资料最全。用 Python 来搭 Agent 基础设施后续集成任何 AI 组件都会顺畅得多。FastAPI 作为 Web 框架最大的优势是性能和开发效率的平衡。它基于 ASGI支持异步比 Flask 更适合做需要并发处理的服务。问数场景下用户提问之后 Agent 内部要经历理解问题、编排任务、调用工具、推理总结多个环节中间有大量 IO 等待异步编程模型能显著提升并发吞吐能力。另外 FastAPI 自带 OpenAPI 文档对调试前后端联调太友好了。我在开发阶段基本靠http://localhost:8000/docs这个页面就能完成所有接口测试不需要额外安装 Postman 之类的工具。2.2 Agent 编排框架自己写还是用现成的这个可能是大家最纠结的问题。我直接说结论现阶段选型优先考虑 LangChain 或 LlamaIndex 这类成熟框架除非你有极强的定制需求否则别从零开始写 Agent 编排。我最终选的是 LangChain 作为 Agent 编排基础原因有三个一是工具调用生态成熟。问数场景里最关键的一环是让模型学会调用数据库查询工具LangChain 对 OpenAI 兼容接口的 function calling 支持很完善而且支持自定义工具类的绑定和参数校验。二是抽象层次合适。它的chain、agent、tool抽象能让我快速把意图识别、工具调用、结果总结这个流程搭出来而不是把精力花在底层提示词拼接和消息轮次管理上。三是对接成本低。不管是接各种模型还是各种向量库LangChain 都提供了统一接口。基础设施阶段把这种兼容性建立好后面切换任何组件都只是改配置的事。当然LangChain 不是没有缺点抽象层多、调试起来略绕、版本升级快导致老代码容易废。但这些问题在基础设施阶段可以靠封装一层自己的接口来解决。我后面详讲。2.3 数据库与元数据管理问数项目的基础设施核心之一就是数据库连接和元数据管理。我这边业务数据主要存在 MySQL 8.0 里同时还有一部分数仓表在 PostgreSQL 中所以基础设施层必须具备多数据源支持能力。在 Python 生态里SQLAlchemy 是数据库访问层的事实标准。我选择用它来做统一的数据访问层而不是直连各种驱动。原因很简单SQLAlchemy 提供了统一的 ORM 和 Core 两层抽象切换数据库引擎时业务代码几乎不用改。对一个要对接多套数据源的中台型问数服务来说这个特性太重要了。元数据管理这块我自己实现了一个轻量级的元数据中心核心就是定期扫描数据库的信息模式把库名、表名、字段名、字段类型、字段注释、表注释、主键、外键这些信息抽取出来存到服务内部的元数据仓库中。这个仓库负责给 Agent 的找表和找字段环节提供依据。2.4 模型接入层设计问数项目最重要的是模型要会调用工具、会做推理所以模型接入层是基础设施中极其重要的一环。我采用 OpenAI 兼容的接口规范来做模型接入层主要原因是兼容性最好而且社区生态最丰富。具体配置上通过环境变量传入模型服务的 base_url、api_key、模型名称三项。如果接的是 OpenAI 官方就填官方的地址如果接的是国产模型厂商大多数也提供了兼容 OpenAI 的接口地址配置上只是改几个环境变量的事。超时控制和重试机制必须建好。模型接口的不稳定是常态所以在基础设施层就要做好三件事连接超时设置connect timeout、读取超时设置read timeout、以及失败重试策略。我在重试策略上采用的是指数退避加抖动避免重试风暴打垮模型服务。3. 项目骨架搭建与核心代码实现3.1 项目目录结构规划基础设施搭建的第一步就是规划好项目骨架。我这里分享一个经过多轮迭代后稳定下来的目录结构它按照模块化 按业务域分层的思路组织lcoder-askdata/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── config.py # 配置中心 │ ├── api/ # HTTP 接口层 │ │ ├── __init__.py │ │ └── v1/ │ │ ├── router.py # v1 路由聚合 │ │ ├── ask.py # 问答接口 │ │ └── health.py # 健康检查接口 │ ├── core/ # 核心业务逻辑层 │ │ ├── __init__.py │ │ ├── agent/ │ │ │ ├── __init__.py │ │ │ ├── chain.py # Agent 链路编排 │ │ │ ├── prompts.py # 提示词管理 │ │ │ └── tools.py # 工具集定义 │ │ ├── llm/ │ │ │ ├── __init__.py │ │ │ ├── client.py # 模型客户端 │ │ │ └── schemas.py # 模型输入输出结构 │ │ ├── database/ │ │ │ ├── __init__.py │ │ │ ├── engine.py # SQLAlchemy 引擎 │ │ │ └── session.py # 会话管理 │ │ └── metadata/ │ │ ├── __init__.py │ │ ├── models.py # 元数据模型 │ │ └── collector.py # 元数据采集器 │ ├── schemas/ # Pydantic 模型 │ │ ├── __init__.py │ │ └── ask.py │ └── utils/ │ ├── __init__.py │ ├── logger.py # 日志配置 │ └── tracing.py # 链路追踪工具 ├── tests/ # 单元测试与集成测试 ├── .env.example # 环境变量示例 ├── pyproject.toml # 项目依赖管理 └── README.md这个结构看起来简单但每一层都有明确的职责边界API 层只做参数校验和响应封装core 层专注于业务逻辑utils 层放横切关注点。agent 目录单独拎出来因为它本身就是问数项目的核心域。3.2 依赖管理与环境配置依赖管理我用的pyproject.toml这是目前 Python 社区推荐的现代方案。相比传统的 requirements.txt它支持更清晰的依赖分组和版本锁定。核心依赖如下[project] name lcoder-askdata version 0.1.0 requires-python 3.11 dependencies [ fastapi0.115.0, uvicorn[standard]0.30.0, pydantic2.7.0, pydantic-settings2.3.0, sqlalchemy2.0.30, pymysql1.1.1, psycopg2-binary2.9.9, langchain0.3.0, langchain-openai0.2.0, openai1.40.0, python-dotenv1.0.1, loguru0.7.2, tenacity8.5.0 ] [project.optional-dependencies] dev [ pytest8.2.0, ruff0.5.0, mypy1.10.0 ]版本这里我要特别提醒一句LangChain 的版本迭代非常快0.1 和 0.3 的 API 差别很大。我上面的版本是实测可用的版本组合大家如果下载时已经出了更高的版本务必先看 changelog不要盲目升级。环境变量这块我建了一个.env.example作为模板提交到仓库真正的.env文件通过.gitignore忽略确保密钥不泄露# 模型服务配置 LLM_BASE_URLhttps://api.example.com/v1 LLM_API_KEYyour-api-key-here LLM_MODELgpt-4o-mini LLM_TEMPERATURE0.1 LLM_TIMEOUT60 # 数据库配置 MYSQL_HOSTlocalhost MYSQL_PORT3306 MYSQL_USERreadonly_user MYSQL_PASSWORDyour-db-password MYSQL_DATABASEbusiness_db # 向量库配置 VECTOR_STORE_HOSTlocalhost VECTOR_STORE_PORT19530 VECTOR_STORE_COLLECTIONtable_metadata_chunks # 服务配置 SERVICE_HOST0.0.0.0 SERVICE_PORT8000 LOG_LEVELINFO ENABLE_TRACINGtrue配置文件加载使用 pydantic-settings它能在启动时自动从环境变量和.env文件加载配置并进行类型校验。这样启动服务时如果有配置缺失或格式错误会立刻报错而不是等到运行到某个环节才炸。3.3 日志与链路追踪的搭建问数 Agent 这种应用日志真的不能随便搞搞就完事。我用的日志库是 loguru相比标准库 logging配置起来省心太多开箱即用就能输出带颜色、带时间、带调用位置的结构化日志。最关键的是要建立链路追踪机制。一个完整的问答请求需要经历多个环节如果每个环节的日志彼此孤立一旦出问题根本拼不出完整的请求链路。我在基础设施层做了一个非常轻量的 trace_id 机制# app/utils/tracing.py import contextvars import uuid # 使用 contextvars 实现在异步环境下传递 trace_id _trace_id_var: contextvars.ContextVar[str] contextvars.ContextVar(trace_id, default) def new_trace_id() - str: 生成新的 trace_id return uuid.uuid4().hex def set_trace_id(trace_id: str None): 设置当前上下文的 trace_id如果没有传入则自动生成 if not trace_id: trace_id new_trace_id() _trace_id_var.set(trace_id) return trace_id def get_trace_id() - str: 获取当前上下文的 trace_id return _trace_id_var.get()然后在日志输出格式里把 trace_id 加进去Loguru 可以通过 filter 或直接绑定到 context 中实现。这样查询日志时只要拿到一个 trace_id整条链路的运行轨迹全部能串起来。链路追踪的埋点位置我建议覆盖这些环节HTTP 请求入口、模型调用前后、工具调用前后、数据库查询前后、Agent 每一步决策。这些埋点在基础设施阶段就做好后面做性能分析和问题定位会轻松太多。3.4 FastAPI 入口与中间件配置入口文件main.py是整个服务启动的枢纽。除了初始化 FastAPI 实例还需要注册中间件、挂载路由、配置异常处理。这段代码把基础设施的架子撑起来# app/main.py from contextlib import asynccontextmanager from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from app.config import get_settings from app.api.v1.router import api_router from app.utils.logger import setup_logger from app.utils.tracing import set_trace_id settings get_settings() logger setup_logger() asynccontextmanager async def lifespan(app: FastAPI): 服务启动和关闭时的资源管理 logger.info(LCODER AskData Agent 服务启动中...) # 这里做资源初始化数据库连接池预热、模型客户端验证、元数据加载 yield logger.info(LCODER AskData Agent 服务已关闭) app FastAPI( titleLCODER AskData Agent Service, description问数项目智能体 - 基础设施版本, version0.1.0, lifespanlifespan, ) # 跨域配置前后端分离部署时必须设置 app.add_middleware( CORSMiddleware, allow_originssettings.CORS_ORIGINS, allow_credentialsTrue, allow_methods[*], allow_headers[*], ) app.middleware(http) async def add_trace_id_middleware(request: Request, call_next): 为每个 HTTP 请求生成并绑定 trace_id trace_id request.headers.get(X-Trace-ID) set_trace_id(trace_id) response await call_next(request) response.headers[X-Trace-ID] get_trace_id() return response app.include_router(api_router, prefix/api/v1)注意 lifespan 里的资源初始化这一段在基础设施阶段我会在这里做数据库连接池预热、模型客户端可用性检查、元数据加载。这样每个请求进来时核心资源已经就绪不会出现第一个请求因为连接池尚未建立而超时的尴尬情况。4. 数据库连接层与元数据中心4.1 多数据源连接池配置问数项目面对的数据源往往不止一个。我这边生产环境就有 MySQL 和 PostgreSQL 两套分别承载不同的业务域。SQLAlchemy 的引擎创建逻辑我封装成了一个独立的模块核心代码如下# app/core/database/engine.py from sqlalchemy import create_engine from sqlalchemy.engine import Engine from sqlalchemy.pool import QueuePool from app.config import get_settings settings get_settings() def create_db_engine( db_type: str, host: str, port: int, user: str, password: str, database: str ) - Engine: 根据数据库类型创建 SQLAlchemy 引擎配置连接池 if db_type mysql: url fmysqlpymysql://{user}:{password}{host}:{port}/{database}?charsetutf8mb4 elif db_type postgresql: url fpostgresqlpsycopg2://{user}:{password}{host}:{port}/{database} else: raise ValueError(f不支持的数据库类型: {db_type}) engine create_engine( url, poolclassQueuePool, pool_size10, # 连接池大小 max_overflow20, # 连接池溢出上限 pool_timeout30, # 获取连接超时时间 pool_recycle1800, # 连接回收周期避免 MySQL 8 小时断开问题 pool_pre_pingTrue, # 使用前预检连接有效性 echoFalse, connect_args{connect_timeout: 10}, ) return engine这里面有几个参数需要特别说明。pool_pre_pingTrue是我强烈建议开启的它会在每次从连接池取出连接前先发一个轻量的探测语句如果连接已经失效就自动丢弃并创建新连接。这个配置能避免 MySQL 服务端因为 wait_timeout 把空闲连接断掉之后应用拿到半死连接然后报错的问题。pool_recycle1800对应的是 MySQL 默认的 8 小时超时限制。虽然前面有 pre_ping 兜底但定期回收连接可以从根本上减少无效连接的数量。4.2 元数据采集器的实现元数据是问数 Agent 的眼睛。如果 Agent 不知道数据库里有哪些表、哪些字段、字段是什么含义它就像一个盲人在摸象生成查询全靠蒙。所以元数据中心是基础设施里绝对不能省的部分。我的实现思路是在服务启动时通过 SQLAlchemy 引擎连接各数据源执行系统表查询把表结构信息批量拉取到本地存储。MySQL 和 PostgreSQL 的系统表查询语句不同我在采集器里分别处理# app/core/metadata/collector.py from sqlalchemy import text from sqlalchemy.engine import Engine class MetadataCollector: 元数据采集器负责扫描数据库结构信息 def collect_from_mysql(self, engine: Engine) - list[dict]: 采集 MySQL 数据库的元数据 sql SELECT t.TABLE_NAME, t.TABLE_COMMENT, c.COLUMN_NAME, c.COLUMN_TYPE, c.IS_NULLABLE, c.COLUMN_COMMENT FROM information_schema.TABLES t LEFT JOIN information_schema.COLUMNS c ON t.TABLE_NAME c.TABLE_NAME WHERE t.TABLE_SCHEMA DATABASE() ORDER BY t.TABLE_NAME, c.ORDINAL_POSITION with engine.connect() as conn: result conn.execute(text(sql)) rows result.fetchall() # 按表名聚合字段信息 tables {} for row in rows: table_name row[0] if table_name not in tables: tables[table_name] { table_name: table_name, table_comment: row[1], columns: [] } tables[table_name][columns].append({ column_name: row[2], column_type: row[3], is_nullable: row[4], column_comment: row[5], }) return list(tables.values())采集到元数据之后我会做两件额外的事情。第一对字段名做标准化处理。比如user_id、userId、用户ID这种种写法统一映射成标准语义方便 Agent 识别。第二生成一份数据字典提示词摘要。这一步是问数项目非常关键的小技巧把表注释和字段注释按照固定的模板组织成文本例如表sales_order销售订单表字段order_id订单IDorder_amount订单金额region区域...。这份摘要会作为上下文注入到 Agent 的提示词中让模型在生成查询时知道有哪些表和字段可以用。4.3 只读账号与安全边界问数项目有一个必须坚守的安全底线Agent 在查询业务数据库时必须使用只读账号绝不能用有写入权限的账号。我在基础设施里专门创建了一个低权限的数据库账号。MySQL 侧的授权语句类似这样-- 创建只读查询账号 CREATE USER askdata_ro% IDENTIFIED BY 强密码; GRANT SELECT ON business_db.* TO askdata_ro%; FLUSH PRIVILEGES;这样做的好处是显而易见的即便 Agent 在生成查询时出现了幻觉生成了 INSERT、UPDATE、DELETE 甚至 DROP 语句数据库也会因为权限不足而直接拒绝执行。这是一道物理层面的保底防线比任何代码层面的校验都可靠。另外我在 SQLAlchemy 引擎层还做了一个补充拦截通过事件监听器检查执行语句的前缀如果发现非 SELECT 开头的语句直接抛异常终止查询。这属于第二道防线防止某些特殊情况下权限配置被绕过。5. 模型接入层与 Agent 客户端构建5.1 统一模型客户端封装模型接入层是 Agent 能力的基础。我封装了一个统一的模型客户端目的是让 core 层不直接依赖某个具体模型的 SDK而是依赖我自己定义的接口。# app/core/llm/client.py from typing import Optional from openai import AsyncOpenAI from app.config import get_settings settings get_settings() class LLMClient: 统一的模型调用客户端支持 OpenAI 兼容接口 def __init__(self): self.client AsyncOpenAI( base_urlsettings.LLM_BASE_URL, api_keysettings.LLM_API_KEY, timeoutsettings.LLM_TIMEOUT, max_retries3, ) self.model settings.LLM_MODEL self.temperature settings.LLM_TEMPERATURE async def chat_completion( self, messages: list[dict], tools: Optional[list[dict]] None, tool_choice: Optional[str] None, temperature: Optional[float] None, ) - dict: 统一的对话补全接口 Args: messages: OpenAI 格式的消息列表 tools: function calling 工具定义列表 tool_choice: 工具选择策略auto 或 none temperature: 采样温度不传则使用默认配置 Returns: 模型返回的完整响应包含 content 和 tool_calls response await self.client.chat.completions.create( modelself.model, messagesmessages, toolstools, tool_choicetool_choice, temperaturetemperature if temperature is not None else self.temperature, ) return response调模型时 temperature 这个参数在问数场景里必须调低。我设置的默认值是 0.1目的是让模型的输出尽量确定和稳定。问数这种数据分析场景需要的是尽量少幻觉、少自由发挥如果你把 temperature 设成 0.7 甚至更高模型可能每次给的查询逻辑都不一样这个对业务来说是不可接受的。max_retries3是 OpenAI SDK 内置的重试策略它会在遇到连接错误、超时、5xx 错误时自动重试。但在服务端这种高并发场景我建议把它调成 1 或者 2重试次数太多会加剧模型服务的压力尤其是在模型服务本身已经过载的情况下。5.2 Agent Skill 的落地方式热词里有个概念叫 AI Agent Skill在问数项目里它指的是 Agent 能够调用的一组技能或者能力单元。基础设施阶段我主要做了三个技能的基础封装。第一个是查表技能。接收自然语言描述的目标表和字段生成一个结构化的表识别结果。这是问数链路里的第一步模型需要根据用户的提问从上百张表中选出可能相关的候选表。第二个是生成查询技能。在确定了表和字段之后根据业务条件生成对应的 SQL 查询语句。这一步最关键的地方是不仅要生成 SQL还要让模型先解释它的生成思路然后才输出 SQL。让模型先思考再输出能显著降低错误率。第三个是结果解读技能。查询结果是一堆数字但用户要的是结论。模型需要基于结果数据和原始问题生成一段自然的文字总结。这段总结要从业务视角出发而不是简单罗列数字。这三块技能对应的提示词我都单独放在prompts.py里管理不混在业务代码中。这样后续调优提示词时不需要改动代码逻辑只需要改对应的模板文件。5.3 工具调用链路的边界控制Agent 调用工具是整个系统里最需要控制边界的环节。在基础设施里我做了三层边界控制第一层是工具白名单。Agent 只能调用在tools.py里显式注册过的工具任何未注册的工具调用都会被拒绝。这防止了模型因为提示词注入或者其他原因尝试去调用我们不该暴露的能力。第二层是查询超时控制。数据库查询语句执行有时间上限我设置了 30 秒的硬超时。超过时间的 SQL 会被强制终止避免一条慢查询拖垮整个服务的数据库连接池。第三层是行数限制。Agent 生成的 SELECT 语句在基础设施层强制追加LIMIT子句。业务上默认最多返回 1000 行数据。这个限制是为了防止模型生成不带条件的全表查询一次拉几百万行出来把内存打爆。这三层边界不是相互替代的关系而是层层叠加上去的保险。我在实际项目里就遇到过模型生成的 SQL 里带了DELETE FROM前缀的情况虽然因为只读权限被数据库拦下来了但也说明代码层面的防线同样必要。6. 测试验证与常见问题排查6.1 基础设施层的自检清单基础设施搭完之后我建议先跑一套自检脚本再往下做业务功能开发。我把自检清单整理成了下面这张表拿过来可以直接当验收标准用检查项验证方法预期结果环境变量加载启动服务时打印关键配置脱敏配置正确无 KeyError模型接口连通性发送一条简单测试消息能收到模型回复MySQL 连接池并发发起 50 个查询连接池稳定无超时PostgreSQL 连接池并发发起 30 个查询连接池稳定无超时元数据采集执行采集函数打印表数量与实际表数量一致只读权限校验尝试执行 DELETE 语句数据库层面直接拒绝超时控制发起一条人为构造的慢查询30 秒后被强制终止行数限制查询全表不指定 WHERE返回最大 1000 行链路追踪发起一条测试请求记录 trace_id日志中可按 trace_id 串起完整链路健康检查接口GET /api/v1/health返回 200 和状态信息这套自检清单我强烈建议固化成 pytest 测试用例跑在 CI 流程里。每次代码变更之后跑一遍测试就知道基础设施有没有被改坏不用等到上线了才发现问题。6.2 我踩过的三个经典坑基础设施搭建过程中我遇到了不少问题挑三个最有代表性的分享出来这几个坑在文档里很少被人提起。坑一连接池耗尽导致服务雪崩。最开始我把连接池的pool_size设置得很大以为越大越能扛并发。结果在一次压测中数据库连接数直接打满数据库拒绝新的连接然后整个服务所有的查询请求全部报错最终表现为服务不可用。后来才想明白连接池不是越大越好每个连接都会占用数据库的服务端资源。正确的做法是设置一个合理的池大小配合排队机制而不是无限放量。我现在的配置是pool_size10, max_overflow20在这个配置下即使短时间有大量并发查询也只是排队等待不会打垮数据库。坑二异步环境下的 trace_id 丢失。我最初用全局变量来存 trace_id在同步代码里没问题但切到异步以后多个并发协程共享同一个全局变量导致日志里的 trace_id 乱了套排查问题时看到的数据全都是串的。解决方法是改用contextvars这是 Python 官方提供的异步上下文变量解决方案。每个异步任务的上下文是独立的协程之间互不干扰。代码我之前已经展示过了这里就不再重复但强烈建议每个人都检查一下自己的日志链路看是否已经踩了这个坑。坑三元数据采集没有增量更新机制。第一版元数据采集器是全量扫描每次服务重启都重新扫一遍所有表结构。表少的时候没感觉等业务库涨到几千张表以后一次全量扫描要跑好几分钟启动时间慢得离谱。后来我改成了增量更新机制采集器启动时先加载全量数据后续每隔一段时间只查询information_schema中更新时间有变化的表。对大部分日常变更来说增量更新几秒钟就能完成效果非常明显。6.3 基础设施的下一步演进方向基础设施从无到有搭建起来只是整个问数项目的第一步。按照我的规划接下来还有几个方向需要继续补齐第一向量检索能力。目前表结构的匹配主要靠模型从全量元数据 dict 中理解。表少的时候没问题但表一多全量塞给模型既不经济也不精准。后面我计划把表字段的注释信息进行向量化存储到向量数据库中通过语义相似度先做一轮粗筛选精准找到最相关的几张表再交给模型做细粒度理解。第二缓存策略。相同的或者相似的问题其实没必要每次都完整走一遍 Agent 链路。加一层基于语义相似度的缓存短时间内重复提问可以直接返回历史结果能大幅降低模型调用量和查询延迟。第三多轮对话上下文管理。现在的基础设施已经支持单轮问答但真实业务场景中用户常常会追问那华东区的数据呢这种省略了主体的多轮问题需要 Agent 记住前面对话的上下文。这块涉及会话状态管理是接下来要补的重点。基础设施搭建确实花时间但它决定了后面所有功能的上限和稳定性。我这次把完整的踩坑过程记录下来就是希望大家能少走一些弯路。下一篇我会接着讲 Agent 核心编排链路的实现细节包括如何设计工具调用的提示词、如何处理模型输出的不稳定以及如何做结果验证和纠错机制。到时候咱们继续聊。