问数项目拆到最细就是一句话让用户用自然语言提问系统把它转成SQL或查询逻辑从库里拿数据再整理成答案和图表给回去。这套东西从demo跑到能真正上线中间隔着一条很宽的河基础设施就是架在河上的桥。这篇继续LCODER实战系列的第二篇重点聊我怎么搭这块基础设施的——不是把依赖装完就完事而是搭一套能支撑开发、调试、上线、回归的完整环境体系。系列第一篇我们定了需求边界不做通用的ChatBI平台只做一个能跑通真实业务问题的问数智能体。边界一旦清晰基础设施的搭建方向也就定了。我在这篇文章里会拆开讲五个层面工程骨架和配置体系、大模型接入层、Agent编排运行时、数据访问层、可观测性设施。每一层我都会给出我的选型理由、核心代码片段和踩坑记录希望能给正在搭类似Agent项目的朋友省点时间。1. 先想清楚问数项目的基础设施到底需要什么1.1 架构需求的具体拆解问数项目虽然表面上是一个对话应用但它比普通的聊天机器人多了一层非常关键的责任它要动数据。这意味着基础设施不能只照顾到模型能不能回答还得照顾到回答能不能复现、能不能追责、权限怎么控制、并发怎么扛。我把问数项目的基础设施需求分成五个子块每一个都在架构设计阶段独立评估运行环境层API服务用什么跑、进程怎么管理、本地开发和线上部署怎么统一。模型接入层要支持多个大模型以后台切换不能绑死一家厂商还要支持流式输出保证用户的等待体验。Agent编排层问答链路不是一次线性调用它可能是判断意图→选表→写SQL→查库→总结结论这样的多步决策每一步都要可观测、可回滚。数据访问层连接池管理、Schema信息缓存、查询超时控制、敏感表拦截。可观测与评估层每条问答完整的链路追踪、SQL生成结果评估、回归测试集。这五块搭好了后面做Agent逻辑才不用回头补窟窿。我见过不少团队先把Agent流程写在Jupyter Notebook里跑通了再想上线的事结果一接真实数据库就各种连锁崩溃。基础设施先行的意义就是让后续的Agent开发在稳定轨道上跑而不是边开火车边铺铁轨。1.2 技术选型的三个判断原则这次选型我给自己定了三条硬性原则。第一可替换性。大模型领域一个月一个样今天用GPT跑得好明天可能就要换成国内模型或者其他开源模型。所以模型接入层必须抽象成网关业务代码不能直接依赖某个厂商的SDK。第二可观测性。Agent的决策链路比传统接口长得多用户说一句帮我看看本月华东区的销售额跟上月比怎么样系统内部可能进行多次模型推理。每一步的输入输出都要有记录否则线上出问题根本没法定位。第三工程惯性要小。团队对LangChain/LangGraph的生态相对熟悉社区资料多遇到问题能快速找到答案。开新坑选个小众编排框架可能很酷但后续招人、维护的成本都得自己吞。基于这三条原则我确定的技术栈如下后端框架Python 3.11 FastAPI轻量、异步友好、天然支持流式输出。Agent编排LangGraph状态图模型适合问数这种多步决策链路。数据库访问SQLAlchemy 2.0 psycopgPostgreSQL专用驱动连接池统一管理。状态存储与缓存Redis PostgreSQL。PostgreSQL存对话状态的持久化快照Redis做短周期缓存。可观测性Langfuse开源版用来做追踪和评估数据完全在自己手里。部署Docker Compose起步先不上K8s把基础设施边界清晰地跑起来再说。这套选型的核心逻辑是每一样东西都负责一块明确的能力且不让任何一层绑死技术栈。哪怕后面把LangGraph换成Spring AI只要编排层接口设计得好迁移代价是可控的。2. 工程骨架与配置体系搭建2.1 项目目录与初始化我用的是一个偏模块化的单体仓库结构。问数项目这个体量一上来就拆微服务是给自己找麻烦但把所有代码塞进几个大文件同样会出问题。折中的方案是单体应用清晰的分层目录。我的目录结构大致长这样lcoder-qabot/ ├── app/ │ ├── main.py # FastAPI入口 │ ├── core/ # 配置、常量、全局组件 │ │ ├── config.py # pydantic-settings 配置加载 │ │ ├── logging.py # 日志初始化 │ │ └── db.py # 数据库引擎与连接池 │ ├── llm/ # 大模型接入层 │ │ ├── gateway.py # 模型网关 │ │ ├── prompts.py # Prompt管理 │ │ └── streaming.py # 流式处理 │ ├── agents/ # Agent编排层 │ │ ├── graph.py # LangGraph图定义 │ │ ├── state.py # 状态schema │ │ └── nodes/ # 各个节点实现 │ ├── tools/ # 工具层 │ │ ├── schema_lookup.py # 表结构查询 │ │ ├── sql_executor.py # SQL执行 │ │ └── registry.py # 工具注册表 │ └── api/ # API路由层 │ └── chat.py # 对话接口 ├── tests/ # 测试与评估集 ├── docker-compose.yml ├── .env.example └── pyproject.toml这样分层的核心意图是每一层只依赖它下面的一层。API路由不直接碰数据库Agent节点不直接调模型厂商SDK工具的注册与执行完全解耦。这不是为了好看而是为了让替换某个环节不需要翻遍整个代码库。初始化踩过的一个坑Python 3.11LangChain相关的依赖之间版本冲突特别频繁。我强烈建议一开始就使用虚拟环境锁依赖不要裸装在系统Python里。我自己用uv管理的依赖速度比pip快不少锁文件也干净。2.2 配置管理如何做到分环境不混乱问数项目的配置项比普通Web服务多得多除了常规的数据库地址、Redis地址还有模型厂商的Key、模型名称、温度参数、Agent最大迭代次数、SQL超时时间、是否启用追踪等等。这些配置混在一起很容易变成一场灾难。我用pydantic-settings做统一配置管理核心思路是用环境变量驱动按环境提供不同的.env文件。# app/core/config.py from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): # 应用 app_name: str lcoder-qabot environment: str dev # 数据库 database_url: str postgresql://user:passlocalhost:5432/qabot db_pool_size: int 10 db_max_overflow: int 20 # Redis redis_url: str redis://localhost:6379/0 # 模型网关 llm_provider: str anthropic llm_model: str claude-sonnet-4-20250514 llm_temperature: float 0.1 llm_timeout: int 60 # Agent agent_max_steps: int 8 sql_timeout: int 15 # 可观测性 langfuse_enabled: bool True langfuse_public_key: str langfuse_secret_key: str langfuse_host: str http://localhost:3000 model_config SettingsConfigDict( env_file.env, env_file_encodingutf-8, extraignore ) settings Settings()这里有个小细节容易被忽略extraignore这个设置决定了配置里出现多余字段时不会直接报错。多人协作时A同学加的配置项还没有同步到B同学的本地.env文件如果没有这个设置服务直接起不来非常影响效率。另外一个重要原则机密信息永远不要写进代码或默认值里。模型厂商Key、数据库密码只通过环境变量注入.env文件加入.gitignore只提交.env.example作为模板。我在项目里专门写了个启动脚本启动时会检查关键配置项是否为空避免服务带着空Key启动然后在运行时才报错。2.3 数据访问层连接池是问数项目的第一道门槛问数项目跟普通CRUD应用最大区别在于它的数据库访问模式是低频高压。用户不会像订单系统那样每秒几百次改数据但每次问数都可能触发一条复杂聚合SQL跑几秒甚至十几秒。这种模式下一旦连接池管理不好几个用户同时问数就能把数据库连接打满。我用SQLAlchemy统一管理数据库连接参数配置上踩过几次坑之后目前用这套相对稳的配置# app/core/db.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from app.core.config import settings engine create_engine( settings.database_url, pool_sizesettings.db_pool_size, max_overflowsettings.db_max_overflow, pool_pre_pingTrue, pool_recycle1800, pool_timeout30, ) SessionFactory sessionmaker( bindengine, autoflushFalse, expire_on_commitFalse )几个参数我说明一下pool_pre_pingTrue每次从池里取连接之前先探活一下。数据库重启、网络闪断后不会把坏连接交给业务代码。这在长连接场景下能省掉大量诡异报错。pool_recycle1800连接超过30分钟强制回收重建规避数据库防火墙空闲超时踢掉连接的问题。pool_timeout30当连接池所有连接都在忙时等待30秒后放弃。问数项目的SQL本身可能就跑很久如果连接池等待时间太短并发稍高就会误报无法获取连接。还有一个必须提的坑SQLAlchemy的session不能跨线程用。Agent的有些节点可能是异步执行如果session在线程间传递会出现看似随机、实则必现的报错。我的做法是每个节点或者每次工具调用都从SessionFactory新建session用完即关。这条经验是拿一个通宵排查换来的希望看这篇文章的人不用再踩一遍。3. 大模型接入层让换模型变成改配置的事3.1 模型网关的接口设计问数项目对模型能力有强依赖但架构上必须把模型当作可插拔组件来对待。原因很简单模型能力迭代太快今天最强的模型三个月后可能就被同行追上而且企业客户可能会提出必须用私有化部署的模型这类要求。如果代码里写满了OpenAI SDK调用到时候替换就是大手术。我封装了一个轻量网关用统一的chat接口封装不同模型厂商的调用差异# app/llm/gateway.py from abc import ABC, abstractmethod from typing import AsyncIterator, Optional class ChatMessage: def __init__(self, role: str, content: str): self.role role self.content content class BaseLLMClient(ABC): abstractmethod async def chat_stream( self, messages: list[ChatMessage], temperature: Optional[float] None, ) - AsyncIterator[str]: ... class AnthropicClient(BaseLLMClient): def __init__(self, api_key: str, model: str): self._client Anthropic(api_keyapi_key) self._model model async def chat_stream(self, messages, temperatureNone): # 将统一的ChatMessage格式转换为Anthropic格式并流式返回 ... class OpenAICompatClient(BaseLLMClient): def __init__(self, api_key: str, model: str, base_url: str None): # 兼容OpenAI、DeepSeek、Moonshot、Ollama等 ... def create_llm_client(provider: str) - BaseLLMClient: 工厂方法根据配置项创建模型客户端 ...网关层有一个设计细节值得多说两句流式接口必须是AsyncIterator[str]。问数项目的用户体验非常依赖于边说边出的效果用户看到第一个字开始输出心理等待时间会显著缩短。如果网关层返回一个完整的字符串等到全部token生成完才响应体验会很差而且调试时看不出模型卡在哪个环节。3.2 Prompt和上下文管理的基础设施问数项目的Prompt比普通聊天复杂它通常包含几部分系统指令、表结构信息、用户问题、对话历史、以及中间步骤的临时结果。把这些内容拼在一起如果不用一套管理机制代码很快就全是字符串拼接加引号的灾难现场。我的做法是把Prompt模板独立成文件并且在模板里明确标注注入点# app/llm/prompts.py from string import Template SQL_GENERATION_PROMPT Template( 你是一个资深数据分析师需要根据数据库表结构回答用户的业务问题。 数据库表结构如下 $schema_info 对话历史 $chat_history 用户当前问题$user_question 请遵循以下约束 1. 只使用表结构中出现的表和字段。 2. 如果需要多表关联必须明确JOIN条件。 3. 对于时间范围描述先转换为标准SQL条件。 4. 输出必须是合法SQL不要加多余解释。 )模板化的好处至少有三层第一Prompt的修改不需要改代码逻辑运营或算法同学直接调文件就行第二模板可以带版本号方便对比不同版本Prompt的效果差异第三模板文件可以纳入代码评审流程改Prompt跟改代码一样有迹可循。上下文管理的核心问题是长度控制。一个真实的问数场景表结构信息动辄几千token对话历史再累积几轮很容易超出模型上下文窗口或者让推理质量断崖式下跌。我做一个了简化的裁剪策略表结构信息优先保留与当前问题关键词相关的表和字段对话历史只保留最近两轮中间过程数据比如生成的中间SQL、查询结果做摘要化处理后再注入下一轮。这套策略不是最优解但相当实用。3.3 流式输出与中断控制流式接口具体怎么暴露到前端是基础设施层要解决的一个细节。我直接用FastAPI的StreamingResponse SSE协议实现起来干净利落。# app/api/chat.py from fastapi import APIRouter from fastapi.responses import StreamingResponse from app.llm.gateway import create_llm_client from app.core.config import settings router APIRouter() router.post(/v1/chat/stream) async def chat_stream(payload: ChatRequest): client create_llm_client(settings.llm_provider) async def event_generator(): # 这里会通过Agent编排层走到具体的模型调用 async for token in client.chat_stream(messages): yield fdata: {token}\n\n return StreamingResponse( event_generator(), media_typetext/event-stream, headers{Cache-Control: no-cache, X-Accel-Buffering: no}, )这里有一个全网搜索都容易忽略的配置项X-Accel-Buffering: no。如果不加这个响应头Nginx默认会缓冲响应内容导致前端收到的不是流式数据而是等到模型生成完一大段才一次性推送。我在本地接口测试一切正常部署到服务器后前端反馈没有流式效果排查半天才发现是Nginx缓冲在中间作祟。模型调用的超时和中断控制也要在一开始就设计好。问数场景里Agent可能要多次调用模型如果单次调用都不设超时一旦模型侧卡住整个请求会悬挂在那里连接池和线程全部被占死。我统一封装了asyncio的wait_for单次模型调用超时时间默认60秒可以在配置项里按场景调整。4. Agent编排运行时状态图不是花架子4.1 为什么用LangGraph而不是纯LangChain Chain问数项目的决策链路不是一条直线。用户问本月和上月的销售对比Agent首先要判断需要哪些表然后生成SQL执行后看结果有没有问题没问题再总结成回答。这个链路里存在分支、循环、甚至可能返工——SQL执行报错时Agent需要带着错误信息重新生成一次SQL。传统的Chain线性链搞不定这种动态路径而LangGraph的状态图模型天生适合表达这类逻辑。我在代码里把Agent流程定义成一个状态图每个节点负责一件事节点之间通过共享状态传递数据# app/agents/graph.py from langgraph.graph import StateGraph, END from app.agents.state import QABotState from app.agents.nodes import ( parse_question_node, generate_sql_node, execute_sql_node, generate_answer_node, ) graph StateGraph(QABotState) # 添加节点 graph.add_node(parse_question, parse_question_node) graph.add_node(generate_sql, generate_sql_node) graph.add_node(execute_sql, execute_sql_node) graph.add_node(generate_answer, generate_answer_node) # 添加边 graph.set_entry_point(parse_question) graph.add_edge(parse_question, generate_sql) graph.add_edge(generate_sql, execute_sql) # 条件边SQL执行失败时回到generate_sql重试超过次数就终止 graph.add_conditional_edges( execute_sql, lambda state: retry if state.sql_error and state.retry_count 2 else answer, {retry: generate_sql, answer: generate_answer}, ) graph.add_edge(generate_answer, END) app graph.compile()拆到节点粒度之后每个节点的输入输出都能单独观测、单独测试、单独调试。我在开发过程中经常只针对某个节点写单测比如在给定的表结构下generate_sql节点能否生成正确SQL这在传统的Chain式代码里几乎不可能做到。4.2 状态管理与跨会话持久化问数项目必须支持多轮对话比如用户先问上个月销售额再追问那华东区呢第二句的语义依赖第一句的上下文。Agent状态里不仅要存当前问题的推理过程还要保留跨轮次的信息。我用LangGraph内置的State和Checkpointer机制来实现。定义状态时明确哪些字段需要在多轮之间持久化哪些是单轮临时数据# app/agents/state.py from typing import TypedDict, Optional, Annotated from operator import add class QABotState(TypedDict, totalFalse): # 跨轮持久化的字段 user_id: str session_id: str chat_history: list[dict] # 单轮推理字段 question: str schema_info: str sql: Optional[str] sql_error: Optional[str] retry_count: int query_result: Optional[list[dict]] answer: Optional[str]Checkpointer的作用是把每一步的节点输出落盘这样进程重启后会话状态还能恢复。我把Checkpointer配到PostgreSQL做成持久化存储。这么做有取舍对比Redis纯内存存储PostgreSQL的延时会高一些但换来了更可靠的恢复能力。问数项目里的会话状态价值密度高丢一节用户可能就要重新说半天背景值得用慢一点的方式存下来。状态设计的另一个坑不要在状态里塞大的中间结果。之前我把查询结果全表带进状态里结果LangGraph每次做状态的持久化快照都会序列化一次完整结果集几万行的查询结果直接导致写库延迟暴涨。现在的做法是SQL执行节点的大结果集只保留摘要和行数回写到状态全量数据另外存Redis或临时文件并给一个短过期时间。4.3 工具注册机制与MCP协议预留问数项目的工具不止执行SQL这一个。实际业务里还有查表结构查字段注释查指标定义看是否有敏感字段等动作。把这些工具直接写死在Agent节点里每加一个都要改动图结构非常繁琐。我用一个工具注册表统一管理每个工具只需要暴露名称、描述和调用函数# app/tools/registry.py from typing import Callable, Any class Tool: def __init__(self, name: str, description: str, fn: Callable): self.name name self.description description self.fn fn self.schema self._build_schema() def _build_schema(self): # 通过inspect获取函数的参数定义生成模型可读的JSON Schema ... class ToolRegistry: def __init__(self): self._tools: dict[str, Tool] {} def register(self, tool: Tool): self._tools[tool.name] tool def get_schemas(self) - list[dict]: return [t.schema for t in self._tools.values()] def execute(self, name: str, **kwargs) - Any: tool self._tools[name] return tool.fn(**kwargs) registry ToolRegistry() registry.register(Tool(lookup_table_schema, 查看数据库表结构, lookup_table_schema)) registry.register(Tool(execute_sql, 执行SQL查询, execute_sql_tool))现在Agent领域MCP协议发展很快很多通用工具将来可能会通过MCP服务暴露出来。我在工具层做了预留注册表内部不做任何和MCP相关的耦合但工具的name和description命名规范严格对齐MCP工具定义标准。将来真需要接MCP服务器时只需要写一个适配层把MCP工具映射进这个注册表就行不用动Agent图结构。还有一个经验之谈工具的描述信息直接影响模型能否正确选对工具。用词要具体要写清楚输入参数的边界。比如execute_sql这个工具描述里如果不写只允许执行SELECT查询模型有可能在Agent推理时试图通过这个工具执行DDL语句酿成事故。工具层必须做参数校验和白名单校验不能完全信任模型的工具调用输出。5. 可观测性Agent项目的基础设施不能只看日志5.1 全链路追踪的埋点设计传统后端的日志-指标-链路追踪三板斧在Agent项目里不太够用。Agent的每个节点都可能调用模型一次用户请求会产生多个Span而且这些Span之间的因果关系比普通RPC调用复杂得多。我看过很多Agent项目出问题时只有一行行的print日志根本无法还原模型为什么走错分支。我在基础设施层直接集成了Langfuse。它跟LangGraph有官方集成能自动捕获每个节点的输入输出、token消耗、延迟以及整个Agent链路的时间线。# app/agents/graph.py from langfuse.callback import CallbackHandler langfuse_handler CallbackHandler( public_keysettings.langfuse_public_key, secret_keysettings.langfuse_secret_key, hostsettings.langfuse_host, ) # 编译图时传入回调 app graph.compile(checkpointercheckpointer)在Graph执行时把Langfuse的回调挂到LangGraph的编译结果上每次调用Agent时langfuse_handler就会自动记录整个链路的trace信息。用户问一个华东区销售环比的问题我能在这个追踪界面里看到Agent先查了哪些表结构、给了模型什么上下文、模型生成了什么SQL、SQL执行结果是什么、最终回答是怎么生成的。这个能力在问题排查时是决定性的。有一次线上用户反馈问某张表的某个指标系统答非所问我顺着trace一看发现模型在生成SQL环节把表名想错了再看schema_info的注入内容发现是我之前设计的表结构裁剪逻辑把真正的表漏掉了。如果没有链路追踪这个问题至少要定位一晚上有了trace十分钟就能看出问题出在哪个节点。5.2 评估集与回归测试基础设施问数项目还有一个基础设施层面的模块最容易被忽略评估。Agent系统的最终效果不是一个功能对不对的问题而是一个在100个真实问题上能达到多高的准确率的问题。没有评估基础设施Agent的迭代就没有抓手要么不敢改要么瞎改。我建了一套轻量评估流水线准备一组带标准答案的测试问题集每次代码或Prompt变更后跑一遍全量回归自动比对输出结果的质量。评估指标的设置要根据环节分开考量评估环节指标判定方式Schema选择表/字段召回率生成的SQL使用字段与标准答案做交集比对SQL生成语法正确率能否通过解析器或直接执行不报错SQL结果结果正确率与标准答案结果集做DM比对忽略行顺序最终回答忠实度大模型评估回答是否基于查询结果、有无幻觉每一条评估结果我会保留对应的trace链接这样任何一个用例挂了之后都能一键跳转到Langfuse里查看当时Agent是怎么推理的。这套机制用下来的效果非常明显它逼着我把每个Prompt的改动都变成可验证的动作而不是我感觉改完效果好了一点。专项数据的准备有一点难度采集真实用户的问题作为种子数据再人工标出期望的SQL和结论。这个过程费时间但收益巨大。前期哪怕没有100条先准备20条核心业务问题也能挡住大部分回归风险。6. 搭建过程中踩过的坑和补救措施6.1 典型问题排查速查表基础设施搭建过程中我遇到了一堆问题有一个查一个整理出一张速查表在这里直接分享出来应该能帮你省掉不少排查时间症状根因解决方法部署后流式输出不生效Nginx缓冲了响应添加响应头X-Accel-Buffering: no连接池报错timeout连接池大小与并发估算不足动态调整pool_size加SQL超时控制避免慢查询占死连接数据库重启后大量报connection broken连接池没有保活开启pool_pre_pingTrue并配置pool_recycleAgent生成SQL时表名不在上下文里Schema裁剪策略把相关表过滤掉了用关键词模糊匹配同时扩大候选表范围宁可多给不能少给模型返回空结果或截断上下文超过窗口或输出长度限制裁剪历史、摘要中间结果调大max_tokens多轮对话丢失上下文LangGraph状态没持久化配置PostgreSQL Checkpointer并传session_id请求偶发500无法复现Session跨线程使用统一在节点内部创建Session不在线程间传递当前时间在Agent的推理里是偏差的模型不知道当前日期在系统Prompt中动态注入当前时间特别是问本月今年这类相对时间工具输出过大状态持久化变慢把全量结果放进了Graph状态状态里只存摘要和行数全量数据另做短时缓存6.2 两个我认为最值得提前做的事第一件事是先做观测后做功能。我不是在说漂亮话而是切身教训。最初我先写了Agent的核心链路才接Langfuse结果想回看之前的运行时状态时才发现那些历史调用完全没有记录。基础设施里可观测性这块应该第一时间就搭好哪怕后面Agent逻辑会推翻重写观测能力也可以一直复用。第二件事是给所有外部依赖设置明确的超时和重试策略。问数项目依赖的模型API、数据库、Redis任何一环都可能抖动。我在基础设施层统一封装了超时控制重试只对幂等操作启用而且重试次数严格控制。这里有一个血的教训一次模型API临时抖动我配置了简单粗暴的无限重试结果把模型厂商的限流直接打爆反而影响了其他正常请求。现在所有外部调用的重试上限是2次每次重试退避时间在2秒以上宁可让这一次请求失败也不能拖垮整个系统。7. 最后再聊几句实用建议基础设施搭完之后我的一个直觉感受是问数项目真正难的从来不是模型能力而是模型外面这一圈支撑系统。模型生成的SQL再漂亮没有稳定的连接池、没有可追溯的链路日志、没有可回归的测试集它就是个实验室里的玩具。这套基础设施的价值在项目上线前显现不出来但一旦开始面对真实用户、真实数据、真实故障你会发现每一块砖都没白砌。给正在搭同类项目的朋友一个建议基础设施的搭建顺序很重要我按工程骨架→配置体系→数据访问→模型接入→Agent编排→可观测性这样的顺序走下来整个链路比较顺。不要一上来就埋头写Agent状态图先花两天把地基夯实后面写业务逻辑的速度会快上一倍不止。下一篇实战记录里我会继续拆解问数项目智能体的核心链路实现——也就是Agent从接收问题到输出答案的具体节点设计和Prompt调优过程。