企业官网建设流程全解析

问数这个项目，说白了就是让业务人员用自然语言直接查企业数据库——"上个月的华东区销售额是多少""同比环比怎么变化"这类问题，AI 自动转成 SQL、跑完返回结果。听起来不复杂，真正动手做才发现，Agent 本身只是最上面那层皮，底下全是基础设施的活儿。这篇是问数项目智能体搭建的第二篇，我先把第一层地基给你夯实：环境、框架、模型接入、存储、可观测性，一套能支撑生产级 Agent 跑起来的基础设施到底该怎么搭。

1. 先想清楚：问数 Agent 的基础设施到底包含什么

1.1 问数项目对基础设施的真实需求

很多人一听到"基础设施搭建"，第一反应是买服务器、配数据库、装 Docker。但对 AI Agent 项目来说，基础设施的范畴要大得多，也更"软"一些。我把它拆成四层来看：

第一层是运行时环境，包括 Python 运行环境、依赖管理、环境变量与密钥管理、API 服务的骨架。这一层解决的是"代码在哪里跑、配置从哪里读"的问题。

第二层是模型接入层，包括大模型 API 的封装、多供应商路由、密钥隔离、超时与重试策略。问数项目对模型能力要求不低，既要懂自然语言转 SQL，又要能根据数据库 schema 做推理，所以模型接入层必须设计成可切换、可降级的，不能把命绑在一家供应商身上。

第三层是Agent 编排层，这是问数项目的核心基础设施。Agent 需要拆解用户问题、选择工具、调用工具、观察结果、再决策。我选了 LangChain 生态里的 LangGraph 来做这件事，后面细讲为什么。

第四层是支撑组件，包括会话记忆存储（Redis）、向量库（用于存放表结构说明和 few-shot 样例）、日志追踪与评估系统。这些组件决定了 Agent 能不能记住上下文、能不能追溯每一步决策、能不能持续优化。

1.2 基础设施搭建的目标：不是跑通，而是可控

我自己搭过好几版 Agent 项目，最深刻的体会是：一个 Demo 级的 Agent 基础设施和一个生产级的，差别不在功能，而在"可控性"。跑通一个"你好，帮我查一下销量"的对话很容易，难的是回答错了你能定位是哪一步错了，模型被攻击时你能快速降级，成本失控时你能立刻限流。

所以这篇搭建分享，我会始终围绕四个关键词：可切换、可观测、可回滚、可评估。每一个组件的选型，都要能支撑这四个目标。别小看这个原则，问数这类项目一旦进了企业内网，业务方每天上千个问题砸过来，模型偶尔抽风是常态，基础设施的职责就是要兜住这些不确定性。

2. 技术选型：为什么是 LangGraph 加这套组合

2.1 编排框架：LangGraph 比裸调 API 好在哪

先回答一个很多新手会问的问题：问数项目为什么要用 Agent 框架，直接写个函数调大模型不行吗？

行，但只适合最简单的场景。问数项目的完整链路是：理解用户意图 -> 补充上下文（加入会话历史、企业术语表、表结构说明） -> 生成 SQL -> 检查 SQL 安全性 -> 执行 SQL -> 解释结果。在这个过程中，模型可能第一次生成的 SQL 有语法错误，或者字段名不对，你需要让它"自己看到报错后重试"。这就是一个典型的带状态、带循环的 Agent 流程。

LangGraph 的核心价值，就是把这种流程建模成一张图：节点是"调模型""执行 SQL""校验结果"，边是"成功"和"失败"的条件跳转。它的状态管理机制特别适合问数场景——每一轮对话的中间变量（用户的原始问题、生成的 SQL、执行结果、错误信息）都挂在共享状态上，节点之间通过状态传递数据。这在普通的 LangChain Chain 里做不到，因为 Chain 是线性单向的，无法处理"错了要重试"这种回环。

2.2 模型层：一套接口接多家模型，做好降级预案

选模型的核心指标有三个：NL2SQL 能力、工具调用稳定性、成本。

我目前的主模型用的是 DeepSeek 和通义千问的旗舰版本，两者的 NL2SQL 表现在主流国产模型里都属于第一梯队，工具调用的 JSON 输出稳定性也够用。备选是智谱 GLM 和 OpenAI 的模型。这里要特别强调一个设计：所有模型接入都走统一接口，通过配置切换，而不是写死在代码里。

我的做法是用一个ModelRouter类，内部维护一个模型列表，每个模型配好供应商 SDK、API Key、模型名、超时时间、权重。默认走主模型，主模型连续报错或超时达到阈值，自动切换到备用模型。对于问数项目来说，模型服务不可用的后果比"回答质量稍微差一点"严重得多，业务方不会管你是哪家模型的 bug，只会说系统挂了。

2.3 存储组件：三类存储各司其职

问数项目里涉及三类数据，千万不要混用一个存储：

• 业务数据：用户的原始业务库，通常是企业里现成的 MySQL / PostgreSQL / ClickHouse，Agent 只做查询不改写。
• 语义数据：表结构描述、字段注释、业务术语表、few-shot 样例。这些数据量不大，但对 NL2SQL 的准确率影响极大。我放在 PostgreSQL 的 pgvector 里，统一管理，方便做相似度检索。
• 状态数据：会话历史、Agent 运行状态、临时缓存。放 Redis，读写快、支持 TTL 自动过期。

这里插一句很多人踩过的坑：有些人喜欢把语义数据放 Elasticsearch，把状态数据也塞进去，结果一个 ES 集群承担了所有职责，维护成本极高。对于问数项目这个体量，三套轻量存储完全够用，别过度设计。

3. 一步步搭起来：从空目录到可运行的基础设施

3.1 项目初始化与依赖管理

我推荐用uv来管理 Python 项目，相比于 pip + requirements.txt，uv 的依赖解析速度快到令人感动，而且 lockfile 机制能保证团队环境一致。

# 安装 uv（macOS / Linux） curl -LsSf https://astral.sh/uv/install.sh | sh # 初始化项目 uv init question-agent cd question-agent # 创建虚拟环境并激活 uv venv source .venv/bin/activate # 添加核心依赖 uv add langchain langgraph langchain-openai redis pgvector sqlalchemy asyncpg pydantic-settings structlog

这里有个细节：langchain-openai这个包不仅支持 OpenAI，也支持所有兼容 OpenAI SDK 协议的模型服务商。DeepSeek、通义千问、智谱都提供了 OpenAI 兼容接口，所以一套 SDK 就能接全部，不需要为每家供应商引入独立 SDK。

3.2 配置管理与密钥隔离

千万不要把 API Key 写在代码里或者.env文件里然后提交到 Git。问数项目一旦进入公司内部，密钥管理就成了一件严肃的事。我的习惯是：

• 本地开发用.env文件，但.env必须在.gitignore中排除。
• 提交一份.env.example，里面只放键名不放真实值。
• 生产环境密钥走密钥管理服务（如云厂商的 KMS / Vault），应用启动时从环境变量注入。

配置类我用pydantic-settings来做，它能在应用启动时做类型校验，配置缺了立刻报错，而不是等到调模型时才发现没有 API Key。

from pydantic_settings import BaseSettings class Settings(BaseSettings): # 模型配置 primary_model: str = "deepseek-chat" fallback_model: str = "qwen-max" temperature: float = 0.1 request_timeout: int = 30 # 存储配置 redis_url: str = "redis://localhost:6379/0" database_url: str = "postgresql+asyncpg://user:pass@localhost:5432/biz_db" vector_store_url: str = "postgresql+asyncpg://user:pass@localhost:5432/semantic_db" class Config: env_file = ".env"

注意temperature的设置，问数场景我强烈建议设成 0.1 以下。SQL 生成是精确任务，不是创意写作，温度太高模型就会"发挥"，字段名都能给你编出来。

3.3 模型接入层：路由、重试与降级

这层是基础设施的重头戏。我封装了一个LLMGateway，负责所有模型调用的统一出口，向上层屏蔽模型供应商差异。

import asyncio import logging from typing import Any from langchain_openai import ChatOpenAI logger = logging.getLogger(__name__) class LLMGateway: """统一模型网关：路由、超时、重试、降级""" def __init__(self, settings: Settings): self.settings = settings self._clients = { "primary": ChatOpenAI( model=settings.primary_model, api_key=settings.primary_api_key, base_url=settings.primary_base_url, temperature=settings.temperature, timeout=settings.request_timeout, ), "fallback": ChatOpenAI( model=settings.fallback_model, api_key=settings.fallback_api_key, base_url=settings.fallback_base_url, temperature=settings.temperature, timeout=settings.request_timeout, ), } self._consecutive_failures = 0 async def invoke(self, messages: list[dict], **kwargs) -> Any: # 主模型优先，连续失败 2 次切换到备用模型 if self._consecutive_failures < 2: client_name = "primary" else: client_name = "fallback" logger.warning("主模型连续失败，切换到备用模型") client = self._clients[client_name] # 超时 + 单次重试 try: resp = await client.ainvoke(messages, **kwargs) self._consecutive_failures = 0 return resp except Exception as exc: self._consecutive_failures += 1 if client_name == "primary": logger.warning("主模型调用失败: %s, 切换备用", exc) fallback = self._clients["fallback"] return await fallback.ainvoke(messages, **kwargs) raise

这里有个实战经验：重试逻辑千万别做成无限重试。模型服务如果抽风，基本是批量的，你重试 5 次大概率 5 次全失败，浪费了时间还积压了请求。我一般只做 1 次重试 + 1 次降级，再不行就直接报错，让上层走兜底逻辑。对问数项目来说，"抱歉，查询服务暂时不可用"比"卡顿 30 秒后返回错误"体验好得多。

3.4 Agent 编排：用 LangGraph 搭出可控制的流程

问数 Agent 的图结构，我是这样设计的：

• 节点1parse_query：接收用户问题，结合会话历史做意图识别和问题改写。
• 节点2retrieve_schema：从向量库检索最相关的表结构、字段说明和 few-shot 样例。
• 节点3generate_sql：把原始问题 + 相关 schema + 样例拼接成 prompt，丢给模型生成 SQL。
• 节点4validate_sql：做基础的安全校验，检测是否包含危险语句（如 DELETE、DROP、多语句执行）。
• 节点5execute_sql：在业务库上执行 SQL，限定返回行数。
• 节点6handle_error：执行出错时，把错误信息回传给模型，让它修正 SQL。
• 节点7format_answer：把查询结果整理成自然语言回复。

from typing import Literal from langgraph.graph import StateGraph, END from dataclasses import dataclass, field @dataclass class AgentState: question: str history: list = field(default_factory=list) schema_context: str = "" sql: str = "" sql_error: str = "" query_result: list = field(default_factory=list) answer: str = "" async def parse_query(state: AgentState) -> dict: # 结合会话历史，改写用户问题，补充缺失的条件 ... async def retrieve_schema(state: AgentState) -> dict: # 从向量库检索相关表结构 ... async def generate_sql(state: AgentState) -> dict: # 基于 schema 上下文和问题生成 SQL ... async def validate_sql(state: AgentState) -> dict: # 安全校验 ... async def execute_sql(state: AgentState) -> dict: # 执行业务库查询 ... def route_after_execute(state: AgentState) -> Literal["format_answer", "handle_error"]: if state.sql_error: return "handle_error" return "format_answer" async def handle_error(state: AgentState) -> dict: # 将错误信息反馈给模型重新生成 SQL ... async def format_answer(state: AgentState) -> dict: # 将查询结果用自然语言回复用户 ... graph = StateGraph(AgentState) graph.add_node("parse_query", parse_query) graph.add_node("retrieve_schema", retrieve_schema) graph.add_node("generate_sql", generate_sql) graph.add_node("validate_sql", validate_sql) graph.add_node("execute_sql", execute_sql) graph.add_node("handle_error", handle_error) graph.add_node("format_answer", format_answer) graph.add_edge("parse_query", "retrieve_schema") graph.add_edge("retrieve_schema", "generate_sql") graph.add_edge("generate_sql", "validate_sql") graph.add_edge("validate_sql", "execute_sql") graph.add_conditional_edges("execute_sql", route_after_execute) graph.add_edge("handle_error", "execute_sql") graph.add_edge("format_answer", END)

这个流程设计的核心思路是：让模型只做它擅长的事。生成 SQL、解释错误是模型的强项；但执行 SQL、校验安全性必须由代码控制，绝不能让模型"决定"直接操作数据库。我在图里加了validate_sql这个强制节点，无论模型说什么，都必须经过代码层的安全校验才能触达数据库。

3.5 记忆层：Redis 里到底存什么

问数项目的记忆分两层：短期会话记忆和长期语义记忆。

短期会话记忆存 Redis，用conversation:{session_id}作为 key，TTL 设为 30 分钟。数据结构上，我存的是一个 JSON 数组，元素是{role: "user" / "assistant", content: "..."}。每次新的提问进来，把历史数组取出来，拼接进 prompt。

这里有个很实际的取舍问题：历史消息全量塞进 prompt，很快就会把上下文窗口撑爆。我的做法是只保留最近 6 轮对话，并在存入历史时对查询结果做截断——如果结果超过 1000 个字符，只保留前 800 字加一句"（结果过长，已截断）"。这个设计是问数项目特有的，因为 SQL 查询结果可能是一大张表格，全塞回上下文既浪费 token 又干扰模型判断。

长期语义记忆存在 pgvector 里，存的是表结构的向量化描述。每一张业务表，我会提前生成一段"表说明文本"，内容包括表名、字段名、字段类型、字段含义、业务口径、常见查询示例。然后把这整段文本用 embedding 模型转成向量，查询时根据用户问题做相似度召回，取 top 5 张最相关的表结构，作为模型生成 SQL 的参考。

这一步是问数项目准确率的生命线。裸奔的模型对业务表一无所知，字段名稍微不规范（比如ct、amt、dt这种缩写）它就只能靠猜。有了 schema 检索，模型就像带上了"字典"，生成 SQL 的准确率在我的实测中从 60% 左右直接拉到 85% 以上。

4. 可观测性：别等出问题了才开始加日志

4.1 日志、追踪、评估三件套

Agent 项目有一个痛点：每一步的输入输出都是非确定性的。普通接口出了问题，看调用链日志就能定位；Agent 出问题，可能发生在任意一个节点，原因可能是模型理解错了、工具传参错了、数据库超时了。所以基础设施层面必须从第一天就把可观测性做进去。

日志方面，我用structlog，所有日志输出为 JSON 格式，并且强制关联一个trace_id。这个 trace_id 在请求入口生成，贯穿整个 Agent 运行周期，这样排查问题时，一条链路拉出来全都有。

import structlog logger = structlog.get_logger() logger.info( "sql_generated", trace_id=state.trace_id, question=state.question, sql=state.sql, model=settings.primary_model, )

追踪方面，我接入了 Langfuse，它可以记录 LangGraph 每一步的输入输出、token 消耗和耗时。Langfuse 对 LLM 应用的调试价值有多大呢？这么说吧，有它在，你甚至能看到模型在某一步"脑子里想了什么"（完整的 prompt 和 response），问题定位从"盲猜"变成"看回放"。

评估方面，我维护了一个"回归集"，大概 80 条问数项目的高频问题，每条都标注好了期望 SQL 和期望结果。每次改完 prompt 或模型配置，就跑一遍回归集，看准确率是升还是降。这一步极其关键，因为 LLM 应用的 Prompt 修改经常是"修好一个 bug 引出三个新 bug"，没有回归集，你跟本不知道改动是好是坏。

4.2 成本、限流与防劣化

基础设施还有一项重要职责：保护钱包和服务稳定性。

问数项目调一次模型，走完整流程可能消耗几千到上万 token。业务方一旦养成了"随手问一句"的习惯，日调用量轻松破万，成本很容易失控。我的措施是：

• 按用户限流：每个用户每分钟最多 6 次调用，超出直接排队或不处理。
• 按 Token 限流：每个会话每天累计消耗 token 超过阈值，触发提醒。
• 结果长度控制：强制 SQL 加LIMIT 100，模型输出的分析结果限制在 500 字内。
• SQL 超时：数据库查询超过 10 秒自动取消，防止慢查询拖垮业务库。

这里要重点提醒一下：Agent 的循环必须有硬上限。用 LangGraph 的时候，模型在handle_error和execute_sql之间有可能反复横跳——SQL 错了改、改了再错。我在图上配置了recursion_limit和自建的重试计数器，最多允许修正 2 次，超出后放弃，回复"查不到，请换个说法"。没有这个限制，一个坏问题有可能悄悄调十几遍模型，账单上看着都肉疼。

5. 搭建过程中踩过的坑，直接给你避雷

5.1 工具调用格式不稳定

问数项目里，让模型先调用"查表结构"工具再生成 SQL，是常见的做法。但有一个问题：模型偶尔会返回不符合约定格式的 JSON，比如字段名多了空格、值带了注释。这会导致解析直接抛异常。

我的解决思路是双保险：一是 Prompt 里给一个"一步不差"的调用示例，作为少样本参考；二是解析失败时不要直接报错，把解析错误信息回传给模型，让它重新输出。实测下来，回传错误让模型自纠这一招，能把工具调用的成功率从 95% 拉到 99% 以上。

5.2 上下文塞爆，模型开始"失忆"

问数项目的 schema 上下文有个特点：单张表的说明可能就几百字，召回 5 张表就是两三千字。再加上系统提示词、历史会话、当前问题，一次请求轻松破万 token。一旦超过模型的上下文窗口，轻则截断导致缺字段，重则直接报错。

我的做法是把 system prompt 精简到极致，只保留角色设定和关键规则，所有业务细节都通过检索动态注入。另外，给每条 schema 文本打上权重，确保被召回的永远是最相关的表，而不是一股脑全塞进去。

5.3 连接池耗尽：Redis 和数据库双双宕机

这个问题非常隐蔽。最初我的代码里每次操作 Redis 都新建连接、用完关闭，并发一高，连接池直接被打满，然后整个 Agent 卡死。数据库那边类似，业务方的数据库连接数本来就有限制，Agent 并发查询一上来，别人正常的业务查询也被拖慢了。

修正方案：Redis 用连接池模式，数据库 SQLAlchemy 连接池大小设为 5，溢出后等待而非无限新建连接。同时在线程模型上，把 Redis 操作和数据库查询都放进 Async 模式，避免多线程下的连接竞争。

5.4 模型"脑补"不存在的表和字段

这是问数项目最头疼的问题。业务库几百张表，模型只在 prompt 里看到了一部分，但它在生成 SQL 时就会脑补出一些不存在的字段名。比如原表是order_amt，模型可能写成了order_amount。

防这个问题的思路是：在execute_sql节点里，先解析生成的 SQL 引用了哪些表和字段，然后去核对是否都存在于 schema 信息里。发现不存在的字段，就把"错误提示 + 正确的字段列表"反馈给模型，让它重新生成。这一步能把 SQL 执行的成功率提升一大截，比单纯靠模型"自觉"靠谱得多。

5.5 Prompt 修改没有版本管理

最后再说一个团队层面的坑。Agent 项目的 Prompt 会频繁调整，今天改了系统提示词，明天改了 few-shot 样例，如果没有版本管理，出了问题根本不知道是哪个改动引入的。

我现在要求所有 Prompt 模板都放进prompts/目录，用 Git 管理，文件名带上适用场景，例如generate_sql_v3.jinja。Langfuse 里也可以看到每一条 trace 实际使用的 Prompt 内容，配合 Git 提交记录，就能精确定位每一次质量变化的原因。

写在最后：一个布置基础设施的小建议

搭建这套基础设施，我的建议是不要一次性追求完美。第一版只要能跑通，模型可切换、日志可追踪、SQL 有安全校验，就算达标。真正的复杂问题都是在真实流量打进来之后才暴露的——连接池打爆了才知道要限流，成本超标了才知道要加 token 预算，SQL 老报错才知道要加 schema 核对。这也是我把可观测性放在优先级前列的原因：只有看得见问题，才谈得上优化。下一篇文章，我打算具体讲讲问数项目的语义层搭建——也就是 schema 整理、few-shot 样例的构造和向量检索调优，那是决定问数准确率的核心环节，我们到时候接着聊。