《LCODER之AI Agent开发实战一》系列走到了第二篇，今天聊聊基础设施搭建。上一篇讲的是把问数项目的整体需求和技术选型定下来，这篇要动真格的了——在写任何一条Agent业务逻辑之前，先把地基打牢。为什么我要单独拿出一篇来讲基础设施？因为问数项目看起来只是一个“自然语言转SQL查询”的智能体，但真正跑起来之后你会发现，它背后牵扯到模型接入、数据连接、工具调用、状态编排、服务化、权限控制这一整条链路。任何一个环节没搭好，后面调Agent的时候就会反复被同一个坑绊倒。
这篇内容适合正在做AI Agent开发、尤其是做Text-to-SQL或数据问答类项目的同学参考。我会直接给出我在LCODER问数项目中实际使用的方案和配置，包括LangGraph的图编排、MCP协议的工具接入、数据库连接池和元数据管理、FastAPI服务化等。有些地方我会解释为什么这么选，踩过的坑也会一并写出来。
1. 整体架构与基础设施选型思路
1.1 问数项目的核心链路拆解
先把“问数”这件事拆到最细。用户输入一句自然语言问题，Agent要完成的事情是：理解问题意图、结合数据库Schema生成查询语句、执行查询获取数据、把数据整理成自然语言回答返给用户。整个链路看起来简单，但每一个环节都有独立的陷阱。
从基础设施的角度来看，这套链路至少要分成四层来设计。第一层是模型接入层，负责管理各类大模型的API接入、密钥、超时、重试和费用统计。第二层是Agent编排层，负责把“理解-生成-执行-修订”这个过程编排成一个可控的图，而不是一个裸的循环调用。第三层是工具与数据层，负责封装数据库查询能力、Schema读取能力，以及通过MCP协议对接外部工具。第四层是服务化层，负责把Agent引擎暴露成HTTP接口，对接前端对话页面。
问数项目在生产环境下还有一个容易被忽视的第五层，就是安全与权限控制。数据权限不是业务逻辑层面的东西，但它必须在基础设施阶段就设计好，否则等Agent逻辑写完了再想加权限，改造成本会翻倍。
1.2 为什么选择LangGraph作为编排底座
我在热词里看到很多人都在问“LangGraph怎么开发AI Agent实践”，这说明大家确实在关注这个方向。我在LCODER项目中最终选择了LangGraph，而不是直接用LangChain的AgentExecutor，也不是裸调LangChain的链式调用。
原因有三点。第一，LangGraph把Agent的执行流程建模成一张图，节点和边都是显式定义的，这对问数项目太重要了。因为“生成SQL-执行SQL-检查结果”这三个节点天然就是有状态的流转过程，用图来编排比用AgentExecutor的黑盒循环要容易控制得多。第二，LangGraph的State机制允许我在不同节点之间传递结构化数据，SQL语句、查询结果、错误信息、重试次数都可以放在State里，排查问题的时候一目了然。第三，LangGraph原生支持条件边和循环，这正好匹配问数项目中“SQL执行失败需要回炉重写”的场景。
当然，如果你所在团队是Java技术栈，Spring AI的Multi-Agent体系也值得研究，它在金融、政企这类Java生态里落地会更顺畅。但就我个人实践而言，Python技术栈加上LangGraph，在快速迭代和Agent流程的精细控制上体验更好。
1.3 MCP协议在基础设施中的位置
MCP（Model Context Protocol）最近是AI Agent开发领域绕不开的话题。简单来说，MCP是一个开放协议，规定了模型在运行时如何发现和调用外部工具。它解决的核心问题是：以前每个Agent接一个数据源就要写一套自定义工具调用逻辑，接口五花八门，现在大家统一用一套协议来描述“工具”和“资源”。
在问数项目里，MCP的价值非常明确。我可以把数据库查询能力封装成标准的MCP工具，也可以把Schema读取、数据字典查询这些能力做成MCP资源。对外部系统来说，他们只需要按照MCP协议跟我对接，不需要关心我内部用的是MySQL还是PostgreSQL、SQL是手工生成还是LLM生成。
基础设施阶段我做的MCP相关工作包括：配置MCP客户端SDK、定义工具描述文件的schema、搭建本地MCP Server的骨架。这些工作在初期看起来很“虚”，但在后续接入外部数据源、对接其他Agent系统的时候，能省掉大量联调成本。
2. 开发环境与核心依赖准备
2.1 Python虚拟环境与项目骨架
先说开发环境。我用的是Python 3.10以上版本，这个版本对LangGraph和MCP SDK的支持最稳定。虚拟环境用标准库的venv即可，不需要额外装virtualenvwrapper这类工具。
mkdir lcoder-question-agent cd lcoder-question-agent python3 -m venv .venv source .venv/bin/activate pip install --upgrade pip项目骨架我建议按分层来组织目录，不要把Agent业务代码和基础设施代码混在一起。我的实际目录结构是这样的：
lcoder-question-agent/ ├── app/ │ ├── agent/ # LangGraph节点、图定义 │ ├── tools/ # 数据库查询工具、MCP工具注册 │ ├── data/ # 连接池、Schema元数据管理 │ ├── api/ # FastAPI路由、SSE输出 │ └── core/ # 配置、日志、密钥管理 ├── tests/ ├── mcp/ # MCP Server相关 ├── .env.example └── pyproject.toml这么分层的目的是让基础设施代码和业务逻辑解耦。连接池、日志、密钥管理这些是纯基础设施，放在core里；Agent的图定义和节点实现属于编排层，放在agent里；数据库查询工具、MCP工具属于工具层，放在tools里。后续哪一层出了问题，可以直接定位到对应目录，不需要在大杂烩式的代码里翻找。
2.2 核心依赖清单与版本说明
我这里直接给出当前项目的核心依赖，并说明为什么选这些版本。
pip install "langgraph>=0.2.0" "langchain-core>=0.3.0" "langchain-openai>=0.2.0" pip install "mcp>=1.0.0" "sqlalchemy>=2.0.0" "asyncpg>=0.29.0" pip install "fastapi>=0.115.0" "uvicorn[standard]>=0.30.0" "sse-starlette>=2.0.0" pip install "pydantic-settings>=2.0.0" "python-dotenv>=1.0.0"几个值得单独说明的点。SQLAlchemy我特意选了2.x版本，因为2.0的异步会话比1.x的1.4异步体验好很多，而问数项目必须走异步，否则在线程池里调数据库很容易把资源占满。asyncpg是PostgreSQL的高性能驱动，比psycopg2在异步场景下快很多。如果你用的是MySQL，可以把asyncpg换成aiomysql或者asyncmy。
MCP的Python SDK目前还在快速迭代中，1.0版本之后协议才算基本稳定，建议至少用1.0以上。sse-starlette是给FastAPI加SSE流式输出用的，问数项目必须做流式输出，否则用户要等十几秒才能看到第一个字，体验会非常差。
2.3 模型接入与密钥管理
模型接入这块，我的核心原则是：在基础设施阶段就做好多模型切换的准备。问数项目中我同时接入了OpenAI的推理模型和Claude系列模型，也给国产模型留下了扩展位。为什么这么做？因为不同模型在SQL生成任务上的表现差异很大，有些模型对Schema的理解能力强，有些模型的工具调用格式稳定，等到调优阶段再换模型是常事。
密钥管理必须用环境变量加pydantic-settings做配置校验，任何密钥都不能硬编码在代码里。我的.env.example长这样：
OPENAI_API_KEY=sk-xxx OPENAI_BASE_URL=https://api.openai.com/v1 OPENAI_MODEL=gpt-4o ANTHROPIC_API_KEY=sk-ant-xxx ANTHROPIC_MODEL=claude-sonnet-4-20250514 DATABASE_URL=postgresql+asyncpg://user:pass@localhost:5432/lcoder DATABASE_MAX_CONNECTIONS=10 DATABASE_TIMEOUT=5在core/config.py里用pydantic-settings来读取并校验这些配置项。注意一点：不要把所有密钥都放在一个.env里直接分发，至少要做到开发环境、测试环境、生产环境的配置隔离。我在实际项目中还接了一个简单的配置中心，把数据库连接串和模型密钥分开管理，这样数据库密码轮换的时候不需要重新发布Agent服务。
3. 数据层基础设施：打通“问数”的关键
3.1 数据库连接与连接池配置
数据层是整个问数项目最核心的基础设施。你在Agent编排上花再多心思，如果数据库连接池配置不合理，查询一上来就超时，用户体验直接崩掉。
我使用的是SQLAlchemy 2.0的异步引擎，配合asyncpg驱动。连接池配置不是默认值拿来就用的，需要根据你的查询并发量来调。我项目的配置是这样：
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker engine = create_async_engine( settings.DATABASE_URL, pool_size=10, max_overflow=5, pool_timeout=10, pool_recycle=1800, echo=False, ) SessionLocal = async_sessionmaker(engine, expire_on_commit=False)这里重点说三个参数。pool_recycle设为1800秒，是为了避免数据库服务端主动断掉空闲连接，尤其是有些云数据库默认空闲超时很短，你不设置回收时间就会出现“执行几次查询后突然报连接已断开”的诡异问题。pool_timeout设为10秒，是为了避免高并发时请求排队太久，如果10秒拿不到连接就直接报错，而不是无限阻塞住整个Agent。max_overflow设小一点，给突发流量留一点缓冲，但也不要太大，否则连接数打满了数据库先挂了。
3.2 Schema元数据管理
问数项目里，你不可能把整库几十张表的结构一股脑塞给LLM，Token会爆炸，模型也会被无关字段干扰。所以Schema必须提前读取、缓存、并按需裁剪。
我维护了一张元数据缓存表，定期从information_schema同步表信息，包括表名、字段名、字段类型、字段注释。关键一步是给每个字段补充业务语义注释，这部分人工介入是值得的。比如status字段，数据库里存的可能是0和1，但业务含义是“订单已支付”和“未支付”，如果你不把这个注释写清楚，LLM就算看到了字段也不知道怎么用。
获取Schema的SQL长这样：
SELECT table_name, column_name, data_type, column_comment FROM information_schema.columns WHERE table_schema = 'public' ORDER BY table_name, ordinal_position;执行这段SQL的结果会缓存到本地文件或Redis里，会话开始时只把当前问题涉及的表结构拼接到System Prompt中。如何判断“当前问题涉及哪些表”？我的做法是先让LLM做一次表选择的粗筛，把问题相关表挑出来，再把这几个表的详细结构发给它生成SQL。
3.3 数据权限与安全边界
数据权限这件事我必须重点强调。问数项目最常见的翻车现场就是：Agent生成了SQL，一执行，把全表数据捞出来了，或者把不该看的敏感字段查出来了。这不仅是技术上绕不开的问题，也是问数项目能不能上线的生死线。
我的处理方式是三层隔离。第一层是数据库账号隔离，Agent使用的数据库账号只授予只读权限，从机制上杜绝Agent执行INSERT、UPDATE、DELETE这类写操作。第二层是Schema裁剪，根据登录用户的业务角色，在元数据同步阶段就把无权限的表和字段过滤掉，LLM根本看不到这些字段，自然也生成不了对应的SQL。第三层是查询结果脱敏，在Executor节点执行查询后，对敏感字段做掩码处理，比如手机号中间四位打码，再交给LLM去做总结。
这三层隔离在基础设施阶段就要搭好，而不是等到Agent逻辑完成后再补。尤其是第一层的只读账号，如果你没在初期就设置好，后面测试时会发现Agent偶尔会“自作聪明”生成一条带条件的UPDATE语句，一旦执行成功，后果不堪设想。
3.4 通过MCP工具暴露数据查询能力
在LCODER项目中，我没有直接在Agent代码里调用数据库函数，而是把查询能力封装成MCP工具。这样做最大的好处是：Agent和底层数据源解耦，以后接外部数据源的时候，不需要改动Agent的编排逻辑。
MCP Server的骨架用官方Python SDK搭建：
from mcp.server.fastmcp import FastMCP mcp = FastMCP("question-data-server") @mcp.tool() async def execute_read_query(sql: str) -> list: """执行SELECT查询，返回结果集。只允许只读操作。""" if not sql.strip().lower().startswith("select"): return {"error": "only SELECT is allowed"} async with SessionLocal() as session: result = await session.execute(text(sql)) rows = result.fetchall() return [dict(row._mapping) for row in rows] @mcp.tool() async def get_table_schema(table_name: str) -> str: """获取指定表的字段结构。""" ...把查询能力封装成MCP工具之后，LangGraph中工具的注册就变成标准的MCP客户端读取方式，而不是在代码里写死函数。这带来一个额外的好处：Agent可以动态发现工具。如果你新增了一个MCP Server，Agent下次运行时就能自动感知到新工具的列表，不需要重新部署代码。
4. Agent运行时与编排层搭建
4.1 LangGraph图结构与状态定义
编排层是Agent的大脑中枢。我用LangGraph定义了一张有向图，把问数项目的执行流程固化下来。State用的是TypedDict，每个字段都有明确的类型和用途。
from typing import TypedDict, List, Optional from langgraph.graph import StateGraph, START, END class QuestionState(TypedDict): question: str # 用户原始问题 relevant_tables: List[str] # 选出的相关表 schema_info: str # 拼接好的schema信息 sql: Optional[str] # 生成的SQL query_result: Optional[str] # 查询结果 error: Optional[str] # 错误信息 retry_count: int # 重试次数 final_answer: Optional[str] # 最终自然语言回答State这个设计很关键，它本质上是一条流水线。我在调试的时候把每个节点的输入输出都打日志，能看到SQL是在哪个节点生成的、是否经过修订、最后怎么变成了回答。如果你用的是裸循环调用的Agent，排查问题时只能看到一个黑盒，很难定位问题在哪一层。
4.2 Planner-Executor-Reviser模式落地
问数项目的Agent我不想只做一个“生成SQL-执行-返回”的线性流，而是采用了Planner-Executor-Reviser三段式结构。这也是最近社区里讨论的比较多的“三阶段、六泳道、三十个核心节点”这种生产级执行思路的简化版。
Planner节点负责根据用户问题选择相关表，拼接Schema信息，生成初始SQL。Executor节点负责执行SQL，捕获异常。Reviser节点判断：如果SQL执行报错，就根据错误信息修改SQL后重跑；如果执行成功但结果为空，也要判断是SQL写错了还是数据本身就没有，再决定是否改写SQL。
def planner_node(state: QuestionState) -> QuestionState: tables = select_relevant_tables(state["question"]) schema = build_schema_info(tables) sql = generate_sql(state["question"], schema) return { "relevant_tables": tables, "schema_info": schema, "sql": sql, "retry_count": state.get("retry_count", 0) + 1 } def executor_node(state: QuestionState) -> QuestionState: result, err = execute_read_query(state["sql"]) return { "query_result": result, "error": err } def reviser_node(state: QuestionState) -> QuestionState: if state["error"]: state["sql"] = fix_sql(state["sql"], state["error"], state["schema_info"]) return state这里有一个非常重要的细节：重试次数必须严格控制。如果Agent陷入“SQL错误-改写-SQL还是错-继续改写”的循环，不仅消耗大量Token，还会让用户等得失去耐心。我的项目里把最大重试次数设为2次，达到上限就直接返回友好错误提示，并建议用户换个问法。
LangGraph的条件边用来控制流程是继续重试还是走结束节点：
from langgraph.graph import START, END graph = StateGraph(QuestionState) graph.add_node("planner", planner_node) graph.add_node("executor", executor_node) graph.add_node("reviser", reviser_node) graph.add_node("answer", answer_node) graph.add_edge(START, "planner") graph.add_edge("planner", "executor") graph.add_conditional_edges( "executor", lambda s: "reviser" if s["error"] and s["retry_count"] < 2 else "answer" ) graph.add_edge("reviser", "executor") graph.add_edge("answer", END)4.3 上下文管理与重试机制
上下文管理这块，我吃了不少亏。最初版本我把用户的完整历史对话一股脑塞给模型，结果对话超过十几轮之后，请求体积越来越大，模型响应越来越慢，费用也直线上升。
现在的方案是基于窗口的上下文裁剪。保留最近的6轮对话作为短期记忆，更早的历史对话压缩成摘要存放在Context里。对问数项目来说，用户通常不会连续追问20轮，6轮窗口完全够用，而且能有效避免无关历史干扰SQL生成。
重试机制分两个层面。模型调用层面，对于网络超时和限流错误，采用指数退避重试，默认最多4次，每次等待时间递增。业务执行层面，就是上面说的Reviser节点的SQL改写重试，控制在2次以内。这两个层面的重试要分开配置，否则会出现“SQL报错导致模型调用重试，模型重试又生成新的SQL”这种混乱的嵌套问题。
4.4 模型调用与成本控制
模型调用层我做了两个基础设施。一个是统一的LLM调用封装，所有对模型的请求都走同一个异步入口，这样方便统计Token消耗、控制并发、统一设置超时。另一个是模型路由机制，在基础设施阶段就预留好，让不同节点可以配置成使用不同的模型。
在问数项目中，SQL生成我用的是推理能力较强的模型，因为复杂SQL的生成对模型的逻辑推理要求很高。而最后的自然语言回答阶段，我可以用响应速度更快的模型，因为这一步就是简单的格式化输出，不需要太强的推理能力。模型路由在基础设施阶段就配好，后面调优时会发现这个决策非常明智，光Token成本就能省下不少。
5. 服务化：把Agent变成可用的API
5.1 FastAPI接口与SSE流式输出
Agent引擎搭好之后，必须暴露成服务才能被前端调用。我用FastAPI来承载这个服务，主要原因有两个：一是原生的异步支持，和LangGraph的异步执行天然契合；二是有成熟的SSE库可以直接用。
流式输出不是可选项，在问数项目里它是刚需。一个复杂的问数请求，从理解问题到生成SQL、执行查询、生成回答，整个流程可能要花10秒到20秒。如果不做流式输出，用户会在界面上干等十几秒没有任何反馈，然后突然看到一整段回复，体验极差。做成流式之后，用户能看到Agent“正在找表”、“正在生成SQL”、“正在查询数据”这些中间过程逐步刷出来，等待感会大大降低。
我的SSE路由实现大致如下：
from fastapi import FastAPI from fastapi.responses import StreamingResponse from sse_starlette.sse import EventSourceResponse app = FastAPI() @app.post("/agent/query") async def agent_query(payload: QuestionRequest): state = { "question": payload.question, "relevant_tables": [], "schema_info": "", "sql": None, "query_result": None, "error": None, "retry_count": 0, "final_answer": None } async def event_generator(): async for event in graph.astream_events(state, version="v2"): if event["event"] == "on_node_start": yield { "event": "status", "data": {"node": event["name"], "message": f"开始执行{event['name']}节点"} } elif event["event"] == "on_node_end": yield { "event": "status", "data": {"node": event["name"], "message": f"{event['name']}节点完成"} } yield {"event": "done", "data": "complete"} return EventSourceResponse(event_generator())这里用astream_events来监听节点事件。前端可以订阅这些事件，在每个节点完成时更新UI状态，用户就能实时看到Agent的执行进度。
5.2 会话管理与状态持久化
问数项目需要做会话管理，因为用户会就同一份数据连续追问。比如先问“上个月各区域的销售额”，然后再问“华东区域呢”，这里的“华东区域呢”依赖前文提到的“上个月”和“销售额”，没有会话历史的话模型根本理解不了。
会话状态的持久化我在基础设施阶段就直接用了Redis，而不是进程内变量。原因很简单：进程内变量无法跨实例共享，一旦部署多个Agent实例，用户请求被分发到不同实例就找不到上下文了。Redis的方案虽然多了一个依赖，但在横向扩展的时候会非常省心。
会话存储结构大致是Key为session_id，Value为JSON存储窗口内的历史消息和压缩后的摘要。同时设置过期时间，默认保留48小时，避免僵尸会话一直占用Redis内存。
5.3 部署配置与成本控制
部署这块，我用Docker打包Agent服务，单纯是为了环境一致性。真正想提醒的是依赖尺寸问题——Agent服务的镜像里不要装一堆用不到的大模型运行库，几百MB的镜像拉取和启动都会消耗时间，我用的是slim版本的基础镜像加独立安装所需依赖。
成本控制是问数项目上线之后必须面对的问题。每次查询背后都是一次甚至多次模型调用，如果没有预算控制，一个月下来账单会非常吓人。我的做法是在基础设施阶段就埋好Token计数和费用统计，每个会话、每个用户的调用量都记录在案，设置每日限额，超过限额自动熔断。
class UsageTracker: def __init__(self): self.daily_limit = { "token_in": 5_000_000, "token_out": 1_000_000, "cost_usd": 50.0 } async def check_and_track(self, user_id: str, usage: dict) -> bool: ...这套用量统计虽然代码量不大，但在基础设施阶段做好非常关键。等到项目上线发现费用失控再去补，已经损失的钱是追不回来的。
6. 常见问题与排查技巧实录
6.1 Agent反复让SQL执行失败，循环出不来
这是我在开发初期遇到最频繁的问题。Agent生成的SQL不是语法错误就是表名拼错，改了一次还是错，又改一次继续错，一直到重试上限才被迫退出。排查后发现根因是Schema信息给得太粗。我只把表名列名发给了模型，没有给字段注释、类型关联和表之间的Join关系，模型在生成SQL时只能靠猜，猜错的概率自然高。
解决思路是强化Schema的语义丰富度。除了字段名和类型，我会补充必要的业务注释，并在Prompt里显式给出表之间的关联关系。比如orders.customer_id = customers.id，模型看到这个就能准确生成Join查询。另外一个技巧是给模型几条常见的“问题-SQL”示例，作为Few-shot放在Prompt里，对SQL生成准确率的提升立竿见影。
6.2 上下文爆掉，Token费用飙升
刚开始做问数Agent的时候，我把数据库查询结果整个塞给模型，让它总结回答。结果一个查询返回5000行数据，这段结果转成文本就有几万Token，一次请求下来费用高得离谱，而且模型处理超长文本的延迟也让人无法接受。
后来我做了三层截断。第一层是对查询结果做行数限制，基础配置最多返回50行。第二层是对结果做字段裁剪，只保留和问题直接相关的列。第三层是对单行内容超长的文本字段做截断，比如一段商品描述只保留前200个字符。这三层截断之后，喂给模型的结果文本基本能控制在3000 Token以内。
6.3 并发一上来，数据库连接直接被打满
第一次压测的时候，10个并发请求同时进来，数据库连接数瞬间被打满，大量查询超时，连带Agent服务也跟着卡顿。排查发现两个问题：一个是连接池的pool_size配置太小，另一个是Agent执行SQL节点和模型调用都在异步事件循环里跑，但数据库查询没有走异步接口，导致线程阻塞。
修复方案是统一走SQLAlchemy的异步会话，并适当调大连接池。同时加了一个信号量来控制对数据库的整体并发，因为Agent场景下你可能同时有N个用户提问，但真正同时打到数据库的查询不应该超过连接池上限。
6.4 MCP工具偶发连接不上
MCP协议引入之后，新问题也随之而来：本地MCP Server偶发连接不上，Agent报工具调用失败。排查发现是MCP客户端默认的超时时间设置太短，而本地MCP Server在冷启动时需要加载数据库连接池配置，初始化耗时超过了客户端等待时间。
解决方案是在MCP服务端做懒加载和初始化预热。连接池在MCP Server启动时就先行创建，而不是等第一个工具调用到来时再创建。同时适当调大客户端的调用超时时间，并加了心跳检测，发现连接断开会自动拉起一个新的MCP客户端。
6.5 排查技巧：日志里的每一步都要能追溯
最后说一个通用排查技巧。在基础设施阶段就一定要把日志规范好。我的Agent服务里，每个节点进入和退出都打印结构化日志，包含请求ID、节点名、耗时、关键参数。SQL查询的执行日志单独记录，包括生成的SQL语句、执行耗时、返回行数。
排查线上问题的时候，只要拿到一个请求ID，就能在日志里把整个Agent的执行路径拉出来，看到底是模型生成SQL太慢，还是查询数据库超时，还是最后回答生成阶段出了问题。这套日志规范虽然花了我半天时间搭，但后面每次排查问题都省下了数倍的时间成本。
根据我个人的实际经验，基础设施搭建阶段多花的时间和心思，会在后续调优阶段十倍地赚回来。尤其是数据权限隔离和元数据管理这两块，表面看不显眼，但它们决定了问数项目到底只是“能跑的demo”还是“能见人的产品”。如果你正在做类似的AI Agent项目，我建议你在动手写业务逻辑之前，先老老实实把这篇里提到的连接池、Schema缓存、工具封装、用量统计这些地基打牢。地基稳了，上面怎么盖楼都不慌。
