先说明一句这个系列写到第(2)篇实验性质非常明显第(1)篇我们把问数项目的边界梳理清楚了知道它本质上是一个NL2SQL的Agent应用——用户用自然语言提问Agent负责理解问题、拆解查询意图、生成SQL、执行查询、最后把结果组织成可读的回答。但真到搭建的时候我犯了一个很典型的错误上来就写功能先做生成SQL的Prompt再调模型结果发现模型连数据库都摸不着、Schema信息塞不进上下文、连数据库的账号权限都没想好、日志里根本看不到一次完整调用链。所以这一篇我决定把基础设施搭建当成独立主题老老实实写清楚。AI Agent应用的基础设施不是配个数据库连接池、挂个缓存就完事。它的运行路径是动态的模型会根据不同的问题决定调用哪个工具、访问哪张表、拼接什么查询。基础不夯实后面每加一个能力都得回来动地基。这篇我会按照自己实际搭建问数Agent时的顺序完整覆盖选型、模块落地、最小闭环、避坑这几块目标是让你看完之后能照着搭出一套可运行、可扩展、可观测的Agent地基。1. 先把基础设施这个词拆开问数Agent的地基到底包含什么很多教程会直接把LangGraph或者Spring AI拉起来写个Demo跑通了就认为基础设施做完了。但Demo能跑和地基扎实是两回事。我先用一个实际场景来说明用户问和上个月相比华东区的营收变化了多少这个问题背后涉及多少基础设施的协作。模型需要知道数据库里有哪些表、每张表的字段含义需要调用一个查询工具去执行SQLSQL执行完返回的结果集可能很大如何截断或摘要如果第一次生成的SQL报错Agent还要能重试整个过程需要可追踪、可评估。这些都不是调一次模型能解决的而是基础设施层的问题。1.1 传统后端的地基和Agent的地基不是一回事传统后端比如订单服务依赖项是确定的数据库、缓存、消息队列。所有接口的调用路径在写代码时就已经固定了基础设施要做的就是管好连接、保证可用性。Agent应用不同它的核心执行路径由模型动态决定。模型在每一轮都会输出下一步调用哪个工具的决策工具返回结果后又要被塞进新的上下文里让模型继续决策。这条链路意味着基础设施必须额外解决几个问题状态管理多轮对话、工具调用、SQL执行中间结果都需要在Agent各节点之间传递。上下文组装数据库Schema、工具描述、历史消息、当前问题每一轮都要拼成模型能理解的上下文。动态工具路由模型决定调用哪个工具基础设施要负责把模型输出映射到真实工具的执行。全链路可观测模型调用、工具调用、SQL执行这些环节跨了多个组件出错时必须有链路追踪能力。这些问题在传统后端里几乎没有对应物。所以我搭建基础设施时没有按照传统MVC的思路去组织而是按Agent运行所需的支撑能力来划分模块。1.2 问数项目必须垫实的五个模块经过几次推翻重来我认为问数项目的基础设施至少包含五块缺一块后期都会很难受模块职责不做的后果配置与密钥管理管理模型API Key、数据库连接串、环境差异环境切换痛苦密钥泄露风险LLM接入层统一模型调用入口支持切换模型、记录Token换模型几乎要重写业务代码Agent运行时状态流转、节点编排、工具注册与调度业务流程混乱多轮交互无法维护数据源连接层数据库连接管理、Schema同步、安全控制查询工具无法安全稳定执行SQL可观测性日志、链路追踪、Token成本统计出问题只能靠猜优化没有依据这五块的落地顺序也有讲究。我的建议是先做配置管理再做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协议现阶段先留扩展位MCPModel 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_encodingutf-8, extraignore, ) # 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 ( fmysqlpymysql://{self.db_user}:{self.db_password} f{self.db_host}:{self.db_port}/{self.db_database} ) settings Settings()这里有个细节extraignore很关键。因为.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( modelmodel or settings.llm_model, temperaturesettings.llm_temperature if temperature is None else temperature, api_keysettings.llm_api_key, base_urlsettings.llm_base_url, max_tokenssettings.llm_max_tokens, timeout30, max_retries2, )用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, totalFalse): 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节点负责生成SQLexecute_sql节点负责执行SQLgenerate_answer节点负责生成最终回答。节点函数签名统一是(state) - partial_statedef generate_sql_node(state: AgentState) - dict: sql llm_service.generate_sql( questionstate[question], schemaget_cached_schema(), contextstate.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(checkpointermemory)MemorySaver是LangGraph内置的内存检查点器它让Agent在每次状态变更后保存快照后续既能支持多轮对话的上下文回溯也能支持人工介入审批、断点恢复。对问数项目来说尤其重要的一点是当SQL执行超时或报错时系统可以基于保存的状态回退到某个节点重新执行而不是让整个会话丢给用户重新发起。3.4 数据源连接与Schema同步问数基础设施里最容易轻敌的是数据库这一层。很多人以为连上数据库、能执行查询就完事其实远不止。我用SQLAlchemy作为数据库访问层工程上做了三件事。第一连接池配置。Agent是多轮交互应用用户可能连续查好几次如果每轮去查询都新建数据库连接资源开销大且容易被打满。我配置了连接池from sqlalchemy import create_engine engine create_engine( settings.database_url, pool_size5, max_overflow10, pool_recycle3600, pool_pre_pingTrue, connect_args{connect_timeout: 5}, )pool_pre_pingTrue会在从连接池取出连接前先探活避免拿到一个已经断开的死连接。这个参数强烈建议加上。第二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_asciiFalse))在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}, configconfig) return result.get(answer)调用一次app.invokeLangGraph会自动按照状态图执行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都塞进PromptToken消耗会非常夸张。我遇到过一个库十几张表、每张表二三十个字段全量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( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10), retry_on_resultlambda 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生成质量优化、多轮对话的追问策略、以及更复杂的多数据源扩展都是在这个骨架上逐步生长的。基础这层花的时间后面一定会成倍赚回来。