《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协议在基础设施中的位置MCPModel 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 langgraph0.2.0 langchain-core0.3.0 langchain-openai0.2.0 pip install mcp1.0.0 sqlalchemy2.0.0 asyncpg0.29.0 pip install fastapi0.115.0 uvicorn[standard]0.30.0 sse-starlette2.0.0 pip install pydantic-settings2.0.0 python-dotenv1.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_KEYsk-xxx OPENAI_BASE_URLhttps://api.openai.com/v1 OPENAI_MODELgpt-4o ANTHROPIC_API_KEYsk-ant-xxx ANTHROPIC_MODELclaude-sonnet-4-20250514 DATABASE_URLpostgresqlasyncpg://user:passlocalhost:5432/lcoder DATABASE_MAX_CONNECTIONS10 DATABASE_TIMEOUT5在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_size10, max_overflow5, pool_timeout10, pool_recycle1800, echoFalse, ) SessionLocal async_sessionmaker(engine, expire_on_commitFalse)这里重点说三个参数。pool_recycle设为1800秒是为了避免数据库服务端主动断掉空闲连接尤其是有些云数据库默认空闲超时很短你不设置回收时间就会出现“执行几次查询后突然报连接已断开”的诡异问题。pool_timeout设为10秒是为了避免高并发时请求排队太久如果10秒拿不到连接就直接报错而不是无限阻塞住整个Agent。max_overflow设小一点给突发流量留一点缓冲但也不要太大否则连接数打满了数据库先挂了。3.2 Schema元数据管理问数项目里你不可能把整库几十张表的结构一股脑塞给LLMToken会爆炸模型也会被无关字段干扰。所以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 ServerAgent下次运行时就能自动感知到新工具的列表不需要重新部署代码。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变成可用的API5.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, versionv2): 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_idValue为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缓存、工具封装、用量统计这些地基打牢。地基稳了上面怎么盖楼都不慌。