问数项目智能体基础设施实战：FastAPI、LangChain与多数据源集成

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 = [ "fastapi>=0.115.0", "uvicorn[standard]>=0.30.0", "pydantic>=2.7.0", "pydantic-settings>=2.3.0", "sqlalchemy>=2.0.30", "pymysql>=1.1.1", "psycopg2-binary>=2.9.9", "langchain>=0.3.0", "langchain-openai>=0.2.0", "openai>=1.40.0", "python-dotenv>=1.0.1", "loguru>=0.7.2", "tenacity>=8.5.0" ] [project.optional-dependencies] dev = [ "pytest>=8.2.0", "ruff>=0.5.0", "mypy>=1.10.0" ]

版本这里我要特别提醒一句：LangChain 的版本迭代非常快，0.1 和 0.3 的 API 差别很大。我上面的版本是实测可用的版本组合，大家如果下载时已经出了更高的版本，务必先看 changelog，不要盲目升级。

环境变量这块，我建了一个.env.example作为模板提交到仓库，真正的.env文件通过.gitignore忽略，确保密钥不泄露：

# 模型服务配置 LLM_BASE_URL=https://api.example.com/v1 LLM_API_KEY=your-api-key-here LLM_MODEL=gpt-4o-mini LLM_TEMPERATURE=0.1 LLM_TIMEOUT=60 # 数据库配置 MYSQL_HOST=localhost MYSQL_PORT=3306 MYSQL_USER=readonly_user MYSQL_PASSWORD=your-db-password MYSQL_DATABASE=business_db # 向量库配置 VECTOR_STORE_HOST=localhost VECTOR_STORE_PORT=19530 VECTOR_STORE_COLLECTION=table_metadata_chunks # 服务配置 SERVICE_HOST=0.0.0.0 SERVICE_PORT=8000 LOG_LEVEL=INFO ENABLE_TRACING=true

配置文件加载使用 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( title="LCODER AskData Agent Service", description="问数项目智能体 - 基础设施版本", version="0.1.0", lifespan=lifespan, ) # 跨域配置，前后端分离部署时必须设置 app.add_middleware( CORSMiddleware, allow_origins=settings.CORS_ORIGINS, allow_credentials=True, 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 = f"mysql+pymysql://{user}:{password}@{host}:{port}/{database}?charset=utf8mb4" elif db_type == "postgresql": url = f"postgresql+psycopg2://{user}:{password}@{host}:{port}/{database}" else: raise ValueError(f"不支持的数据库类型: {db_type}") engine = create_engine( url, poolclass=QueuePool, pool_size=10, # 连接池大小 max_overflow=20, # 连接池溢出上限 pool_timeout=30, # 获取连接超时时间 pool_recycle=1800, # 连接回收周期，避免 MySQL 8 小时断开问题 pool_pre_ping=True, # 使用前预检连接有效性 echo=False, connect_args={"connect_timeout": 10}, ) return engine

这里面有几个参数需要特别说明。pool_pre_ping=True是我强烈建议开启的，它会在每次从连接池取出连接前先发一个轻量的探测语句，如果连接已经失效就自动丢弃并创建新连接。这个配置能避免 MySQL 服务端因为 wait_timeout 把空闲连接断掉之后，应用拿到半死连接然后报错的问题。

pool_recycle=1800对应的是 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（订单ID），order_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_url=settings.LLM_BASE_URL, api_key=settings.LLM_API_KEY, timeout=settings.LLM_TIMEOUT, max_retries=3, ) 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( model=self.model, messages=messages, tools=tools, tool_choice=tool_choice, temperature=temperature if temperature is not None else self.temperature, ) return response

调模型时 temperature 这个参数在问数场景里必须调低。我设置的默认值是 0.1，目的是让模型的输出尽量确定和稳定。问数这种数据分析场景，需要的是尽量少幻觉、少自由发挥，如果你把 temperature 设成 0.7 甚至更高，模型可能每次给的查询逻辑都不一样，这个对业务来说是不可接受的。

max_retries=3是 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 基础设施层的自检清单

基础设施搭完之后，我建议先跑一套自检脚本再往下做业务功能开发。我把自检清单整理成了下面这张表，拿过来可以直接当验收标准用：

这套自检清单我强烈建议固化成 pytest 测试用例，跑在 CI 流程里。每次代码变更之后，跑一遍测试就知道基础设施有没有被改坏，不用等到上线了才发现问题。

6.2 我踩过的三个经典坑

基础设施搭建过程中，我遇到了不少问题，挑三个最有代表性的分享出来，这几个坑在文档里很少被人提起。

坑一：连接池耗尽导致服务雪崩。

最开始我把连接池的pool_size设置得很大，以为越大越能扛并发。结果在一次压测中，数据库连接数直接打满，数据库拒绝新的连接，然后整个服务所有的查询请求全部报错，最终表现为服务不可用。

后来才想明白：连接池不是越大越好，每个连接都会占用数据库的服务端资源。正确的做法是设置一个合理的池大小，配合排队机制，而不是无限放量。我现在的配置是pool_size=10, max_overflow=20，在这个配置下，即使短时间有大量并发查询，也只是排队等待，不会打垮数据库。

坑二：异步环境下的 trace_id 丢失。

我最初用全局变量来存 trace_id，在同步代码里没问题，但切到异步以后，多个并发协程共享同一个全局变量，导致日志里的 trace_id 乱了套，排查问题时看到的数据全都是串的。

解决方法是改用contextvars，这是 Python 官方提供的异步上下文变量解决方案。每个异步任务的上下文是独立的，协程之间互不干扰。代码我之前已经展示过了，这里就不再重复，但强烈建议每个人都检查一下自己的日志链路，看是否已经踩了这个坑。

坑三：元数据采集没有增量更新机制。

第一版元数据采集器是全量扫描，每次服务重启都重新扫一遍所有表结构。表少的时候没感觉，等业务库涨到几千张表以后，一次全量扫描要跑好几分钟，启动时间慢得离谱。

后来我改成了增量更新机制：采集器启动时先加载全量数据，后续每隔一段时间只查询information_schema中更新时间有变化的表。对大部分日常变更来说，增量更新几秒钟就能完成，效果非常明显。

6.3 基础设施的下一步演进方向

基础设施从无到有搭建起来，只是整个问数项目的第一步。按照我的规划，接下来还有几个方向需要继续补齐：

第一，向量检索能力。目前表结构的匹配主要靠模型从全量元数据 dict 中理解。表少的时候没问题，但表一多，全量塞给模型既不经济也不精准。后面我计划把表字段的注释信息进行向量化，存储到向量数据库中，通过语义相似度先做一轮粗筛选，精准找到最相关的几张表，再交给模型做细粒度理解。

第二，缓存策略。相同的或者相似的问题，其实没必要每次都完整走一遍 Agent 链路。加一层基于语义相似度的缓存，短时间内重复提问可以直接返回历史结果，能大幅降低模型调用量和查询延迟。

第三，多轮对话上下文管理。现在的基础设施已经支持单轮问答，但真实业务场景中，用户常常会追问"那华东区的数据呢"，这种省略了主体的多轮问题，需要 Agent 记住前面对话的上下文。这块涉及会话状态管理，是接下来要补的重点。

基础设施搭建确实花时间，但它决定了后面所有功能的上限和稳定性。我这次把完整的踩坑过程记录下来，就是希望大家能少走一些弯路。下一篇我会接着讲 Agent 核心编排链路的实现细节，包括如何设计工具调用的提示词、如何处理模型输出的不稳定，以及如何做结果验证和纠错机制。到时候咱们继续聊。