企业官网建设流程全解析

先说明一句，这个系列写到第(2)篇，实验性质非常明显：第(1)篇我们把问数项目的边界梳理清楚了，知道它本质上是一个NL2SQL的Agent应用——用户用自然语言提问，Agent负责理解问题、拆解查询意图、生成SQL、执行查询、最后把结果组织成可读的回答。但真到搭建的时候，我犯了一个很典型的错误：上来就写功能，先做"生成SQL"的Prompt，再调模型，结果发现模型连数据库都摸不着、Schema信息塞不进上下文、连数据库的账号权限都没想好、日志里根本看不到一次完整调用链。

所以这一篇，我决定把"基础设施搭建"当成独立主题老老实实写清楚。AI Agent应用的基础设施，不是配个数据库连接池、挂个缓存就完事。它的运行路径是动态的，模型会根据不同的问题决定调用哪个工具、访问哪张表、拼接什么查询。基础不夯实，后面每加一个能力都得回来动地基。这篇我会按照自己实际搭建问数Agent时的顺序，完整覆盖选型、模块落地、最小闭环、避坑这几块，目标是让你看完之后能照着搭出一套可运行、可扩展、可观测的Agent地基。

1. 先把"基础设施"这个词拆开：问数Agent的地基到底包含什么

很多教程会直接把LangGraph或者Spring AI拉起来写个Demo，跑通了就认为基础设施做完了。但Demo能跑和地基扎实是两回事。我先用一个实际场景来说明：用户问"和上个月相比，华东区的营收变化了多少"，这个问题背后涉及多少基础设施的协作。

模型需要知道数据库里有哪些表、每张表的字段含义；需要调用一个查询工具去执行SQL；SQL执行完，返回的结果集可能很大，如何截断或摘要；如果第一次生成的SQL报错，Agent还要能重试；整个过程需要可追踪、可评估。这些都不是"调一次模型"能解决的，而是基础设施层的问题。

1.1 传统后端的地基和Agent的地基不是一回事

传统后端，比如订单服务，依赖项是确定的：数据库、缓存、消息队列。所有接口的调用路径在写代码时就已经固定了，基础设施要做的就是管好连接、保证可用性。

Agent应用不同，它的核心执行路径由模型动态决定。模型在每一轮都会输出"下一步调用哪个工具"的决策，工具返回结果后又要被塞进新的上下文里让模型继续决策。这条链路意味着基础设施必须额外解决几个问题：

• 状态管理：多轮对话、工具调用、SQL执行中间结果，都需要在Agent各节点之间传递。
• 上下文组装：数据库Schema、工具描述、历史消息、当前问题，每一轮都要拼成模型能理解的上下文。
• 动态工具路由：模型决定调用哪个工具，基础设施要负责把模型输出映射到真实工具的执行。
• 全链路可观测：模型调用、工具调用、SQL执行这些环节跨了多个组件，出错时必须有链路追踪能力。

这些问题在传统后端里几乎没有对应物。所以我搭建基础设施时，没有按照传统MVC的思路去组织，而是按"Agent运行所需的支撑能力"来划分模块。

1.2 问数项目必须垫实的五个模块

经过几次推翻重来，我认为问数项目的基础设施至少包含五块，缺一块后期都会很难受：

这五块的落地顺序也有讲究。我的建议是先做配置管理，再做LLM接入层，然后搭Agent运行时，接着把数据源接进来，最后补可观测性。这样每一层都有下层支撑，调试的时候不会相互干扰。

2. 技术选型上的三次取舍：为什么最终这么定

问数Agent的技术选型，网上能搜到大量方案，但很多方案是"为了用某个技术而用"，不是从项目实际需求出发。我在这里记录自己做的三次关键取舍，理由比结论更重要。

2.1 编排框架：LangGraph还是Spring AI

团队里有人熟悉Java生态，提出可以用Spring AI的Multi Agent模块。这个方案的好处是和现有Java微服务体系无缝衔接。但我最终选了Python + LangGraph。

原因有三：

第一，问数项目的核心是NL2SQL，这个方向上的生态工具、示例代码、开源实现，Python侧最丰富，遇到问题能参考的资料多。Spring AI相对年轻，Multi Agent的能力还在快速演进，遇到边界问题只能自己啃。

第二，Agent编排需要灵活处理循环、条件分支、回溯。LangGraph的核心抽象是StateGraph，把Agent定义成一张带状态的图，节点就是处理函数，边就是状态转移条件。模型决策导致的动态路径，在图里表达起来非常自然。StateGraph天然支持循环——模型认为SQL结果不对，可以回退到生成节点重新生成，这在传统链式编排里要写一堆if-else，在图里就是一个条件边。

第三，LangGraph的状态管理对Python的TypedDict和Pydantic支持得很好，多轮对话的中间状态可以结构化管理。

但如果你是纯Java团队，且Agent业务不复杂，Spring AI也能做，尤其适合直接嵌入已有的Spring Boot服务。我这里的选择只对问数这个场景负责。

2.2 模型接入：统一走OpenAI兼容协议

模型接入我做了个很"笨"但很实用的决定：无论底层用哪个模型，接入层一律走OpenAI兼容协议。

现在国内外主流模型厂商基本都提供了OpenAI兼容的接口，base_url换一下、api_key换一下，同一个ChatCompletion API就能在不同模型之间切换。我在llm.py里做了个工厂，读取配置后返回统一的LangChain ChatOpenAI实例。

这个设计带来的直接好处是：在不同业务阶段可以切换不同模型。开发调试时用便宜的轻量模型，跑正式链路时换更强的主力模型，成本优化和效果升级互不阻塞。问数这类任务对模型的SQL生成能力很敏感，如果没有这层抽象，测模型就得改业务代码。

2.3 MCP协议：现阶段先留扩展位

MCP（Model Context Protocol）近期的讨论度很高，它的目标是统一模型与外部工具、数据源之间的交互协议。对问数项目来说，MCP最大的想象空间是：未来Agent访问不同数据源（数据库、数仓、API、文件）时，不需要各自实现一套连接协议，而是通过统一的MCP Server来提供工具。

我目前的处理方式是：不强行上MCP，但把数据源访问能力封装成标准工具函数，预留MCP Server适配层。原因很简单，现阶段问数项目只需要连接一个OLTP数据库，用LangGraph的工具调用机制就能解决，引入MCP会增加一层网络通信和调试成本。但封装成标准工具函数这个动作很重要，等MCP生态成熟后，可以平滑地把这些工具包成MCP Server暴露出去。

选择永远是围绕场景做的。如果你的项目天生就要面对十几个异构数据源，那MCP值得在第一版基础设施里就引入，它能避免挨个写插件式适配器的痛苦。

3. 五个基础模块的落地过程

这节是真正的实操部分。我会按顺序把五个模块的落地方式写清楚，每一步都给出关键代码和设计理由。

3.1 配置与密钥管理

第一个落地的是配置管理。看起来很基础，但在Agent项目里，配置项比传统项目更多更碎：模型名称、API Key、Base URL、温度参数、数据库连接、Schema缓存时间、SQL执行超时、最大重试次数、Redis地址（如果用Redis做状态存储）……如果用一堆散落的常量，环境一换就乱套。

我用pydantic-settings做配置管理，配合.env文件和环境变量两层覆盖机制。核心配置文件长这样：

from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", extra="ignore", ) # LLM配置 llm_provider: str = "openai_compatible" llm_model: str = "gpt-4o-mini" llm_base_url: str = "" llm_api_key: str = "" llm_temperature: float = 0.0 llm_max_tokens: int = 2048 # 数据库配置 db_host: str = "localhost" db_port: int = 3306 db_user: str = "" db_password: str = "" db_database: str = "" # Agent运行参数 schema_cache_ttl: int = 3600 sql_timeout_seconds: int = 10 sql_max_rows: int = 100 @property def database_url(self) -> str: return ( f"mysql+pymysql://{self.db_user}:{self.db_password}" f"@{self.db_host}:{self.db_port}/{self.db_database}" ) settings = Settings()

这里有个细节：extra="ignore"很关键。因为.env里可能存放了其他语言的配置项或者队友的临时变量，设置为ignore可以避免因为多出的键直接启动失败。

我不建议把密钥写进代码或者配置文件提交到Git仓库。.env文件要加入.gitignore，同时提供一个.env.example模板，把键名列出来、值留空，让新加入的同事复制后填写自己的配置。这是基础设施的第一道防线，省心。

3.2 LLM接入层

LLM接入层在所有模块里最不能省。原因我前面提过：问数项目对模型能力的敏感度很高，测试和正式环境可能用不同模型，甚至同一个环境里要根据任务的复杂度路由到不同模型。

我封装了一个llm.py模块，对外暴露两个函数：get_llm()用于获取ChatOpenAI实例，count_tokens()用于粗略估算Token消耗。

from functools import lru_cache from langchain_openai import ChatOpenAI from config import settings @lru_cache def get_llm(model: str | None = None, temperature: float | None = None) -> ChatOpenAI: return ChatOpenAI( model=model or settings.llm_model, temperature=settings.llm_temperature if temperature is None else temperature, api_key=settings.llm_api_key, base_url=settings.llm_base_url, max_tokens=settings.llm_max_tokens, timeout=30, max_retries=2, )

用lru_cache缓存实例，避免每次调用都重新构造。注意max_retries我只设了2，后面避坑部分会细说为什么不能把这个参数设很大。

模型接口统一之后，我在上层做了个简单的服务类，把"生成SQL"和"生成回答"两个动作封装起来：

class LLMService: def generate_sql(self, question: str, schema: str, context: list[dict]) -> str: llm = get_llm() prompt = build_sql_prompt(question, schema, context) response = llm.invoke(prompt) return extract_sql(response.content)

这样业务层完全不知道底层模型是什么，只依赖LLMService这个抽象。后面如果要引入RAG或者动态Few-shot，也只改这一个类。

3.3 Agent运行时骨架

Agent运行时我选了LangGraph的StateGraph。第一步是把Agent的状态结构定义清楚。问数Agent的状态至少包含以下字段：

from typing import TypedDict, Annotated, Optional from operator import add class QueryToolResult(TypedDict): columns: list[str] rows: list[list] class AgentState(TypedDict, total=False): question: str # 用户当前问题 sql: Optional[str] # 生成的SQL sql_error: Optional[str] # SQL执行报错信息 query_result: Optional[QueryToolResult] # 查询结果 answer: Optional[str] # 最终回答 messages: Annotated[list, add] # 对话历史 retry_count: int # 重试次数

messages用Annotated[list, add]标注，这是LangGraph里Reducer的用法：多个节点都能往messages里追加内容，最终自动合并。这个设计在多轮对话场景特别好用，各个节点不用手动把历史消息传来传去，只需要往共享状态里追加。

然后定义节点。问数Agent至少需要三个节点：generate_sql节点负责生成SQL，execute_sql节点负责执行SQL，generate_answer节点负责生成最终回答。节点函数签名统一是(state) -> partial_state：

def generate_sql_node(state: AgentState) -> dict: sql = llm_service.generate_sql( question=state["question"], schema=get_cached_schema(), context=state.get("messages", []), ) return {"sql": sql, "messages": [{"role": "assistant", "content": f"生成的SQL: {sql}"}]}

StateGraph的构建非常直观：

from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver graph = StateGraph(AgentState) 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.add_edge(START, "generate_sql") graph.add_edge("generate_sql", "execute_sql") graph.add_edge("execute_sql", "generate_answer") graph.add_edge("generate_answer", END) memory = MemorySaver() app = graph.compile(checkpointer=memory)

MemorySaver是LangGraph内置的内存检查点器，它让Agent在每次状态变更后保存快照，后续既能支持多轮对话的上下文回溯，也能支持人工介入审批、断点恢复。对问数项目来说尤其重要的一点是：当SQL执行超时或报错时，系统可以基于保存的状态回退到某个节点重新执行，而不是让整个会话丢给用户重新发起。

3.4 数据源连接与Schema同步

问数基础设施里最容易轻敌的是数据库这一层。很多人以为连上数据库、能执行查询就完事，其实远不止。

我用SQLAlchemy作为数据库访问层，工程上做了三件事。

第一，连接池配置。Agent是多轮交互应用，用户可能连续查好几次，如果每轮去查询都新建数据库连接，资源开销大且容易被打满。我配置了连接池：

from sqlalchemy import create_engine engine = create_engine( settings.database_url, pool_size=5, max_overflow=10, pool_recycle=3600, pool_pre_ping=True, connect_args={"connect_timeout": 5}, )

pool_pre_ping=True会在从连接池取出连接前先探活，避免拿到一个已经断开的死连接。这个参数强烈建议加上。

第二，Schema同步。模型无法凭空知道数据库有什么表、每张表什么含义、哪些字段是业务核心，必须把Schema信息读取出来，以结构化文本形式提供给模型。我用SQLAlchemy的Inspector读取：

from sqlalchemy import inspect, MetaData def load_schema(engine) -> str: inspector = inspect(engine) metadata = MetaData() schema_parts = [] for table_name in inspector.get_table_names(): cols = [] for col in inspector.get_columns(table_name): col_desc = f"{col['name']} {col['type']}" if col.get("comment"): col_desc += f" // {col['comment']}" cols.append(col_desc) schema_parts.append(f"表 {table_name}:\n" + "\n".join(f" - {c}" for c in cols)) return "\n\n".join(schema_parts)

这里我直接用字段注释作为模型的语义信息。所以建表时给字段加COMMENT不是可选项，而是问数项目的基础设施要求。比如此前用CREATE TABLE IF NOT EXISTS orders (id INT ..., total_amount DECIMAL(10,2) COMMENT '订单总金额', region VARCHAR(50) COMMENT '销售区域', created_at DATETIME COMMENT '下单时间')，模型看到这些注释才能把"华东区的营收"映射到region='华东'和SUM(total_amount)。

Schema读取结果缓存到内存或Redis，缓存时间由schema_cache_ttl控制，避免每次请求都重新读取一次数据库元信息。

第三，查询安全。我在基础设施层就强制所有查询走只读账号，限制一次查询最大返回行数、设置执行超时。这些在避坑部分详细展开。

3.5 日志、追踪和Token成本统计

可观测性在Agent项目里不是可选项。传统接口的日志可以靠RequestId串起来，Agent项目则多了一层复杂度：一次用户问题会引发多轮模型调用、多次工具调用，每一轮模型调用都有对应的输入输出和Token数。没有链路追踪，出问题基本只能靠复现。

我自研了一套轻量追踪模块，关键思路是给每一次用户请求生成一个trace_id，用它串联所有环节的日志。

import time import uuid import json import logging logger = logging.getLogger("agent.trace") def trace_step(trace_id: str, step: str, detail: dict) -> None: record = { "trace_id": trace_id, "step": step, "ts": time.time(), "detail": detail, } logger.info(json.dumps(record, ensure_ascii=False))

在Agent各节点开始和结束时埋点：

• 模型调用前记录传入的Prompt大小，调用后记录返回内容和Token消耗
• SQL执行前记录SQL文本，执行后记录行数和耗时
• 节点异常时记录完整堆栈

除了日志，我还单独记录Token统计。问数项目的成本几乎全在Token上，模型每次接收很大一段Schema、工具返回的查询结果，Token量很容易超预期。我必须知道每天每个用户问题消耗了多少Token，才谈得上优化。

def record_token_usage(trace_id: str, model: str, prompt_tokens: int, completion_tokens: int) -> None: cost_map = {"gpt-4o-mini": {"prompt": 0.00015, "completion": 0.0006}} cost = ( prompt_tokens / 1000 * cost_map.get(model, {}).get("prompt", 0) + completion_tokens / 1000 * cost_map.get(model, {}).get("completion", 0) ) trace_step(trace_id, "token_usage", { "model": model, "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "estimated_cost": round(cost, 6), })

不要小看这个设计。没有它，后面做模型切换评估时，你连"候选模型到底比现在的模型贵多少"都说不清楚。

4. 第一次跑通最小闭环："上个季度华东区的营收是多少"走完完整链路

基础设施搭完，我建议先别急着加各种花哨功能，而是强制自己跑通一个最小闭环。这一步能一次性验证配置管理、LLM接入、Agent运行时、数据源连接、可观测性是否真的能协同工作。

4.1 最小闭环代码

核心入口非常简单：

from graph import app # 编译好的StateGraph def ask(question: str, session_id: str, thread_id: str = "default"): config = { "configurable": {"thread_id": thread_id}, "metadata": {"session_id": session_id}, } result = app.invoke({"question": question}, config=config) return result.get("answer")

调用一次app.invoke，LangGraph会自动按照状态图执行：generate_sql->execute_sql->generate_answer。

为了保证"上个季度华东区的营收是多少"这样的问题能被正确回答，我把execute_sql节点的一小段示例贴出来展示工具调用怎么实现：

def execute_sql_node(state: AgentState) -> dict: sql = state.get("sql") if not sql: return {"sql_error": "没有可执行的SQL", "query_result": None} trace_step(state.get("trace_id", ""), "execute_sql_start", {"sql": sql}) try: start = time.time() with engine.connect() as conn: # 统一加LIMIT，防止返回过多行数据 limited_sql = ensure_row_limit(sql, settings.sql_max_rows) result = conn.execute(text(limited_sql)) columns = list(result.keys()) rows = [list(r) for r in result.fetchmany(settings.sql_max_rows)] elapsed = time.time() - start trace_step(state.get("trace_id", ""), "execute_sql_end", { "row_count": len(rows), "elapsed": elapsed }) return {"query_result": {"columns": columns, "rows": rows}} except Exception as e: return {"sql_error": str(e), "query_result": None}

注意我用了ensure_row_limit：所有Agent生成的SQL在真正执行前都会被包上一层LIMIT子句。这个属于安全兜底，因为模型有概率生成不带LIMIT的SQL，一旦表数据量大，容易拖垮数据库。

4.2 调用链全程拆解

一次最小闭环会经历这么几步：

第一步，用户输入问题，LangGraph初始化状态，调用generate_sql节点。这一步Prompt包含三部分：数据库Schema、对话历史、用户问题。模型输出SQL。

第二步，execute_sql节点拿到SQL，在执行前统一做安全处理，然后交给连接池执行。执行结果被结构化成{columns, rows}，写回状态。

第三步，generate_answer节点把原始问题、SQL、查询结果一起交给模型，让模型生成对用户友好的回答。

第四步，全流程日志通过trace_id聚合，可以看到每个节点耗时、每个模型调用的Token数、SQL执行结果。

我第一次跑通时用的是"上个月订单总金额是多少"这个问题，看到返回"上个月订单总金额是1,234,567.89"的那一刻，整个基础链路才算真正落地。这里我强烈建议你也从这类简单聚合查询开始验证，而不是一上来就测"同比增长率"这类需要多表Join、子查询、甚至多步推理的复杂问题。基础设施验证要遵循最小化原则：先把简单链路跑通，再逐步增加复杂度。

5. 基建期的避坑清单：这些问题等写真实业务时才会暴露

最后一部分，列一下我在这个阶段踩过、以及帮朋友排查过的坑。每一条都不是理论推演，是真实线上或准线上环境出现过的。

5.1 数据库只读账号与权限隔离

问数Agent的SQL是模型生成的，没人能保证它永远正确——甚至不能保证它不会生成恶意语句。因此在基础设施层就必须做权限隔离。

我创建了一个只读账号，仅授予SELECT权限：

CREATE USER 'agent_reader'@'%' IDENTIFIED BY '这里填强密码'; GRANT SELECT ON biz_db.* TO 'agent_reader'@'%'; FLUSH PRIVILEGES;

同时，应用层还做了两层保护：一是禁止Agent的SQL里出现;分隔的多语句（防止拼接多条SQL），二是对写操作关键字（INSERT、UPDATE、DELETE、DROP、ALTER等）做黑名单拦截。注意只读账号和关键字拦截是两个独立维度，不能因为有了只读账号就放弃应用层拦截。

5.2 Schema信息太多导致Prompt爆炸

问数项目的一个典型问题是：数据库表很多、单表字段很多，把所有Schema都塞进Prompt，Token消耗会非常夸张。我遇到过一个库，十几张表、每张表二三十个字段，全量Schema文本超过了1万Token，每次生成SQL光Schema就花费了超过一半的上下文。

解决办法是给Schema做分级处理。最常用的方式是给每张表打上领域标签，在生成SQL阶段只把和用户问题相关的表Schema放进Prompt。这个"表选择"动作可以由模型做，也可以由业务规则预设。比如问题是"营收"相关，就优先把订单表、支付表、退款表的Schema塞进去。

另外一个方案是：把Schema信息放到一个工具里，让Agent需要时再按需查询表结构，而不是一开始就全部塞进上下文。这更省Token，但实现复杂度更高，要权衡。

5.3 模型限流引发的连锁抖动

模型API都会有速率限制。Agent任务因为要在一轮交互里调用多次模型API（生成SQL、生成回答、可能还有报错重试），触发限流的概率比普通应用高得多。

我把max_retries设为2而不是默认值或者更大的数字，是因为问数场景对延迟敏感，用户等不起无限重试。重试策略建议用指数退避：

import time from tenacity import retry, stop_after_attempt, wait_exponential @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10), retry_on_result=lambda r: r is None or r.status_code == 429, ) def llm_call_with_retry(func, *args, **kwargs): return func(*args, **kwargs)

另外，如果判断是持续性的限流（比如组织级别的TPM配额不够），与其无限重试，不如直接告诉用户"系统繁忙，请稍后再试"，并把这个异常上报到监控。重试是解决瞬时抖动的，不是解决容量问题的。

5.4 Agent状态在每一步的传递与克隆

LangGraph的状态传递看着简单，但有个坑：如果你在某个节点里直接修改了传入的state对象的内部可变对象（比如字典、列表），而这个对象同时被多个节点引用，就可能出现脏数据。

我的习惯是：所有节点函数都遵循"只读取入参，返回新的partial state"的写法，不要修改传入的state本身。Python的函数传参是引用传递，如果直接修改state["sql"]，在并发或重试场景会有隐患。LangGraph官方推荐的写法也是返回dict，让框架去合并状态，所以这一步做得规范，后面加并发、加人工审批这些高级功能时才不会翻车。

还有一个细节：自定义的trace_id我建议在入口处就生成好，写进初始状态，然后在每个节点里通过state.get("trace_id")取出来，统一埋点。否则追踪日志会变成一盘散沙，根本无法串起来看一次完整的用户请求。

问数Agent的基础设施搭建到这一步，已经具备一个可运行、可观测、可迭代的骨架。后面的SQL生成质量优化、多轮对话的追问策略、以及更复杂的多数据源扩展，都是在这个骨架上逐步生长的。基础这层花的时间，后面一定会成倍赚回来。