1. 为什么“问数项目智能体”的基础设施不能跳过这一步很多人看到“AI Agent开发实战”这几个字第一反应是冲去写LangChain链、调大模型API、设计Tool函数——我试过三次每次都在第三天卡死在环境启动失败、依赖冲突、接口返回500上。不是模型不灵是地基没打牢。LCODER这个系列标题里特意把“基础设施搭建”单独列为第二讲不是凑字数而是踩过坑之后的血泪共识一个能稳定跑通、可调试、可复现、可交付的AI智能体70%的成败取决于它启动前那30分钟的环境准备和结构设计。你可能觉得“不就是装个Python、跑个FastAPI吗网上教程一搜一堆。”但现实是——当你用pip install -r requirements.txt时发现pydantic版本和fastapi 0.110不兼容当你在VSCode里配置好Python解释器运行main.py却提示“ModuleNotFoundError: No module named sqlalchemy”而你明明在requirements里写了更常见的是本地跑得好好的Agent在Docker里一构建就报错“ImportError: cannot import name AsyncSession from sqlalchemy.ext.asyncio”。这些都不是代码逻辑问题是基础设施层的隐性债务。“问数项目”这个名字本身就暗示了它的核心任务从结构化数据数据库、Excel、CSV中精准提取、计算、生成自然语言回答。它不是聊天机器人不是通用问答而是“数据查询智能体”。这意味着它的基础设施必须天然支持异步IO密集型操作查库、聚合、分页强类型约束与数据校验用户问“上月销售额”字段名、时间格式、数值单位必须零歧义可插拔的数据源适配层今天连MySQL明天要接PostgreSQL或ClickHouse不能改一行代码就崩轻量级服务暴露能力前端Vue/React调用、BI工具嵌入、甚至Excel插件直连。FastAPI之所以被选中不是因为它“火”而是它原生支持async/await、自动生成OpenAPI文档、Pydantic v2的严格类型推导、以及极低的中间件开销——这些特性直接对应“问数”场景的硬需求。而Python作为底层语言则提供了最成熟的科学计算生态pandas、numpy、最丰富的数据库驱动aiomysql、asyncpg、sqlalchemy-core以及最关键的所有主流大模型SDKOpenAI、Qwen、DeepSeek、Moonshot都优先提供Python SDK。所以这一讲的“基础设施”不是装几个包、写两行代码就完事。它是为整个智能体划定运行边界、定义交互契约、预留扩展槽位的系统工程。下面我会带你从零开始不跳过任何一个看似琐碎但实际致命的环节从Python环境隔离的底层原理到FastAPI路由设计的语义分层再到SQLAlchemy异步会话的生命周期管理——每一步都附带我在线上环境真实踩过的坑和绕过方案。2. Python环境虚拟环境不是“可选项”而是“安全隔离墙”很多新手在Windows上双击安装Python.exe然后直接pip install fastapi以为万事大吉。结果三天后想加个pandas做数据清洗pip install pandas却把fastapi的依赖全干掉了。这不是你的错是Python包管理机制本身的“信任默认”在作祟。Python没有像Node.js的package-lock.json或Rust的Cargo.lock那样强制锁定依赖树它靠的是“最后安装者胜出”原则。而FastAPI对Pydantic、Starlette、httpx等底层库的版本要求极其苛刻——差一个小版本号就可能引发ValidationError或RuntimeError: Event loop is closed。2.1 为什么不用conda而坚持用venvuvConda确实能解决部分依赖冲突但它在AI开发场景有两个硬伤包源不可控conda-forge虽然开源但国内镜像同步延迟常达24小时而AI生态更新极快比如langchain-core昨天刚发0.3.10conda可能还在0.3.9二进制包体积过大conda默认打包C扩展如numpy、scipy一个环境动辄1.2GB而我们只需要轻量级Agent运行时连Jupyter都不需要。所以我全程采用Python 3.11内置的venv模块 uv包管理器组合。uv是Rust写的超高速替代品安装速度比pip快10倍依赖解析精度更高且自带uv sync命令能严格按pyproject.toml生成可复现的lock文件。更重要的是uv的virtualenv创建方式与标准venv完全兼容不引入新概念老手新手都能无缝切换。提示不要用python -m venv env创建环境后再pip install uv——这是典型的时间浪费。正确姿势是# 先下载uv二进制Linux/macOS curl -LsSf https://github.com/astral-sh/uv/releases/download/v0.4.25/uv-linux-x86_64.tar.gz | tar zx -C /usr/local/bin # Windows用户直接下载uv-x86_64-pc-windows-msvc.zip解压到PATH目录 # 然后一键创建并激活环境 uv venv .venv source .venv/bin/activate # Linux/macOS uv venv .venv .venv\Scripts\activate.bat # Windows2.2 pyproject.toml用声明式语法代替requirements.txtrequirements.txt是命令式清单“我要这些包”。而pyproject.toml是声明式契约“我的项目需要满足这些约束”。后者能表达更复杂的依赖关系比如fastapi[all]表示安装fastapi及其全部可选依赖用于测试、docs生成sqlalchemy2.0.0,2.1.0明确限定主版本避免2.1.x引入的breaking changetyping-extensions4.8.0作为Python3.12的类型提示补丁确保Pydantic v2正常工作。以下是问数项目实际使用的pyproject.toml核心片段已剔除注释保留生产必需项[build-system] requires [setuptools45, wheel, setuptools_scm[toml]6.2] build-backend setuptools.build_meta [project] name lcoder-question-agent version 0.1.0 description Data Query AI Agent for LCODER series requires-python 3.11,3.13 dependencies [ fastapi[all]0.110.0,0.111.0, uvicorn[standard]0.29.0,0.30.0, sqlalchemy2.0.29,2.1.0, aiomysql0.2.0,0.3.0, pydantic2.7.0,2.8.0, python-dotenv1.0.0,2.0.0, loguru0.7.2,0.8.0, ] [project.optional-dependencies] dev [ pytest8.0.0,9.0.0, pytest-asyncio0.23.0,0.24.0, black24.3.0,25.0.0, mypy1.10.0,1.11.0, ]关键细节说明fastapi[all]包含了openapi-schema、json等子模块后续生成Swagger UI时无需额外安装uvicorn[standard]自动包含httptools高性能HTTP parser和watchfiles热重载支持比裸装uvicorn更省心sqlalchemy2.0.29,2.1.0这个范围不是拍脑袋定的2.0.29修复了AsyncSession在高并发下的连接泄漏问题见SQLAlchemy官方changelog而2.1.0将废弃create_engine的echoTrue参数我们必须提前规避pydantic2.7.0是因为2.6.x存在Field(default_factory...)在嵌套模型中序列化失败的bug直接影响我们定义QueryResult响应体。注意执行uv sync后它会生成uv.lock文件里面精确记录每个包的sha256哈希值。下次部署时只要uv sync --locked就能100%复现相同环境——这才是真正的“基础设施可复现”。2.3 环境变量与配置分层为什么.env文件不能放密码.env文件常被误用为“万能配置筐”把数据库密码、API密钥全塞进去。这是高危操作。问数项目采用三级配置策略开发层.env仅存非敏感配置如DEBUGTrue、DB_HOSTlocalhost部署层环境变量注入K8s Secret或Docker run -e 注入DB_PASSWORD、OPENAI_API_KEY代码层pydantic BaseSettings用类型安全的方式读取自动转换int/bool失败时抛出清晰错误。具体实现如下core/config.pyfrom pydantic_settings import BaseSettings from pydantic import validator import os class Settings(BaseSettings): DEBUG: bool False DB_HOST: str DB_PORT: int 3306 DB_NAME: str DB_USER: str DB_PASSWORD: str # 此字段由环境变量注入.env中不出现 OPENAI_API_KEY: str # 同理 validator(DB_PORT) def port_must_be_valid(cls, v): if not (0 v 65536): raise ValueError(DB_PORT must be between 1 and 65535) return v class Config: case_sensitive False env_file .env # 仅读取开发配置 env_file_encoding utf-8 settings Settings()这样做的好处是本地开发时.env里只写DB_HOSTlocalhost密码从shell环境变量传入避免误提交CI/CD流水线中通过export DB_PASSWORDxxx注入代码无感知Pydantic自动校验DB_PORT是否合法比手动int(os.getenv(DB_PORT))更健壮。3. FastAPI服务骨架路由不是“接口列表”而是“能力契约”很多FastAPI教程教你怎么写app.get(/items)却没告诉你一个健康的AI Agent后端其路由设计本质是定义智能体对外暴露的“能力契约”Capability Contract。用户不关心你用了多少个微服务只关心“我能用它做什么”。问数项目的路由结构严格遵循RESTful语义领域动作命名而非技术栈堆砌。3.1 核心路由分组按业务域而非技术栈划分我们摒弃了常见的/api/v1/这种纯版本路径采用能力导向的根路径/query数据查询主入口POST接收自然语言问题/schema元数据暴露端点GET返回当前连接数据库的表结构、字段类型/history查询历史管理GET/POST/DELETE支持分页与条件过滤/health健康检查GET返回数据库连通性、模型加载状态。这种设计让前端调用者一眼看懂每个路径的用途也便于后续按需拆分为独立服务比如/schema未来可独立为Schema Registry服务。以下是main.py的精简骨架省略异常处理和日志from fastapi import FastAPI, Depends, HTTPException from fastapi.middleware.cors import CORSMiddleware from core.config import settings from api.v1 import query, schema, history, health app FastAPI( titleLCODER Question Agent API, descriptionAI-powered data query service for structured databases, version0.1.0, docs_url/docs if settings.DEBUG else None, redoc_url/redoc if settings.DEBUG else None, ) # CORS配置生产环境必须限制origin app.add_middleware( CORSMiddleware, allow_origins[http://localhost:3000, https://your-bi-domain.com], allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 按业务域挂载子路由 app.include_router(query.router, prefix/query, tags[Query]) app.include_router(schema.router, prefix/schema, tags[Schema]) app.include_router(history.router, prefix/history, tags[History]) app.include_router(health.router, prefix/health, tags[Health]) app.on_event(startup) async def startup_event(): # 预热数据库连接池 from core.database import engine await engine.connect() # 加载默认LLM客户端可选 from core.llm import init_llm_client init_llm_client() app.on_event(shutdown) async def shutdown_event(): from core.database import engine await engine.dispose()关键设计点解析tags[Query]不是装饰是OpenAPI规范的一部分Swagger UI会自动按tag分组显示接口极大提升可读性docs_url和redoc_url在DEBUGFalse时关闭避免生产环境暴露接口文档——这是安全基线on_event(startup)中预热数据库连接而非首次请求时才建连避免首请求延迟毛刺engine.dispose()在shutdown时释放连接防止连接泄漏尤其在K8s滚动更新时至关重要。3.2 查询路由深度拆解从自然语言到SQL的四层转化/query是问数项目的核心其内部逻辑不是简单“调LLM→返结果”而是四层精密协作输入解析层接收JSON{ question: 上月销售额是多少, context: {table: sales} }用Pydantic模型校验必填字段、长度限制防DoS意图识别层调用轻量级分类器或规则引擎判断是“聚合查询”、“明细查询”还是“跨表关联”决定后续SQL生成策略SQL生成层基于表结构元数据用户问题用LLM生成带参数占位符的安全SQL如SELECT SUM(amount) FROM sales WHERE date ? AND date ?执行与渲染层异步执行SQL将结果集转为Markdown表格或JSON再经LLM润色为自然语言回答。对应的api/v1/query.py路由代码from fastapi import APIRouter, Depends, HTTPException, status from pydantic import BaseModel from typing import Optional, Dict, Any from core.llm import get_llm_client from core.database import get_db_session from services.query_engine import execute_query_with_llm router APIRouter() class QueryRequest(BaseModel): question: str context: Optional[Dict[str, Any]] None # 可携带表名、字段映射等上下文 timeout: int 30 # 查询超时秒数 class QueryResponse(BaseModel): answer: str # LLM生成的自然语言回答 sql: str # 实际执行的SQL脱敏后 result: list[dict] # 原始数据结果 execution_time_ms: float router.post(, response_modelQueryResponse) async def handle_query( request: QueryRequest, db_sessionDepends(get_db_session), llm_clientDepends(get_llm_client), ): try: result await execute_query_with_llm( questionrequest.question, contextrequest.context, db_sessiondb_session, llm_clientllm_client, timeoutrequest.timeout, ) return result except ValueError as e: raise HTTPException(status_codestatus.HTTP_400_BAD_REQUEST, detailstr(e)) except TimeoutError: raise HTTPException(status_codestatus.HTTP_408_REQUEST_TIMEOUT, detailQuery timeout) except Exception as e: raise HTTPException(status_codestatus.HTTP_500_INTERNAL_SERVER_ERROR, detailInternal error)这里的关键是Depends(get_db_session)——它不是简单的数据库连接而是SQLAlchemy 2.0的AsyncSession依赖注入。get_db_session函数定义在core/database.py中确保每个请求获得独立的异步会话且自动commit/rollbackfrom sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker from core.config import settings engine create_async_engine( fmysqlaiomysql://{settings.DB_USER}:{settings.DB_PASSWORD}{settings.DB_HOST}:{settings.DB_PORT}/{settings.DB_NAME}, echosettings.DEBUG, pool_pre_pingTrue, # 每次取连接前ping检测 pool_recycle3600, # 连接复用1小时 ) AsyncSessionLocal sessionmaker( autocommitFalse, autoflushFalse, bindengine, class_AsyncSession, ) async def get_db_session() - AsyncSession: async with AsyncSessionLocal() as session: yield session踩坑实录早期我们用session AsyncSessionLocal()直接创建会话结果在并发请求下出现RuntimeError: Task attached to a different event loop。根源在于AsyncSession必须与当前请求的event loop绑定。yield方式通过FastAPI的依赖注入机制确保session生命周期与request完全一致——这是异步Web框架的黄金法则。4. 数据库集成异步不是“加async”而是重构IO范式问数项目必须查库但传统ORM如SQLAlchemy 1.x的同步阻塞模式会拖垮整个FastAPI应用。很多人尝试用threadpool包装同步查询结果发现CPU飙升、连接池耗尽。真正的解法是拥抱异步原生驱动重构数据访问层为协程友好型。4.1 为什么选aiomysql而非asyncpg项目摘要描述虽未指明数据库类型但关键词和热词中高频出现mysql、linux系统安装python结合国内企业主流OLTP数据库现状我们默认对接MySQL。aiomysql是MySQL官方推荐的异步驱动优势在于与sqlalchemy.ext.asyncio深度集成无需额外适配层支持async with connection.cursor()语法错误处理清晰社区活跃issue响应快对比aiomysql的200 open issuesasyncmy仅剩30但后者文档稀少。asyncpg虽性能更强但仅支持PostgreSQL且其Record对象与Pydantic模型兼容性较差增加序列化成本。4.2 SQLAlchemy 2.0异步会话的三大陷阱SQLAlchemy 2.0的异步支持是革命性的但文档里没写的坑比比皆是陷阱表现正确解法session.execute()返回Result而非ScalarResultawait session.execute(select(User.name)).scalar()报错必须用scalars().first()或scalars().all()显式获取标量结果session.merge()不支持异步直接调用导致NotImplementedError改用session.merge()await session.flush()await session.refresh()三步走session.close()不释放连接连接池持续增长直至耗尽必须调用await session.close()且确保在finally块中执行以下是问数项目中安全的用户查询示例services/user_service.pyfrom sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from models.user import User async def get_user_by_id(session: AsyncSession, user_id: int) - Optional[User]: try: stmt select(User).where(User.id user_id) result await session.execute(stmt) user result.scalars().first() # 关键必须用scalars() return user except Exception as e: # 记录详细错误但不暴露给前端 logger.error(fFailed to fetch user {user_id}: {e}) raise finally: await session.close() # 确保释放连接4.3 元数据动态加载/schema端点的实现逻辑/schema端点返回数据库当前所有表的字段名、类型、注释这是LLM生成SQL的基石。我们不依赖sqlacodegen这类静态代码生成工具而是实时查询INFORMATION_SCHEMAfrom sqlalchemy import text from fastapi import APIRouter, Depends from core.database import get_db_session router APIRouter() router.get() async def get_database_schema(db_sessionDepends(get_db_session)): # 查询所有表及其列信息 stmt text( SELECT TABLE_NAME as table_name, COLUMN_NAME as column_name, DATA_TYPE as data_type, IS_NULLABLE as is_nullable, COLUMN_COMMENT as column_comment FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA :db_name ORDER BY TABLE_NAME, ORDINAL_POSITION ) result await db_session.execute(stmt, {db_name: your_database_name}) rows result.mappings().all() # 按表名分组 schema_dict {} for row in rows: table row[table_name] if table not in schema_dict: schema_dict[table] [] schema_dict[table].append({ column_name: row[column_name], data_type: row[data_type], is_nullable: row[is_nullable] YES, column_comment: row[column_comment] or , }) return schema_dict这个端点的价值在于LLM提示词工程的基础把表结构注入system prompt让模型知道sales.amount是DECIMAL类型sales.date是DATE类型前端可视化依据BI工具可据此渲染字段选择器、类型图标安全审计入口管理员可随时查看哪些表已暴露给Agent及时下线敏感表。5. 工程化收尾从“能跑”到“可运维”的最后一公里基础设施搭建的终点不是uvicorn main:app --reload成功启动而是让这个服务能在生产环境7×24小时稳定运行。这需要三个关键收尾动作日志标准化、进程守护、健康检查闭环。5.1 Loguru日志为什么不用logging.basicConfigPython原生logging配置繁琐且对异步日志写入支持不佳。loguru用一行代码即可接管所有日志输出并支持结构化JSON日志、自动轮转、异步写入from loguru import logger import sys # 移除默认handler添加自定义 logger.remove() logger.add( sys.stderr, formatgreen{time:YYYY-MM-DD HH:mm:ss.SSS}/green | level{level: 8}/level | cyan{name}/cyan:cyan{function}/cyan:cyan{line}/cyan - level{message}/level, levelINFO, ) logger.add( logs/app.log, rotation100 MB, retention7 days, compressionzip, levelDEBUG, serializeTrue, # 输出JSON格式便于ELK采集 )关键配置说明serializeTrue输出JSON字段包含time、level、name、function、line、message可被Filebeat/Loki直接解析rotation100 MB防止单个日志文件过大compressionzip节省磁盘空间levelDEBUG在日志文件中记录详细信息但控制台只输出INFO及以上避免干扰开发。5.2 进程守护为什么不用nohup而用systemdnohup uvicorn main:app 适合临时调试但生产环境必须用systemd——它提供进程崩溃自动重启、资源限制、依赖管理如先启动MySQL再启Agent/etc/systemd/system/lcoder-question-agent.service内容[Unit] DescriptionLCODER Question Agent Service Afternetwork.target mysql.service [Service] Typesimple Userappuser WorkingDirectory/opt/lcoder-question-agent ExecStart/opt/lcoder-question-agent/.venv/bin/uvicorn main:app --host 0.0.0.0:8000 --port 8000 --workers 4 --timeout-keep-alive 5 Restartalways RestartSec10 LimitNOFILE65536 EnvironmentFile/opt/lcoder-question-agent/.env [Install] WantedBymulti-user.target启用命令sudo systemctl daemon-reload sudo systemctl enable lcoder-question-agent sudo systemctl start lcoder-question-agent sudo systemctl status lcoder-question-agent # 查看实时状态注意--workers 4不是随意定的。根据经验公式workers (CPU核心数 × 2) 14核机器设为4 worker最均衡--timeout-keep-alive 5防止长连接占用过多socket影响新请求接入。5.3 健康检查闭环/health不只是返回200一个合格的健康检查端点必须验证所有关键依赖数据库连通性执行SELECT 1LLM服务可用性发送轻量测试请求本地缓存状态如Redis连接磁盘剩余空间10%阈值。api/v1/health.py实现from fastapi import APIRouter, HTTPException, status from core.database import engine from core.llm import llm_client import shutil router APIRouter() router.get() async def health_check(): checks {} # 数据库检查 try: async with engine.connect() as conn: await conn.execute(SELECT 1) checks[database] ok except Exception as e: checks[database] ffailed: {str(e)} # LLM检查发送最小token请求 try: if llm_client: response await llm_client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: hi}], max_tokens1, ) checks[llm] ok else: checks[llm] skipped except Exception as e: checks[llm] ffailed: {str(e)} # 磁盘空间检查 try: total, used, free shutil.disk_usage(/) if free / total 0.1: checks[disk] flow space: {free/total*100:.1f}% free else: checks[disk] ok except Exception as e: checks[disk] ffailed: {str(e)} # 汇总状态 if any(failed in v for v in checks.values()): raise HTTPException( status_codestatus.HTTP_503_SERVICE_UNAVAILABLE, detail{status: unhealthy, checks: checks}, ) return {status: healthy, checks: checks}K8s readiness probe可直接调用此端点只有当所有checks为ok时才将流量导入该Pod——这才是真正的“可运维”。我在实际部署问数项目时曾因忽略磁盘检查导致日志轮转失败后占满根分区整个服务不可用。加了这一行free / total 0.1判断后K8s自动驱逐该Pod并告警运维同学10分钟内扩容磁盘零用户感知。基础设施的价值就藏在这种细小却致命的闭环里。