这几年 AI Agent 从概念到落地最不缺的就是 PPT 和 demo。真正把 Agent 推到生产环境的时候最先遇到的往往不是模型能力不够而是基础设施缺位密钥散落在代码里、日志查不到一次工具调用链路、数据库查询把线上库拖慢了、模型供应商一换整个流程重写。我在 LCODER 团队做“问数项目”时第一阶段把业务需求梳理清楚第二阶段就老老实实打地基。这篇就讲我们是怎么搭建这套 AI Agent 基础设施的适合正在做或者准备做数据问答智能体的开发者也适合想系统学习 Agent 工程化落地的人参考。“问数项目”做什么简单说就是让业务人员用自然语言查询数据。用户输入“上个月华东区销售额环比是多少”Agent 理解问题、生成 SQL、执行查询、返回一段人能看懂的结论。这个链路看着不长但它横跨模型调用、任务编排、数据库访问、日志监控、权限控制好几个层面。任何一个环节没想清楚上线后都会变成半夜打电话叫你起来的故障。基础设施搭建阶段就是把这一条链路上的“地基”一件一件摆正。1. 开工前先想清楚问数项目的基础设施到底要解决什么问题1.1 问数项目的本质是一条数据链路问数项目本质上是把“自然语言到 SQL再到结果解释”这条链路自动化。用户的一句话要经历这些环节意图理解、字段映射、SQL 生成、SQL 执行、结果摘要、回答生成。每个环节都可能出错而且错误会层层放大模型生成 SQL 时漏掉一个过滤条件查询结果就是错的但看起来还挺像那么回事。所以基础设施搭建不能只围绕“调用大模型”这一个点而是要把整条数据链路当成一个系统来设计。模型接入层只负责文字生成编排层负责控制流程数据访问层负责安全高效地执行查询可观测层负责记录每次决策和工具调用。只有每一层边界清晰后续排查问题时才不会把所有锅都甩给“模型太笨”。1.2 基础设施的四条原则我给团队定过四条原则后面所有搭建工作都围绕它们展开可替换模型供应商不能绑定。今天用 A 厂商明天换成 B 厂商不能动业务代码。可观察每次 Agent 调用都要能回放。完整记录 prompt、工具参数、SQL、查询结果、最终回答。可限流失控查询不能拖垮数据库。无论是 SQL 还是并发请求都要有兜底机制。可回滚模型、提示词、工具配置化。线上出问题时能在一分钟之内切回旧版本。这四条不是理论。可替换保证你不被单一供应商拿捏可观察保证每次回答错误都能定位是模型问题、工具问题还是数据问题可限流是生产环境的生死线可回滚是最后一道安全网。基础设施阶段把这些原则落实后面迭代才有底气。1.3 技术选型边界我们的技术栈是这样定的你完全可以按自己的环境调整语言Python 3.11。Agent 生态基本都在 Python 这边。编排框架LangGraph。因为问数流程天然是“生成 SQL、执行、再回复”的状态流转LangGraph 的状态图模型很适合。工具协议MCP。后面要接数仓、告警平台、知识库统一用 MCP 暴露工具Agent 端不重复造轮子。LLM优先兼容 OpenAI 协议具体模型可切换我们主力用 qwen-plus 和 deepseek-chat。数据库示例用 MySQL 语义本地开发用 SQLite。生产建议单独账号、只读权限。可观测外部用 LangSmith内部保留结构化日志作为兜底。选型有一个原则能不自己写框架就不自己写但底层细节一定要吃透。框架只是帮我们管理复杂度不能代替我们思考。2. 开发脚手架与配置管理一个能持续演进的工程底板2.1 项目目录按职责拆模块而不是按页面拆基础设施搭建第一步是搭一个可持续演进的目录结构。我的习惯是让“工具”成为独立模块每增加一个数据源就加一个文件不破坏主流程。lcoder_askdb/ ├── app/ │ ├── agent/ # LangGraph 编排层 │ │ ├── graph.py │ │ ├── state.py │ │ └── nodes.py │ ├── tools/ # 工具集一个文件一个工具 │ │ ├── sql_query.py │ │ └── registry.py │ ├── llm/ # 模型工厂 │ │ ├── factory.py │ │ └── errors.py │ ├── db/ # 数据库访问层 │ │ ├── engine.py │ │ └── safe_query.py │ ├── config.py # 全局配置 │ └── logging_conf.py # 结构化日志 ├── mcp_server/ # MCP 服务端 ├── tests/ ├── pyproject.toml └── .env.example依赖管理我推荐用 uv。它比 pip 快一个数量级还能直接管理 Python 版本。项目初始化的时候直接uv init之后所有依赖加进pyproject.toml配合锁文件保证开发环境一致。这个阶段别省时间环境不一致带来的坑比想象中多。2.2 配置中心别在代码里硬编码任何东西配置管理的核心工具是 pydantic-settings。它能在启动时自动读取环境变量和 .env 文件还能做类型校验。问数项目的配置类大概是这样的from pydantic_settings import BaseSettings class Settings(BaseSettings): model_config {env_file: .env, extra: ignore} # 模型配置 llm_provider: str openai_compatible llm_model: str qwen-plus llm_base_url: str # 生产环境从环境变量读取 llm_api_key: str llm_temperature: float 0.1 llm_timeout_seconds: int 60 # 数据库配置 db_url: str sqlite:///./askdb.db db_pool_size: int 5 db_max_overflow: int 10 db_statement_timeout_ms: int 5000 # 查询限制 query_max_rows: int 100 query_max_sql_length: int 2000 # 可观测 langsmith_api_key: str langchain_tracing_v2: bool False settings Settings()这里有个实操细节.env绝对不能进 git但.env.example必须进仓库。.env.example里写好每个配置项的含义和取值示例让新同学五分钟之内能跑起来。另外所有配置项集中在config.py里 import别在业务代码里到处读os.getenv否则一个月后你自己都记不清哪些变量在哪些地方用到了。2.3 密钥管理第一道安全红线密钥管理是基础设施里最容易被忽视的部分。很多项目把 API Key 写在.env里就算完成但生产环境还应该做到密钥由部署平台统一注入环境变量开发者本地使用个人 Key代码仓库里任何位置不允许出现明文密钥。我们在 CI 里加了扫描工具只要检测到疑似的高权限密钥就会让流水线失败。判断标准就一条仓库泄露后攻击者能不能拿到模型调用额度或者数据库访问权。如果能这就算安全事故。密钥这件事没有中间态宁可严格一万次也不能松一次。3. 模型接入层让 Agent 能对话也能换模型3.1 统一模型工厂一次封装全局复用模型接入层在问数项目里承担一个核心职责屏蔽底层差异。不管是 qwen、deepseek 还是 OpenAI对上层的 Agent 逻辑来说都只应该是一个“能接收消息并返回文本和工具调用”的黑盒。我们用工厂函数创建模型实例核心代码如下from langchain_openai import ChatOpenAI from app.config import settings def get_llm(): return ChatOpenAI( modelsettings.llm_model, base_urlsettings.llm_base_url, api_keysettings.llm_api_key, temperaturesettings.llm_temperature, timeoutsettings.llm_timeout_seconds, max_retries1, )这个base_url的设计很关键。很多国产模型厂商提供 OpenAI 兼容接口只要 base_url 和 model 名字对得上业务代码一行都不用改。本地开发甚至可以指向 Ollama 跑 7B 小模型。这就是“可替换”原则的落地。3.2 重试、超时与错误归一化模型接口不是本地函数它可能超时、限流、返回 5xx。所以模型层必须统一做三件事超时控制、重试策略、错误归一化。一个实用的重试策略是对网络错误和 5xx 做指数退避重试初始等待 1 秒最多重试 3 次对 429 限流则要根据 Retry-After 头等待对 4xx 错误不重试因为重试也没用。所有错误统一包装成LLMError向上抛出编排层只认这一个异常类型不跟具体厂商的异常直接打交道。import time, asyncio from openai import APITimeoutError, RateLimitError async def call_llm_with_retry(func, max_retries3): delay 1.0 for attempt in range(max_retries): try: return await func() except (APITimeoutError, RateLimitError) as e: if attempt max_retries - 1: raise LLMError(f模型调用失败: {e}) from e await asyncio.sleep(delay) delay * 2这个封装不要在业务代码里散落就放在模型工厂文件里。所有get_llm()返回的客户端后续调用都走统一封装。3.3 兼容多模型的两个细节第一工具调用能力。问数项目必然要让模型调用查询工具所以选的模型必须支持 function calling。qwen-plus、deepseek-chat 都支持但某些本地小模型不支持。如果必须用不支持的模型就要在提示词里要求输出固定 JSON再在代码里做解析和校验。这个降级方案我在后面问题排查部分会讲到。第二temperature 参数。问数任务追求的是精确而非创意temperature 建议设 0 或 0.1。设太高模型同一个问题五次生成五种 SQL根本没法稳定复现。我们团队实测下来0.1 是准确率和鲁棒性都比较好的取值。4. Agent 编排运行时用 LangGraph 把流程变成可控状态机4.1 为什么是 LangGraph 而不是一个死循环很多 Agent demo 就是一个 while 循环调用模型、拿工具结果、再调用模型直到结束。这么做 demo 没问题但它不满足生产环境的三点要求可控、可回放、可中途干预。LangGraph 把流程定义成状态图每个节点做一类事情节点之间按条件路由整个过程都基于显式的状态流转。问数项目的状态定义如下from typing import TypedDict class AskState(TypedDict): question: str # 用户原始问题 history: list # 多轮对话历史 sql: str | None # 生成的 SQL query_result: list | None # 查询结果 answer: str | None # 最终回答 error: str | None # 错误信息定义好状态后图的结构就是首先 receive_query 节点接收并预处理问题然后 analyze 节点让模型基于表结构生成 SQL接着 execute 节点执行查询工具最后 respond 节点把结果翻译成自然语言。中间如果 SQL 执行失败就进入 error_handle 节点修正 SQL 重试一次。4.2 节点与条件路由的实践LangGraph 的节点就是一个接收 state、返回新 state 的函数。我习惯把每个节点放在nodes.py里独立成函数避免一个文件几百行。from langgraph.graph import StateGraph, END def analyze_node(state: AskState) - AskState: sql generate_sql(state[question], state[history]) state[sql] sql return state def execute_node(state: AskState) - AskState: result, error safe_query(state[sql]) state[query_result] result state[error] error return state def respond_node(state: AskState) - AskState: state[answer] summarize_answer(state[question], state[query_result]) return state graph StateGraph(AskState) graph.add_node(receive_query, receive_query_node) graph.add_node(analyze, analyze_node) graph.add_node(execute, execute_node) graph.add_node(respond, respond_node) graph.add_edge(receive_query, analyze) graph.add_edge(analyze, execute) graph.add_conditional_edges( execute, lambda state: respond if state[error] is None else analyze ) graph.add_edge(respond, END)条件路由的意思就是如果查询成功直接生成回答如果查询失败回到 analyze 节点重新生成 SQL。但这里要加一个重试次数上限比如最多重新生成两次超过就直接返回错误避免模型死循环消耗 token。4.3 工具注册与参数校验把模型关进“笼子”里问数项目里模型需要调用的是一个查询工具。工具注册阶段最重要的不是“能不能调通”而是“模型能不能按规矩调”。我们用 LangChain 的tool装饰器加 Pydantic 参数描述模型会看到结构化的工具说明。from langchain_core.tools import tool from pydantic import BaseModel, Field class QueryInput(BaseModel): sql: str Field(description要执行的只读 SQL必须带 LIMIT 100) tool(safe_query, args_schemaQueryInput) def safe_query_tool(sql: str) - tuple: 执行只读 SQL 查询支持 MySQL/SQLite自动处理超时和行数限制 return safe_query(sql)这里有个关键设计工具描述里要明确写“必须带 LIMIT”schema 里也要带。模型是概率系统我们通过描述约束它再通过后端强校验兜底。底线是即使模型不听话数据库层面也不能出问题。5. 数据访问层安全高效地“问数”数据访问层是问数项目最核心的底座。前面模型写得再好这一层出事就是大事故。所以这块我把安全放在第一位效率放在第二位。5.1 只读账号与连接池生产环境必须为 Agent 单独建一个数据库账号权限只有 SELECT不给 INSERT、UPDATE、DELETE、DDL。这一步是为了防止提示词注入的最坏情况。即使模型被诱导生成了恶意 SQL只读账号也会把损失降到最低。连接池方面直接用 SQLAlchemy 统一管理。根据预估并发设置连接池大小我一般给一个保守值from sqlalchemy import create_engine engine create_engine( settings.db_url, pool_sizesettings.db_pool_size, max_overflowsettings.db_max_overflow, pool_pre_pingTrue, pool_recycle3600, )pool_pre_pingTrue很重要它会在每次从连接池取出连接时先做一次轻量检查避免拿到坏连接。pool_recycle3600是防止数据库端超时断开。5.2 SQL 安全校验正则只是第一步很多教程只教“用正则过滤 DELETE、DROP”但实际生产里正则不够。模型可能生成多语句 SQL比如SELECT ...; DELETE ...单条正则很难覆盖所有变体。更可靠的方案是用 sqlglot 这类 SQL 解析器把 SQL 解析成 AST再做以下几项检查只允许 SELECT 语句其他语句直接拒绝。禁止多语句拼接只取第一个语句后面的一律丢弃。强制限制返回行数如果原 SQL 没有 LIMIT在外面包一层子查询加上 LIMIT。设置数据库 statement timeout防止慢查询拖垮数据库。同时在查询工具内部加一层后端兜底from sqlglot import parse_one, exp def validate_sql(sql: str) - str: sql sql.strip().rstrip(;) try: statement parse_one(sql) except Exception as e: raise ValueError(fSQL 解析失败: {e}) if not isinstance(statement, exp.Select): raise ValueError(只允许 SELECT 查询) if ; in sql: raise ValueError(不允许多语句执行) return sql加一层统一的safe_query函数把校验、超时、行数限制全部封装进去。Agent 只能通过这个函数访问数据库不能直连 engine。5.3 表结构与元数据注入模型要生成正确的 SQL前提是它知道数据库里有什么。我们会在 analyze 节点的提示词里注入核心表的建表语句和注释。注入的内容包括表名、字段名、字段类型、字段注释、索引信息、常用过滤条件示例。比如 prompt 里会有这么一段数据库表结构 - 表名: sales_order - 字段: - id BIGINT 主键 - region VARCHAR 区域华东、华北、华南等 - amount DECIMAL(10,2) 订单金额 - order_date DATE 下单日期 - created_at DATETIME 创建时间 注意查询时务必加上时间范围条件避免全表扫描。表结构可以直接写进系统提示词也可以作为工具参数描述的一部分。如果数据库表非常多不可能全部塞进 context就需要通过检索的方式筛选相关表这正好用到后面的 RAG 辅助。5.4 轻量级 RAG 辅助字段映射用户问题里的词往往和数据库字段名对不上。用户说“毛利”表里叫gross_profit说“客户”表里叫customer_name。这个映射问题单靠提示词很难完全解决。我们的做法是把字段名、字段注释、业务常用别名做成向量索引在生成 SQL 前先做一次相似度召回把候选字段信息注入到 analyze 节点。这一步算“锦上添花”的基础设施可以在第一版跑通后再加。最开始我建议先把固定表结构注入提示词跑通全链路再迭代 RAG 辅助。基础设施搭建讲究“先简后繁”别一天就想上满所有能力。6. MCP 协议接入让外部工具变成标准件6.1 为什么选 MCP 协议MCP 全称 Model Context Protocol它解决的是 Agent 和外部工具之间的标准化问题。没有 MCP 之前接一个数据库写一套自定义封装接一个告警平台再写一套每个 Agent 都在重复造轮子。MCP 把工具暴露成标准服务Agent 端只需要统一连接 MCP Server就能发现并调用所有暴露出来的工具。问数项目用 MCP 的好处是后续要接数据仓库、BI 报表、知识库、企业 API 时每个模块做成独立的 MCP ServerAgent 编排层不需要跟着改。这是基础设施层面的“面向未来设计”。6.2 用 FastMCP 暴露一个查询工具我们用 FastMCP 框架来写 MCP Server代码非常简洁。下面是一个最小可用的查询工具服务端from fastmcp import FastMCP from app.db.safe_query import safe_query mcp FastMCP(askdb-tools) mcp.tool() def query_sales(region: str | None None, start_date: str | None None) - list[dict]: 按区域和时间范围查询销售数据返回最多 100 条 sql SELECT * FROM sales_order WHERE 11 if region: sql f AND region {region} if start_date: sql f AND order_date {start_date} sql LIMIT 100 return safe_query(sql) if __name__ __main__: mcp.run()这里故意用了参数化的方式生成 SQL而不是让模型直接传入完整 SQL。虽然灵活性降低但安全性和可控性大幅提高。在这个阶段我们允许两种模式并行一种是模型直接生成 SQL 并由校验层兜底另一种是通过工具参数约束查询维度。生产环境我更推荐后者。6.3 Agent 端连接 MCP ServerAgent 端可以通过 langchain-mcp-adapters 把 MCP 工具转换成 LangChain 能识别的工具。基础设施阶段我们只需要验证一条链路MCP Server 启动后Agent 能从工具列表里发现query_sales能正确传递参数能拿到返回结果。链路跑通后再考虑同时挂多个 MCP Server 的问题。有一个坑提前提醒不同的 MCP Server 如果工具名重复Agent 端会有命名冲突。我在工具命名时统一加了业务前缀比如sales_query、schema_list而不是笼统的query。这个习惯在后端接入更多数据源时会省很多事。7. 可观测性与评估没有日志的 Agent 没法修7.1 结构化日志每条记录都能回放Agent 应用和普通 Web 应用不一样它的一次请求会触发多次模型调用和多次工具调用链路长且非线性。普通文本日志根本没法看。必须从一开始就上结构化日志推荐 JSON 行格式每个事件一行 JSON。关键事件我会打五类llm_request、tool_call、tool_result、node_start、node_end。每个事件都挂上同一个trace_id。这样排查问题时可以用 trace_id 把一次完整交互串起来。日志字段至少包含时间、trace_id、节点名、事件类型、关键数据和耗时。import logging, json def log_event(trace_id: str, event_type: str, node: str, data: dict): logging.getLogger(agent).info(json.dumps({ trace_id: trace_id, event: event_type, node: node, **data, }, ensure_asciiFalse))一个基本原则任何可能影响最终回答的因素都要记录。模型用的提示词要记录工具传入的参数要记录SQL 执行耗时要记录。宁可日志多得翻不过来也不能在出问题时什么都查不到。7.2 接入 LangSmith 做完整链路追踪如果条件允许强烈建议接入 LangSmith。它对 LangChain/LangGraph 的应用几乎是零侵入式接入自动捕获每一步的模型调用、工具调用、token 消耗和耗时。我在 config.py 里留了开关开发环境关掉测试和生产环境打开。接入步骤很简单设置环境变量LANGCHAIN_TRACING_V2true、LANGCHAIN_API_KEY...再设置 project name。之后每次运行 Agent就能在 LangSmith 里看到完整的链路回放。它最大的价值是快速定位“模型为什么生成这个 SQL”你能看到系统提示词、用户问题、模型中间推理、工具返回结果全部一清二楚。如果你不想依赖外部服务也可以用 Langfuse 或者自建一个 trace 表。但对小团队来说先上 LangSmith 以最快速度获得可观测性是性价比最高的选择。7.3 准备评估集让每次优化都可回归基础设施阶段就要开始建评估集不要等到上线之后。我们准备了 20 个 golden 问题覆盖常见业务场景简单汇总、多条件过滤、时间对比、字段映射、无法回答的问题。每个问题都标注了期望的 SQL 要点和期望答案。question,expected_sql_note,expected_answer 上个月华东区销售额是多少,必须包含 region华东 和订单日期在当月,返回销售总额 哪个客户贡献了最多的订单,必须按 customer_name 分组后 sum(amount) 排序,返回客户名和金额 查询所有订单总量,不允许全表扫描,返回总数每次修改模型、提示词或工具定义后跑一遍评估集。这个动作能让“感觉好多了”变成“指标从 82% 涨到 87%”。没有评估集的情况下做 Prompt 调优约等于蒙着眼睛开车。8. 常见问题与排查技巧实录8.1 模型输出 JSON 不稳定工具解析频繁失败现象模型不返回标准工具调用格式而是把工具参数写成 Markdown 代码块或附带解释文字解析时直接报错。后台日志会暴露一个规律这类问题几乎都出现在更换模型之后。不同厂商对工具调用的实现不完全一样个别模型在复杂场景下会退化到“用文本回答而不是调用工具”。我的排查步骤是先看 LangSmith trace确认这个问题出现在模型调用环节还是工具解析环节。确认模型版本是否支持 function calling确认 base_url 指向正确。如果模型本身支持但偶尔不稳定可以在提示词里加强约束并做文本解析兜底把返回内容里的 JSON 片段用正则提取出来再交给 Pydantic 解析。如果多次重试仍失败走异常处理节点返回“暂时无法回答请联系管理员”。8.2 SQL 多次重试单次请求累计耗时过长现象一条查询请求总耗时超过 60 秒原因不是单次 SQL 慢而是模型生成 SQL 后执行失败跳到 analyze 节点重新生成循环了好几次。每次都消耗十几秒叠加起来就超时了。我的解决办法是三层超时控制LangGraph 整体设置递归次数上限比如最多重试 2 次超过就终止。analyze 节点的模型调用设置 20 秒超时执行节点设置 5 秒数据库超时。外部请求入口设置 90 秒总超时。此外重试逻辑里增加“失败原因透传”把上一次的数据库错误信息拼进第二次生成的提示词里。比如告诉模型“刚才执行的 SQL 报错Unknown column gross_profit请检查字段名”。这样模型重试时才有方向而不是从头开始瞎猜。8.3 Token 限制导致工具 schema 过大现象表多字段多的时候把所有表结构写进提示词prompt 越来越长最终触达上下文窗口。更麻烦的是模型被大量无关表结构干扰生成 SQL 的准确率反而下降。解决思路是“按需注入”先用 5 万条左右的核心表结构缓存到本地作为固定的系统提示词。剩下的表通过向量检索召回只注入与用户问题相关的表和字段。字段描述做压缩每条只保留字段名、类型、中文注释去掉冗余说明。实测下来按需注入比一股脑全部塞进 context 的准确率高不少也节省 token 成本。8.4 并发测试时数据库连接被打满现象压测脚本同时发 50 个请求数据库连接池打满排队请求直接超时甚至拖垮了同库的其他业务。这个问题的排查方向不应该是单纯调大连接池。要分三层处理数据库只读账号数量有限先压住 Agent 入口的并发比如用队列限流同一时刻最多处理 10 个请求。针对 LLM 调用做限流和排队避免瞬时打爆模型接口的 QPS。数据库层给 Agent 账号限制最大连接数防止它占用过多资源影响其他业务。同时把慢查询和全表扫描 SQL 在数据库端用审计日志记录下来回头分析哪些表和查询模式需要优化。基础设施阶段不做这步上线后被业务方投诉“系统变慢”时再处理就非常被动了。写到这里问数项目的基础设施部分基本搭完了。我个人在实际操作中的体会是基础设施是最不显眼但决定性最强的阶段后面 90% 的稳定性问题都能在这里找到根因。这块省下的时间后面会用十倍的时间还回去。下一步我们会在这套底座上正式实现问数智能体的各个业务节点包括意图路由、SQL 生成优化、多轮对话记忆、结果可视化到那一步再继续分享实战过程。