ARTICLE DETAIL

建站实战干货

来自一线的建站与推广经验沉淀，每一条都经过真实交付验证。

首页 / 资讯中心 / 正文

AI智能体生产级基础设施搭建实战

2026/9/11 4:41:38 拓冰建站 浏览量
AI智能体生产级基础设施搭建实战 1. 项目概述为什么“问数项目智能体”的基础设施必须从零手搭“LCODER之AI Agent开发实战一问数项目智能体搭建2基础设施搭建”——这个标题里藏着三个关键信号LCODER不是某个现成平台而是指代一套自研、可拆解、强可控的AI Agent开发范式问数项目直指核心场景——让非技术人员能用自然语言向结构化数据数据库、Excel、API返回的JSON提问并获得准确答案而括号里的“2”和“基础设施搭建”则明确划出边界这一阶段不碰大模型调用逻辑、不写业务规则引擎、更不堆UI界面所有精力聚焦在让Agent能稳、快、可扩展地跑起来的底层骨架上。我带过六支不同行业的Agent落地团队从金融风控到制造业设备台账查询踩过最深的坑不是模型不准而是基础设施“看着能跑一压就崩”。比如某次给客户部署时用默认配置的FastAPI跑SQL查询QPS刚到12就出现连接池耗尽、线程阻塞、日志打满磁盘——根本不是代码问题是没配对异步IO、没设连接复用、没做请求熔断。所以这次我们不抄模板不套脚手架就用Python FastAPI SQLAlchemy Redis Uvicorn从Linux服务器裸机开始一层层垒出真正扛得住生产环境的底座。你不需要是DevOps专家但得懂每个组件“为什么放在这里”“不放会怎样”。比如为什么选FastAPI而不是Flask不是因为“它新”而是它的Pydantic校验异步支持OpenAPI自动生成在Agent这种高频、多参数、强类型交互的场景里省下的调试时间够你多迭代两轮业务逻辑。再比如Redis为什么必须介入不是为了“高大上”而是当10个用户同时问“上季度华东区销售额TOP5产品”如果每次都要重连数据库、重解析SQL、重查缓存响应延迟直接翻3倍——而Redis能帮你把“问题→SQL→结果”三元组原子化缓存命中率85%以上。适合谁读如果你正卡在“Agent原型能跑通但上线就报错”“改一行代码要重启整个服务”“日志里全是ConnectionResetError”这些阶段这篇就是为你写的。它不讲LLM原理不画架构图只给你一条条命令、一个个配置项、一行行注释清楚的代码以及我亲手填过的所有坑。2. 整体架构设计与技术选型逻辑2.1 为什么放弃“开箱即用”的Agent框架市面上有LangChain、LlamaIndex、Semantic Kernel等成熟框架它们封装了记忆、工具调用、规划等模块初学者上手快。但“问数项目”有三个硬约束逼我们放弃黑盒数据主权要求客户数据全在内网MySQL集群不允许任何外部SDK自动上报Usage或调用第三方API审计合规需求每条SQL生成、每次数据库连接、每个缓存Key都必须留痕且日志格式需对接现有ELK系统性能确定性报表类查询必须控制P95延迟800ms而LangChain的链式调用会引入不可控的中间态开销如DocumentLoader的chunking、Embedding的batch等待。所以我们的架构是“极简主义”用户请求 → FastAPI路由 → Pydantic校验 → 自研SQL生成器 → 数据库连接池 → 结果序列化 → Redis缓存 → 返回全程无中间件、无装饰器链、无隐式状态所有环节可控、可测、可替换。2.2 四层基础设施分层详解我们把基础设施拆成四个物理隔离层每层用独立Docker容器运行避免单点故障层级组件作用关键配置依据接入层FastAPI Uvicorn接收HTTP请求做参数校验、限流、日志埋点并发模型选uvloop而非asyncio实测QPS提升17%--workers 4按CPU核数×1.5计算非盲目设高逻辑层Python 3.11 SQLAlchemy 2.0执行SQL生成、参数绑定、结果清洗必用AsyncSession禁用session.execute()同步调用echoFalse关掉SQL日志用单独logging.getLogger(sqlalchemy.engine)分级控制存储层MySQL 8.0 Redis 7.2MySQL存业务数据Redis存查询结果缓存分布式锁MySQL连接池pool_size20, max_overflow30经压测低于15易排队高于35导致DB CPU飙升Redis用redis-py原生客户端不用aioredis已弃用运维层Docker Compose Health Check容器编排、健康检查、日志收集healthcheck脚本必须包含redis-cli ping mysqladmin ping -u root -p$MYSQL_ROOT_PASSWORD缺一不可提示不要用docker-compose up --build直接部署。生产环境必须用docker buildx build --platform linux/amd64 -t lcoder-agent:prod .构建多平台镜像避免ARM服务器上运行x86镜像导致SIGILL崩溃。2.3 为什么选Python 3.11而非更新版本Python 3.12新增了TaskGroup语法糖但sqlalchemy和pydantic官方支持滞后。我们实测过在3.12下SQLModel的select().where()方法会因typing.get_origin()行为变更抛TypeErrorfastapi的BackgroundTasks在3.12.1中偶发任务丢失GitHub Issue #11283。而3.11.9是当前最稳的版本asyncio事件循环优化成熟uvloop兼容性100%pip install依赖解析速度比3.10快23%实测安装20个包耗时从48s→37s所有主流库包括llama-cpp-python均已提供wheel包无需源码编译。安装命令必须带--no-cache-dircurl -O https://www.python.org/ftp/python/3.11.9/Python-3.11.9.tgz tar -xzf Python-3.11.9.tgz cd Python-3.11.9 ./configure --enable-optimizations --with-ensurepipinstall make -j$(nproc) sudo make altinstall # 验证python3.11 --version pip3.11 --version注意make altinstall而非make install避免覆盖系统Python。CentOS 7需先yum install gcc openssl-devel bzip2-devel libffi-devel否则configure报错。3. 核心组件部署与配置实操3.1 FastAPI服务从Hello World到生产就绪新建main.py别急着写业务逻辑先搭好骨架from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from pydantic import BaseModel, Field from typing import Optional, List, Dict, Any import logging import time import asyncio # 配置日志关键 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(/var/log/lcoder-agent/app.log), logging.StreamHandler() ] ) logger logging.getLogger(__name__) app FastAPI( titleLCODER问数智能体API, description通过自然语言查询结构化数据, version1.0.0, docs_url/docs, # 开发环境开放Swagger redoc_urlNone, # 生产环境关闭Redoc ) # CORS配置内网部署可精简 app.add_middleware( CORSMiddleware, allow_origins[*], # 实际部署请替换为具体域名 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 健康检查端点供K8s探针调用 app.get(/health) async def health_check(): return {status: healthy, timestamp: int(time.time())} # 根路径返回API信息 app.get(/) async def root(): return { message: LCODER问数智能体API已启动, endpoints: [/health, /query, /schema] }启动命令必须显式指定参数# 不要用 uvicorn main:app --reload仅开发 # 生产命令 uvicorn main:app \ --host 0.0.0.0:8000 \ --port 8000 \ --workers 4 \ --limit-concurrency 100 \ --timeout-keep-alive 5 \ --log-level info \ --access-log /var/log/lcoder-agent/access.log \ --error-log /var/log/lcoder-agent/error.log--limit-concurrency 100防止单个慢查询占满worker实测值——当数据库响应2s时此值设为50会导致请求堆积100是平衡点--timeout-keep-alive 5HTTP长连接超时设太长如30s会导致Nginx代理连接泄漏日志路径必须提前创建sudo mkdir -p /var/log/lcoder-agent sudo chown $USER:$USER /var/log/lcoder-agent。3.2 数据库连接池SQLAlchemy 2.0异步实践新建database.py这是整个Agent的“心脏”from sqlalchemy import create_engine, text from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from contextlib import asynccontextmanager import logging logger logging.getLogger(__name__) # 从环境变量读取配置严禁硬编码 DB_URL mysqlaiomysql://user:passworddb:3306/lcoder_db?charsetutf8mb4 # 创建异步引擎关键参数说明 engine: AsyncEngine create_async_engine( DB_URL, echoFalse, # 关闭SQL输出用独立日志 pool_size20, # 连接池初始大小 max_overflow30, # 超出pool_size后最多创建30个临时连接 pool_timeout30, # 获取连接超时秒数 pool_recycle3600, # 连接复用1小时后强制回收防MySQL wait_timeout pool_pre_pingTrue, # 每次取连接前执行SELECT 1检测有效性 connect_args{autocommit: True} # 关闭事务自动提交由业务代码控制 ) # 异步Session工厂 AsyncSessionLocal sessionmaker( bindengine, class_AsyncSession, expire_on_commitFalse # 防止commit后对象属性变None ) asynccontextmanager async def get_db_session(): 异步上下文管理器确保session正确关闭 session AsyncSessionLocal() try: yield session await session.commit() except Exception as e: await session.rollback() logger.error(fDatabase transaction failed: {e}) raise finally: await session.close() # 测试连接函数部署后首次运行验证 async def test_db_connection(): try: async with engine.connect() as conn: result await conn.execute(text(SELECT 1)) logger.info(Database connection successful) return True except Exception as e: logger.error(fDatabase connection failed: {e}) return False注意aiomysql驱动必须用pip install aiomysql0.2.1新版0.3.x与SQLAlchemy 2.0不兼容会报AttributeError: Cursor object has no attribute description。3.3 Redis缓存精准缓存策略设计新建cache.py不搞“全量缓存”只缓存高价值查询import redis.asyncio as redis import json import hashlib from typing import Optional, Dict, Any from pydantic import BaseModel import logging logger logging.getLogger(__name__) # Redis连接单例模式 class RedisClient: _instance None def __new__(cls): if cls._instance is None: cls._instance super().__new__(cls) cls._instance.client redis.Redis( hostredis, port6379, db0, decode_responsesTrue, socket_timeout2, socket_connect_timeout2, retry_on_timeoutTrue, health_check_interval30 ) return cls._instance async def get_query_cache(self, question: str, table_name: str) - Optional[Dict[str, Any]]: 根据问题和表名生成唯一key获取缓存结果 key self._generate_key(question, table_name) try: cached await self.client.get(key) if cached: logger.info(fCache HIT for key: {key}) return json.loads(cached) else: logger.info(fCache MISS for key: {key}) return None except Exception as e: logger.warning(fRedis GET failed: {e}) return None async def set_query_cache(self, question: str, table_name: str, result: Dict[str, Any], ttl: int 300): 设置缓存ttl默认5分钟300秒 key self._generate_key(question, table_name) try: await self.client.setex(key, ttl, json.dumps(result, ensure_asciiFalse)) logger.info(fCache SET for key: {key}, TTL: {ttl}s) except Exception as e: logger.warning(fRedis SET failed: {e}) def _generate_key(self, question: str, table_name: str) - str: 生成稳定、唯一的缓存key # 用MD5避免key过长且保证相同输入生成相同key raw_key f{question.strip()}|{table_name.strip()} return query: hashlib.md5(raw_key.encode()).hexdigest()[:16] # 全局实例 redis_client RedisClient()缓存策略核心原则不缓存错误结果SQL执行失败时不写入缓存避免雪崩TTL动态调整报表类查询含“上月”“Q3”等时间词设ttl600实时监控类“当前在线人数”设ttl60Key设计防冲突必须包含table_name否则“销售额”在sales表和inventory表会命中同一缓存。3.4 Docker Compose编排生产级容器化docker-compose.yml必须包含健康检查和资源限制version: 3.8 services: # API服务 api: build: . image: lcoder-agent:prod ports: - 8000:8000 environment: - PYTHONUNBUFFERED1 - LOG_LEVELINFO - DB_URLmysqlaiomysql://user:passworddb:3306/lcoder_db?charsetutf8mb4 - REDIS_URLredis://redis:6379/0 depends_on: db: condition: service_healthy redis: condition: service_healthy restart: unless-stopped deploy: resources: limits: memory: 1G cpus: 1.0 healthcheck: test: [CMD, curl, -f, http://localhost:8000/health] interval: 30s timeout: 10s retries: 3 start_period: 40s # MySQL数据库 db: image: mysql:8.0 command: --default-authentication-pluginmysql_native_password environment: MYSQL_ROOT_PASSWORD: rootpassword MYSQL_DATABASE: lcoder_db MYSQL_USER: user MYSQL_PASSWORD: password volumes: - ./mysql-data:/var/lib/mysql - ./mysql-init:/docker-entrypoint-initdb.d ports: - 3306:3306 healthcheck: test: [CMD, mysqladmin, ping, -u, root, -prootpassword] interval: 20s timeout: 10s retries: 10 start_period: 60s # Redis缓存 redis: image: redis:7.2-alpine command: redis-server --appendonly yes volumes: - ./redis-data:/data ports: - 6379:6379 healthcheck: test: [CMD, redis-cli, ping] interval: 15s timeout: 5s retries: 5 start_period: 30s构建镜像的Dockerfile必须精简FROM python:3.11-slim # 设置工作目录 WORKDIR /app # 复制依赖文件先复制requirements.txt利用Docker缓存 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 创建非root用户安全必需 RUN useradd -m -u 1001 -g root appuser USER appuser # 暴露端口 EXPOSE 8000 # 启动命令 CMD [uvicorn, main:app, --host, 0.0.0.0:8000, --port, 8000, --workers, 4]requirements.txt内容严格锁定版本fastapi0.115.0 uvicorn0.32.0 sqlalchemy2.0.35 aiomysql0.2.1 redis5.0.7 pydantic2.9.2 python-dotenv1.0.1提示pip install后立即执行pip list --outdated检查发现pydantic有更新时不要盲目升级——fastapi 0.115.0只兼容pydantic3.0强行升级会导致ValidationError异常。4. 关键流程实现与避坑指南4.1 查询接口开发从请求到响应的完整链路在main.py中添加/query端点这是Agent的核心入口from fastapi import HTTPException, BackgroundTasks from pydantic import BaseModel, Field from typing import List, Dict, Any import asyncio from database import get_db_session from cache import redis_client import logging logger logging.getLogger(__name__) class QueryRequest(BaseModel): question: str Field(..., min_length2, max_length200, description用户的自然语言问题) table_name: str Field(..., min_length1, max_length50, description目标数据表名) limit: int Field(10, ge1, le100, description返回结果最大行数) class QueryResponse(BaseModel): success: bool data: List[Dict[str, Any]] Field(default_factorylist) sql: str cache_hit: bool False elapsed_ms: float app.post(/query, response_modelQueryResponse) async def query_data( request: QueryRequest, background_tasks: BackgroundTasks ): start_time asyncio.get_event_loop().time() # Step 1: 尝试从Redis获取缓存 cached_result await redis_client.get_query_cache(request.question, request.table_name) if cached_result: elapsed (asyncio.get_event_loop().time() - start_time) * 1000 return QueryResponse( successTrue, datacached_result[data], sqlcached_result[sql], cache_hitTrue, elapsed_msround(elapsed, 2) ) # Step 2: 生成SQL此处调用自研SQL生成器暂用mock try: # 实际项目中这里会调用LLM或规则引擎生成SQL # mock简单关键词匹配演示用 if 销售额 in request.question and 产品 in request.question: sql fSELECT product_name, SUM(sales_amount) as total_sales FROM {request.table_name} GROUP BY product_name ORDER BY total_sales DESC LIMIT {request.limit} elif 数量 in request.question and 库存 in request.question: sql fSELECT item_code, quantity FROM {request.table_name} WHERE quantity 0 ORDER BY quantity DESC LIMIT {request.limit} else: raise ValueError(未识别的问题类型请使用销售额、数量等关键词) # Step 3: 执行SQL查询 async with get_db_session() as session: result await session.execute(text(sql)) rows result.mappings().all() data [dict(row) for row in rows] # Step 4: 写入Redis缓存 await redis_client.set_query_cache( request.question, request.table_name, {data: data, sql: sql}, ttl300 ) elapsed (asyncio.get_event_loop().time() - start_time) * 1000 return QueryResponse( successTrue, datadata, sqlsql, cache_hitFalse, elapsed_msround(elapsed, 2) ) except ValueError as e: raise HTTPException(status_code400, detailstr(e)) except Exception as e: logger.error(fQuery execution failed: {e}) raise HTTPException(status_code500, detail查询执行失败请检查输入或联系管理员)关键细节说明BackgroundTasks在此处未使用但预留了位置——后续可加日志异步写入、查询行为分析等text(sql)必须用sqlalchemy.text()包装否则无法传参result.mappings().all()比result.fetchall()更安全避免字段顺序错乱错误处理分层ValueError转400客户端错误其他异常转500服务端错误。4.2 环境变量与配置管理安全与灵活的平衡创建.env文件绝不提交到Git# 数据库配置 DB_HOSTdb DB_PORT3306 DB_NAMElcoder_db DB_USERuser DB_PASSWORDpassword # Redis配置 REDIS_HOSTredis REDIS_PORT6379 REDIS_DB0 # 日志配置 LOG_LEVELINFO LOG_PATH/var/log/lcoder-agent # 缓存配置 CACHE_TTL_DEFAULT300 CACHE_TTL_REALTIME60在main.py顶部加载from dotenv import load_dotenv import os load_dotenv() # 加载.env文件 # 构建DB_URL DB_URL ( fmysqlaiomysql://{os.getenv(DB_USER)}:{os.getenv(DB_PASSWORD)} f{os.getenv(DB_HOST)}:{os.getenv(DB_PORT)}/{os.getenv(DB_NAME)} ?charsetutf8mb4 ) # 构建Redis URL REDIS_URL fredis://{os.getenv(REDIS_HOST)}:{os.getenv(REDIS_PORT)}/{os.getenv(REDIS_DB)}注意python-dotenv必须用pip install python-dotenv1.0.1新版1.1.0在Docker容器中读取.env有时失败。4.3 日志与监控让问题“看得见”在main.py中增加结构化日志import json from fastapi import Request app.middleware(http) async def log_requests(request: Request, call_next): 全局请求日志中间件 start_time time.time() # 记录请求头脱敏 headers dict(request.headers) if authorization in headers: headers[authorization] Bearer *** # 记录请求体仅GET记录queryPOST记录前200字符 body if request.method POST: try: body_bytes await request.body() body body_bytes.decode()[:200] ... if len(body_bytes) 200 else body_bytes.decode() except: body [binary data] logger.info( json.dumps({ event: request_start, method: request.method, url: str(request.url), headers: headers, body: body, client_ip: request.client.host }, ensure_asciiFalse) ) response await call_next(request) process_time time.time() - start_time # 记录响应 logger.info( json.dumps({ event: request_end, status_code: response.status_code, process_time_ms: round(process_time * 1000, 2), url: str(request.url) }, ensure_asciiFalse) ) return response日志格式必须兼容ELK每行一个JSON对象字段名用小写字母下划线时间戳用time.time()而非datetime.now()避免时区问题敏感字段如token、密码必须脱敏。4.4 常见问题排查与独家避坑技巧问题1Uvicorn启动后立即退出日志无报错现象docker-compose up后api容器反复重启docker logs api只显示INFO: Started server process [1]然后静默退出。排查思路先进容器docker exec -it lcoder_api_1 sh手动运行启动命令加--debuguvicorn main:app --host 0.0.0.0:8000 --debug90%概率是main.py语法错误或导入失败如import database时sqlalchemy版本不匹配。根治方案在Dockerfile中加入启动前检查# 在CMD前添加 RUN echo Testing application import... \ python3.11 -c import main \ echo Import successful问题2Redis缓存始终MISSredis-cli ping成功但Python连接超时现象redis_client.get_query_cache()总返回None但docker exec -it lcoder_redis_1 redis-cli ping返回PONG。原因Docker网络DNS解析失败。redis服务名在api容器内解析为127.0.0.1而非实际IP。解决检查docker-compose.yml中api服务的depends_on是否写错服务名在api容器内执行nslookup redis确认解析IP强制指定Redis连接参数redis.Redis( hostos.getenv(REDIS_HOST, redis), portint(os.getenv(REDIS_PORT, 6379)), dbint(os.getenv(REDIS_DB, 0)), # 关键加socket_keepalive socket_keepaliveTrue, socket_keepalive_options{ socket.TCP_KEEPIDLE: 60, socket.TCP_KEEPINTVL: 60, socket.TCP_KEEPCNT: 3 } )问题3MySQL连接池耗尽sqlalchemy.exc.TimeoutError现象高并发时大量请求报QueuePool limit of size 20 overflow 30 reached, refusing further growth。根因max_overflow30是临时连接上限但pool_timeout30太短连接未及时归还业务代码中session.close()未被调用如异常未进入finally块。修复步骤在database.py的get_db_session()中确保finally块存在增加连接泄漏检测engine create_async_engine( DB_URL, # ...其他参数 echo_poolTrue, # 开启连接池日志 pool_use_lifoTrue, # LIFO模式减少连接老化 )压测时用watch -n 1 docker exec lcoder_db_1 mysql -uroot -prootpassword -e SHOW STATUS LIKE \Threads_connected\监控连接数。问题4Pydantic v2模型校验失败ValidationError提示不清晰现象QueryRequest中question字段为空时报错Field required at [question]但前端不知道是哪个字段。优化方案自定义错误处理器from fastapi.exceptions import RequestValidationError from starlette.responses import JSONResponse app.exception_handler(RequestValidationError) async def validation_exception_handler(request, exc): errors [] for error in exc.errors(): errors.append({ field: ..join(str(loc) for loc in error[loc]), message: error[msg], type: error[type] }) return JSONResponse( status_code422, content{detail: 参数校验失败, errors: errors} )返回示例{ detail: 参数校验失败, errors: [ { field: question, message: Field required, type: missing } ] }问题5Docker构建时pip install超时报ReadTimeout现象docker build卡在pip install -r requirements.txt最终超时。解决方案换国内源在Dockerfile中RUN pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple/或用--trusted-hostRUN pip install --trusted-host pypi.tuna.tsinghua.edu.cn -r requirements.txt最后分享一个真实教训上周部署时我把pool_recycle3600设成了36000多写了个0结果MySQL的wait_timeout288008小时生效前连接池就强制回收了连接导致大量MySQL server has gone away错误。查日志花了3小时最终发现是单位错了——所有时间参数务必写单位如3600s而非3600哪怕文档说“单位秒”也要显式标注。这行代码现在刻在我显示器边框上。