做 AI Agent 的人都知道决定一个智能体项目最终能不能落地一半看模型调教另一半看基础设施。这里说的“基础设施”不是买几台服务器、装个数据库那么简单而是承载对话、记忆、数据查询、权限控制、日志追踪这一整套运行逻辑的底层架子。LCODER 问数项目正好走到这个节点需求已经拆完接下来就得一块块砌地基。问数项目说白了是一个能用自然语言对话、并且真正去数据库里把答案查出来的智能体。用户问一句“上个月华东区销售额是多少”它内部要完成意图理解、表定位、SQL 生成、安全校验、查询执行、结果解释一整条链路。上一篇文章我们把整体方案理清楚了这篇直接进入实操基础设施搭建。我会把依赖安装、工程骨架、数据源接入、大模型调用、向量知识库、服务封装和日志追踪完整串一遍最后把踩过的坑整理成清单。适合正在做 Text-to-SQL、企业知识问答或类似 Agent 项目的朋友参考。1. 问数智能体的基础设施到底要搭哪些东西1.1 先盘需求一次问答背后发生了什么很多同学一上来就急着写代码结果搭到一半发现少个组件又回头改结构。我的习惯是先把一次完整问答背后的路径画出来再反推需要什么基础设施。一次典型的问数请求长这样用户输入问题比如“查询库存超过 100 的商品名称和数量”。Agent 加载会话历史结合当前问题判断用户意图。从元数据/向量知识库中检索相关表结构、字段含义、口径说明。组装 system prompt把数据字典和业务规则交给大模型。大模型生成 SQL必要时调用工具二次校验或改写。安全层检查 SQL确认是查询语句再交给数据库执行。拿到查询结果后让模型生成人话解释返回给用户。从这条路反推基础设施至少要包含五块大模型接入层、Agent 调度核心、数据源管控层、知识库与记忆组件、对外服务与可观测性模块。每块都需要提前做好抽象和配置否则后期每接一个数据源、每换一次模型都要伤筋动骨。1.2 技术选型Python 还是 Java框架该选谁我没法替所有团队做决定但可以分享一下问数项目这个场景下的选型逻辑。如果团队以 Java 为主而且智能体要嵌入到现有的微服务、审批流、权限体系里那走 Spring AI 或 LangChain4j 更顺毕竟和 Spring Boot 生态天然打通。但如果目标是快速跑通、频繁迭代 prompt 和 Agent 逻辑Python 生态依然是 AI 开发里最成熟的。我这次选的是 Python FastAPI原因很直接sql 解析、数据处理、embedding、prompt 调试这些工具链最全出问题随手就能写脚本验证。框架方面LangChain 这类重量级框架不是必选项。问数项目如果只是“问题 - SQL - 结果”的单轮链路自研三五个模块就够了强行上 LangChain 反而引入大量用不到的概念层。只有当后面要做多智能体编排、复杂工具调用、自动规划时才值得引入 LangGraph 这层抽象。我的建议是初版用自己的代码把主链路跑通保留替换空间等复杂度上来再往框架上迁移。2. 环境准备与工程骨架搭建2.1 开发环境初始化我习惯用 Python 3.11兼容性比 3.12 稳妥语法特性也完全够用。环境初始化我用 venv 就够了不需要额外装 conda除非团队里有人依赖 conda 做跨项目隔离。mkdir lcoder-ask-agent cd lcoder-ask-agent python3.11 -m venv .venv source .venv/bin/activate pip install -U pip依赖这块初版尽量少装用到什么加什么避免后期升级互相打架。下面这份是我认为搭一个问数 Agent 最精简的依赖清单pip install openai pandas sqlalchemy pydantic-settings \ langchain-core langchain-openai langchain-community \ tiktoken fastapi uvicorn qdrant-client \ DB-GPT # 你没有看错这个库后面的版本可以解等一下DB-GPT 那行是我笔误问数项目初期不需要引这么大一个框架。实际装这些就够pip install openai pandas sqlalchemy pydantic-settings \ langchain-core langchain-openai langchain-community \ tiktoken fastapi uvicorn qdrant-client装完后验证一下版本避免过新或过旧的包踩兼容坑。比如 LangChain 社区包经常做 Breaking Change版本锁死很重要。建议生成一份 requirements-lock.txt 提交到仓库保证同事拉下来环境一致。2.2 项目目录与配置规范项目骨架我直接给出当前在用的结构你可以按自己团队习惯微调lcoder-ask-agent/ ├── .env ├── .env.example ├── configs/ │ └── settings.yaml ├── app/ │ ├── main.py # FastAPI 入口 │ ├── api/ │ │ └── v1/ │ ├── agent/ │ │ ├── executor.py # Agent 主调度 │ │ ├── prompts.py # 提示词模板 │ │ └── tools.py # 工具封装 │ ├── llm/ │ │ ├── client.py # 大模型统一客户端 │ │ └── schemas.py │ ├── datasource/ │ │ ├── models.py │ │ ├── metadata.py # 元数据采集 │ │ └── connector.py │ ├── memory/ │ │ ├── session_store.py │ │ └── history.py │ ├── knowledge/ │ │ ├── embedder.py │ │ ├── vector_store.py │ │ └── retriever.py │ ├── schemas/ │ ├── core/ │ │ ├── config.py │ │ └── logging.py │ └── utils/ └── tests/目录不是为了好看而是让每个人都能快速定位代码。agent 目录放调度逻辑llm 目录放模型接入datasource 放数据源相关knowledge 放知识库检索。以后谁接手一眼就知道改哪里。配置分离是基础设施的命脉。密钥类配置放 .env# .env.example OPENAI_API_KEYsk-xxx OPENAI_BASE_URLhttps://api.xxx.com/v1 LLM_MODELgpt-4o-mini EMBEDDING_MODELtext-embedding-3-small DATABASE_URLpostgresql://user:passlocalhost:5432/ask_db VECTOR_DB_URLqdrant://localhost:6333非敏感配置放 configs/settings.yaml比如超时时间、重试次数、表数量上限。然后用 pydantic-settings 统一读取from pydantic_settings import BaseSettings from pydantic import Field class Settings(BaseSettings): openai_api_key: str Field(aliasOPENAI_API_KEY) openai_base_url: str Field(aliasOPENAI_BASE_URL) llm_model: str Field(defaultgpt-4o-mini, aliasLLM_MODEL) embedding_model: str Field(defaulttext-embedding-3-small, aliasEMBEDDING_MODEL) database_url: str Field(aliasDATABASE_URL) vector_db_url: str Field(aliasVECTOR_DB_URL) request_timeout: int 60 settings Settings()这样配置加载就是一个对象代码里到处 from config import settings干净又不容易出错。2.3 数据模型设计会话、消息、数据源问数 Agent 需要持久化三类数据会话、消息、数据源配置。我给出一版可以直接落库的表结构CREATE TABLE session ( id VARCHAR(64) PRIMARY KEY, user_id VARCHAR(64), title VARCHAR(255), created_at TIMESTAMP DEFAULT now(), updated_at TIMESTAMP DEFAULT now() ); CREATE TABLE message ( id BIGSERIAL PRIMARY KEY, session_id VARCHAR(64), role VARCHAR(16), content TEXT, sql_text TEXT, cost_tokens INT DEFAULT 0, latency_ms INT DEFAULT 0, created_at TIMESTAMP DEFAULT now() ); CREATE TABLE datasource ( id SERIAL PRIMARY KEY, name VARCHAR(64), db_type VARCHAR(32), host VARCHAR(128), port INT, database VARCHAR(128), username VARCHAR(64), password_encrypted TEXT, metadata_json JSONB, is_active BOOLEAN DEFAULT true, created_at TIMESTAMP DEFAULT now() );session 和 message 分开存是因为一次会话有多轮问答消息要能按时间顺序拉出来拼上下文还能单独记录每轮的 sql_text 和 token 消耗。datasource 单独建表是为了以后一个 Agent 服务多个业务库管理界面里增删改数据源时不必动代码。密码字段一定不能明文存用加密算法处理后落库读取时再解密。这个在安全上也值得多花十分钟处理别图省事。3. 数据源层与大模型层落地3.1 数据源连接账号权限、连接池、元数据采集问数项目的第一个硬骨头是数据源层。第一步给智能体单独建一个只读账号不要用业务账号直连数据库。只读能保证 Agent 即使被注入攻击也无法篡改数据。以下是在 PostgreSQL 里的创建方式CREATE USER ask_bot WITH PASSWORD strong_password; GRANT CONNECT ON DATABASE business_db TO ask_bot; GRANT USAGE ON SCHEMA public TO ask_bot; GRANT SELECT ON ALL TABLES IN SCHEMA public TO ask_bot; ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO ask_bot;第二步是连接池配置。直接用 SQLAlchemy 的 create_engine参数要按业务量调整from sqlalchemy import create_engine engine create_engine( settings.database_url, pool_size5, max_overflow10, pool_timeout30, pool_recycle3600, pool_pre_pingTrue, connect_args{connect_timeout: 10} )pool_recycle 设为 3600 秒避免数据库端主动断开后连接池还拿着失效连接。pool_pre_ping 在取连接时先探测一下能省掉很多“连接已关闭”的报错。第三步是元数据采集。要让模型生成准确的 SQL它得知道库里有哪些表、每个字段是什么含义。最可靠的方式是查系统表比如 PostgreSQL 的 information_schemaSELECT table_name, column_name, data_type, column_comment FROM information_schema.columns WHERE table_schema public ORDER BY table_name, ordinal_position;元数据采集完成后建议做一层缓存不要每个请求都去查 information_schema否则表一多数据库压力很大。我的做法是每小时刷新一次元数据到内部存储同时提供一个手动刷新接口表和字段变更后点一下就能更新。安全方面还有几条硬规则只允许执行 SELECT 语句禁止 INSERT/UPDATE/DELETE/DDL查询强制加 LIMIT单条查询执行时间超过 10 秒直接取消。这几条规则看似简单却是问数智能体能从 demo 走向生产的保护网。3.2 大模型接入统一接口、重试与限流大模型接入层最忌讳把某个厂商的 SDK 散落在业务代码里。我封装了一个统一的 LLMClient对外只暴露 chat 方法内部再决定调用哪家模型。from openai import OpenAI class LLMClient: def __init__(self, config): self.client OpenAI( api_keyconfig.openai_api_key, base_urlconfig.openai_base_url, timeoutconfig.request_timeout ) self.model config.llm_model def chat(self, messages, temperature0.2, max_tokens2048): resp self.client.chat.completions.create( modelself.model, messagesmessages, temperaturetemperature, max_tokensmax_tokens, ) return resp.choices[0].message.contenttemperature 在问数场景里我固定用 0.2甚至直接用 0。因为生成 SQL 不是创作任务需要的是确定性和可复现性温度越高越容易出语法错误或幻觉 SQL。重试逻辑必须做但要有原则。429 限流和 5xx 服务端错误可以重试用指数退避加抖动4xx 参数错误直接抛异常重试多少次都没用。我的重试设置是初始等待 1 秒每次翻倍最多重试 3 次。大模型接入层还要预留降级策略。比如主模型超时可以切到备用的轻量模型先返回一条“正在处理”的兜底消息或者走异常队列人工处理。一开始不用做得很重但接口抽象要留好这个位置。3.3 首次跑通“问数”主链路基础设施搭完最好立刻写一个最小可运行的链路验证所有组件能串起来。这个阶段不要写花哨的 Agent 逻辑就做四件事接收问题、拼 prompt、调模型生成 SQL、执行查询返回结果。class AskExecutor: def __init__(self): self.llm LLMClient(settings) self.engine get_engine() def ask(self, question: str) - dict: system_prompt 你是一个数据分析助手。请根据表结构信息生成 SQL 查询语句只输出 SQL不要输出多余内容。 messages [ {role: system, content: system_prompt}, {role: user, content: question} ] sql self.llm.chat(messages) result self.execute_sql(sql) return {sql: sql, result: result}这个版本虽然很简陋但它能帮助我们尽早确认数据库连得上、模型调得通、结果出得来。很多项目死在第一步就是因为一上来就追求完美架构结果三个月还在“调包”。先把链路打通后面再一层层加安全和精度。4. 向量知识库与动态记忆4.1 为什么要给智能体加一套知识库很多人觉得表结构信息全部塞进 system prompt 不就行了为什么还要向量库原因是表一多prompt 根本塞不下。假设一个业务库有 200 张表每张表的字段描述平均 500 字全量塞进去就是 10 万字远超模型上下文窗口成本也扛不住。更合理的办法是让 Agent 按需获取用户问“库存”我们从知识库里把库存相关的 3-5 张表结构捞出来拼进 prompt。这个思路类似给新同事一本数据字典遇到问题先查再动手而不是整本书背下来。向量知识库存四类内容建表语句和字段说明、指标计算口径、常用查询模板、历史问答对。其中“指标口径”尤其重要比如“销售额”到底含不含税模型不可能自己知道知识库里写清楚后prompt 组装时带上下文能明显减少口径错误。4.2 Embedding 与入库流程实操Embedding 模型我分两套方案。线上原型阶段用 OpenAI 的 text-embedding-3-small效果好、接入快私有化部署时用本地的 BGE-M3 或 Qwen3-Embedding。选择本地模型的原因是数据不出内网对数据敏感的企业这往往是硬性要求。入库流程本质上就是“元数据 - 文本块 - 向量 - 存储”。下面这段是核心逻辑你可以直接复用from sentence_transformers import SentenceTransformer class MetadataIndexer: def __init__(self): self.model SentenceTransformer(BAAI/bge-m3) self.metadata load_metadata() # 来自元数据采集模块 def build_documents(self): docs [] for table in self.metadata: text ( f表名{table[table_name]}\n f表注释{table[table_comment]}\n f字段列表\n ) for col in table[columns][:50]: text f- {col[column_name]}{col[data_type]}{col[column_comment]}\n docs.append({id: ftable_{table[table_name]}, text: text, metadata: {table: table[table_name]}}) return docs def embed_and_upsert(self): docs self.build_documents() vectors self.model.encode([d[text] for d in docs]) # 写入 qdrant 或 pgvector这里有两个细节要注意。第一字段超过 50 个的表不要全塞取业务含义最明确的核心字段否则检索容易带偏。第二每个字段还可以单独建一条索引比如“库存数量”“库存金额”这种别名信息单条检索召回率更高。向量库选型方面问数项目表数量在几百张以内用 PostgreSQL 自带的 pgvector 就够了少一个组件、少一份运维成本。如果要做千人千面的个性化语义检索数据量大、并发高再上 Qdrant 或 Milvus。初期别把架构撑太大。4.3 检索增强的提示词组装有了知识库真正的效果取决于怎么把检索结果组装进 prompt。我的做法是用户问题先用 embedding 向量化。在向量库中检索 top-5 到 top-10 相关的表结构。把这几个表的结构按统一格式拼成“数据字典”片段。连同指标口径、历史相似问答一起塞进 system prompt。代码示例class Retriever: def __init__(self): self.embedder get_embedder() self.store get_vector_store() def retrieve(self, question: str, top_k5): q_vec self.embedder.encode(question) hits self.store.search(q_vec, top_ktop_k) return [h.payload[text] for h in hits] def build_prompt(question: str, retriever: Retriever): related_tables retriever.retrieve(question) dict_text \n\n.join(related_tables) prompt f 你是一个数据分析助手。以下是相关数据表结构 {dict_text} 请根据用户问题生成 SQL。注意 1. 只返回 SQL不要解释。 2. 如果表结构与问题无关请输出 ERROR:NO_TABLE_MATCH。 3. 优先使用字段注释明确反映业务含义的字段。 用户问题{question} return prompt组装 prompt 时注意给模型留一条“拒绝通道”当检索到的表和问题不相关时应该输出 ERROR:NO_TABLE_MATCH而不是硬编一条 SQL。早期版本没有这个模型经常在无关表上强行生成 SQL结果查询跑出来根本不对。加上这个约束后能挡住很大一部分低质量请求。5. 服务接口与可观测性5.1 用 FastAPI 把智能体包成服务Agent 写得再好也要通过服务暴露出去才能被页面、IM 机器人、小程序调用。我选了 FastAPI因为异步支持好天然适合这种 IO 密集型的 Agent 服务。基础接口先定三个from fastapi import FastAPI from pydantic import BaseModel, Field app FastAPI(titleLCODER 问数 Agent) class AskRequest(BaseModel): session_id: str question: str class AskResponse(BaseModel): session_id: str answer: str sql: str app.post(/v1/ask) async def ask(req: AskRequest): # 这里需要将来改成异步或交给线程池执行 result executor.ask(req.session_id, req.question) return AskResponse(**result) app.get(/v1/health) async def health(): return {status: ok}务必要注意 FastAPI 中同步路由会阻塞事件循环。如果 executor.ask 是同步函数在高并发下整个服务会被卡住。最简单的做法是把耗时操作丢给线程池import asyncio from concurrent.futures import ThreadPoolExecutor pool ThreadPoolExecutor(max_workers8) app.post(/v1/ask) async def ask(req: AskRequest): loop asyncio.get_running_loop() result await loop.run_in_executor(pool, executor.ask, req.session_id, req.question) return AskResponse(**result)后期如果流量上来再把大模型客户端切换成 async 版本后面我会专门展开。初版用线程池足够稳定。5.2 链路日志与 token 成本追踪问数 Agent 调试时最烦的是什么是你只知道用户问了一句“华东区销售额”然后系统返回了乱码却不知道中间哪一步出了问题是检索没召回还是模型 SQL 写错了还是数据库执行超时解决方案是给每个请求生成一个 request_id然后所有日志都带这个 ID 串起来{ request_id: req_8f3a2e, session_id: sess_001, step: retrieve, related_tables: [orders, order_items], latency_ms: 45 }我实现了一个简单的 trace 装饰器在每个关键步骤检索、生成 SQL、执行 SQL、生成回答上都打点。这样一次问答的耗时分布一目了然也方便后续做全链路监控。token 成本追踪同样重要。每次模型调用把输入 token、输出 token、模型名、请求时间记录下来。时间久了这些数据就是优化模型选型和 prompt 的第一手依据。比如发现某个月模型调用成本暴涨翻日志一看原来是 prompt 里塞的表结构变多了就能针对性地做压缩。6. 常见问题与避坑实录6.1 数据库连接与权限相关第一类高频问题出在连接上。我遇到过最多的是本地连得好好的部署到容器里就报 connection refused。排查时先确认数据库白名单有没有放容器网段其次检查连接串是不是写到了环境变量里被容器覆盖了。权限问题也很常见。只读账号建好了但新表默认没有授权给该账号导致 Agent 查新表一直报 permission denied。解决办法是给只读账号设置默认权限ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO ask_bot;连接池被占满同样经常出现。根因往往是某条慢查询卡了 30 秒占着连接不释放。把 execute 语句包一层超时控制超过阈值立刻取消比调大连接池更有效。6.2 提示词与上下文管理模型生成 SQL 不稳定的问题多半不是模型不行而是 prompt 里的上下文不够。比如订单表有 order_id 和 payment_order_id 两个字段模型没看到注释就容易选错。我的办法是把字段注释写得更口语化甚至可以加“别名”比如“order_id: 业务订单号用户下单生成的唯一编号不是支付单号”。历史消息太长是另一个坑。多轮对话后把整段历史都塞进上下文既费 token 又让模型抓不住重点。我的策略是做“滑动窗口 摘要”最近 5 轮完整保留更早的历史让模型压缩成一段摘要塞进 system prompt。还有一个特别容易漏的问题用户问法模糊比如只问一句“看看这个月的数据”模型没有上下文根本不知道“这个月”是哪张表。这种情况建议 Agent 主动反问而不是猜一个表去查。我在提示词里明确加了规则上下文不明确时先列出可用的表和筛选条件让用户确认。6.3 向量检索不理想向量检索效果差十有八九是索引内容本身质量不好。有些业务表的字段注释是空的模型自然检索不到。解法是先跑一遍元数据把注释为空的字段清单导出来找业务方补注释。这是让问数效果变好性价比最高的一步远胜于反复调 prompt。分块策略也会影响检索。我一开始按字段单独建索引结果一个语义完整的表结构被切得七零八落检索时明明表是对的都匹配不上。后来改成“整表结构为一条文档 关键字段单列索引”的混合方案效果立刻就上来了。相似度阈值也要拍脑袋定个合理值。我实测下来BGE-M3 的相似度阈值设在 0.45 到 0.55 之间比较合适。低于 0.45检索结果基本是噪音高于 0.55又容易漏召回。但这个值随 embedding 模型变化最好在真实数据上抽几十个问题跑一遍再看。6.4 并发与稳定性问题服务刚上线时最多的问题是并发一高表现就不稳定。第一个坑是 FastAPI 同步路由阻塞事件循环前面已经说过用线程池或全异步解决。第二个坑是 embedding 模型和 LLM 调用抢资源。本地部署 embedding 模型时如果服务是单机跑embedding 推理会占满 CPU影响 LLM 调用响应。初期做法是先给 embedding 加上缓存对相同或相似的问题直接命中缓存不重复计算。第三个坑是大模型调用的超时和重试设置不合理。网络抖动时如果只重试一次可能撑不过去但如果无限重试又会拖垮服务。我把重试次数定为 3 次每次间隔 1 秒、2 秒、4 秒翻倍同时设置总超时 60 秒兜底确保请求不会无限吊着。再补一个容易被忽略的稳定性问题模型返回的 SQL 里可能带着 Markdown 代码块标记比如 sql 开头。执行前要做字符串清理否则 SQLAlchemy 直接报语法错误。这个坑看着小实际踩的人不少。基础设施这层其实没什么玄学就是按“数据源可控、模型可换、检索可查、日志可追”四个原则一点点搭。LCODER 问数项目把这块地基打好之后后面加工具调用、加多智能体编排才会顺。我个人最大的体会是宁可前期多花半小时把日志和配置抽象做完整也不要等线上出问题再去翻 print。最后再分享一个小技巧所有模型调用和 SQL 执行都记录到一个独立的 cost 表里用久了你会发现它对后续优化模型选型和 prompt 策略的价值比预期大得多。