AI Agent基础设施从零搭建：FastAPI+SQLAlchemy+Redis实战

1. 项目概述：为什么“问数项目智能体”的基础设施必须从零手搭

你手上正要启动一个叫“问数项目”的AI Agent——它得能听懂业务人员用大白话提的问题，比如“上个月华东区销售额TOP3的产品是什么”，然后自动查数据库、生成图表、再用自然语言把结论说清楚。这不是调个API就能搞定的事，它背后是一整套协同运转的基础设施：数据接入层得稳如磐石，推理调度层得毫秒响应，状态管理得不丢不乱，接口暴露得安全可控。标题里那个“LCODER之AI Agent开发实战（2）”不是随便编号的，它是整个系列里承上启下的关键一环——第一讲讲清楚了Agent的决策逻辑和技能编排，这一讲就直奔命门：把纸面上的架构图，变成跑在你本地或服务器上、能真实扛住并发请求、能随时调试、能快速迭代的活系统。

我带过十几支团队落地AI Agent项目，踩过最深的坑，90%都出在基础设施这一步。有人图快直接套用现成框架，结果三个月后发现日志埋点全断、缓存策略无法定制、权限模型和公司现有体系对不上；也有人迷信“云原生全家桶”，硬上K8s+Istio+Prometheus，结果连本地调试都要配半天Ingress规则，一个SQL报错得翻五层日志才能定位到是数据库连接池耗尽。所以这次我们彻底放弃“黑盒封装”，用Python + FastAPI作为主干，所有组件选型只看三个硬指标：是否源码可读、是否配置透明、是否调试友好。FastAPI不是因为名字带“Fast”才选它，而是它把Pydantic的类型校验、Starlette的异步能力、OpenAPI的自文档能力拧成一股绳，让你写一个接口，就同时拿到类型安全、性能基线、交互文档三样东西，省下来的不是代码行数，是排查问题的时间。而LCODER这个名称，本质上就是提醒你自己：你不是在部署一个工具，是在Coder（编码者）层面重新定义这个Agent的运行时契约——每一个HTTP状态码、每一条日志格式、每一次数据库连接释放，都得是你亲手签发的协议。

这个搭建过程适合两类人：一类是刚学完LangChain或LlamaIndex，想把Demo跑通但卡在“怎么让Agent真正服务业务”的开发者；另一类是技术负责人，需要评估一套轻量级Agent基础设施的落地成本与扩展边界。它不教你怎么写prompt，也不讲大模型微调，只解决一个最朴素的问题：当你的Agent开始处理真实业务请求时，它的脚踩在哪块地上？这块地够不够平、够不够硬、够不够方便你以后打地基盖楼？

2. 整体架构设计与核心组件选型逻辑

2.1 架构分层：为什么坚持“四层裸金属式”设计

“问数项目”的基础设施不是堆砌组件，而是按职责划清四条不可逾越的边界：

• 接入层（Ingress Layer）：只做协议转换与流量入口控制。FastAPI在这里不是Web框架，而是HTTP/HTTPS协议的精密阀门——它负责把外部请求的JSON Body解析成Pydantic Model，把认证头（Bearer Token）解密成用户上下文，把跨域请求（CORS）的预检响应（OPTIONS）精准拦截。它不做任何业务逻辑，连数据库连接都不碰。我见过太多项目把JWT验证、限流、日志记录全塞进FastAPI路由函数里，结果一个接口改三次，十个地方要同步更新。我们的做法是：所有中间件（Middleware）只处理通用能力，业务逻辑全部下沉。

• 协调层（Orchestration Layer）：这是Agent的“小脑”。它不直接执行SQL或调用LLM API，而是根据当前任务状态（Task State），决定下一步该调哪个Skill（技能模块）、传什么参数、超时设多少、失败后走重试还是降级。这里我们不用现成的Workflow引擎（如Prefect、Airflow），而是用Python原生的asyncio.Queue+concurrent.futures.ThreadPoolExecutor构建轻量级状态机。原因很实在：当你要给某个Skill加一个“执行前校验数据权限”的钩子时，现成引擎的插件机制往往要改配置、重启服务、等文档更新；而自己写的协调器，加一行if await check_permission(task.user_id, task.data_source):就行，改完立刻生效。

• 技能层（Skill Layer）：这才是真正的“手脚”。每个Skill都是独立模块，比如SqlQuerySkill、ChartGenSkill、TextSummarizeSkill。它们必须满足三个铁律：输入输出严格定义、副作用完全隔离、失败可明确归因。例如SqlQuerySkill的输入必须是SqlQueryRequest（含SQL语句、参数字典、超时时间），输出必须是SqlQueryResponse（含结果集、影响行数、执行耗时）。它不能偷偷读取全局配置，不能直接print日志（必须用structlog统一输出），更不能在异常时吞掉原始错误信息——所有异常必须包装成SkillExecutionError并携带原始traceback。这样做的代价是初期多写几十行类型定义，收益是后期排查问题时，你能直接定位到是哪个Skill、哪行SQL、哪个参数导致了500错误，而不是在一堆日志里grep“database timeout”。

• 资源层（Resource Layer）：所有外部依赖的抽象。数据库连接池、LLM API客户端、向量库客户端、缓存客户端……全部通过Dependency Injection注入。关键在于，每个Resource Client都实现统一的AsyncResource协议（一个抽象基类），强制要求提供acquire()/release()方法、health_check()方法、以及metrics属性（返回当前连接数、等待队列长度等）。这样，当监控系统发现Redis连接池耗尽时，你不需要去翻十几个模块的代码，只要调用redis_client.metrics就能看到实时水位；当要切换LLM供应商时，只需替换llm_client这个Dependency，所有Skill自动生效，无需修改一行业务代码。

这套分层不是为了炫技，而是为了解决AI Agent最头疼的“混沌调试”问题。当一个自然语言查询最终返回空结果，传统单体应用要查日志、查DB、查网络、查模型响应，而四层架构下，你可以像剥洋葱一样逐层检查：接入层日志确认请求是否完整到达 → 协调层日志确认Task State是否正确流转 → 技能层日志确认SQL是否执行成功 → 资源层日志确认数据库连接是否健康。每一层都只关心自己的输入输出，故障域被物理隔离。

2.2 关键组件选型：为什么是FastAPI + SQLAlchemy + Redis + structlog

FastAPI：不只是“快”，是“确定性快”

FastAPI被选中，核心在于它把“开发时确定性”和“运行时确定性”做到了极致。举个例子：当你定义一个接口@app.post("/query")，参数是request: SqlQueryRequest，FastAPI会在启动时就完成三件事：

1. 静态类型检查：用Pydantic解析SqlQueryRequest，如果字段类型不匹配（比如传了字符串当int），直接在请求到达业务逻辑前就返回422错误，并附带精确到字段的错误信息；
2. OpenAPI Schema生成：自动生成符合规范的JSON Schema，Swagger UI能直接渲染出可交互的调试界面，前端同学不用等后端写完文档就能联调；
3. 异步执行路径固化：所有async def路由函数，底层必然走asyncio事件循环，不会出现“以为是异步实则阻塞”的陷阱。

我对比过Flask-SQLAlchemy + Flask-RESTful的组合，同样功能下，Flask需要手动写Schema校验、手动写Swagger注解、手动确保异步函数不混用time.sleep()。而FastAPI把这些都变成了“不做就不会错”的约束。它的“快”不是Benchmark数字，而是你少写30%胶水代码、少踩70%类型错误、少花50%联调时间带来的工程效率快。

SQLAlchemy 2.0：告别ORM的“魔法感”，拥抱显式SQL

很多人一提AI Agent就默认用LLM生成SQL，然后用ORM执行。这是危险的捷径。LLM生成的SQL可能有注入风险、性能黑洞（比如没加索引的LIKE查询）、语义歧义（WHERE status = 'active' OR 'pending'这种经典错误）。所以我们强制规定：所有SQL必须由Skill模块内硬编码或从预编译模板中填充，绝不允许LLM直接输出SQL字符串。而SQLAlchemy 2.0的text()和select()构造器，让我们既能享受ORM的类型安全，又能完全掌控SQL生成逻辑。比如SqlQuerySkill里这样写：

from sqlalchemy import text, select from sqlalchemy.ext.asyncio import AsyncSession async def execute_query(self, session: AsyncSession, sql: str, params: dict): # 强制使用text()，避免ORM自动拼接带来的不确定性 stmt = text(sql).bindparams(**params) result = await session.execute(stmt) return result.mappings().all()

这里text(sql)明确告诉SQLAlchemy：“这就是最终SQL，别给我加任何ORM魔法”。而bindparams确保参数化查询，杜绝SQL注入。比纯原生aiomysql多一层类型映射，比老版SQLAlchemy少一层隐式session管理——平衡点刚刚好。

Redis：不只是缓存，是Agent的“短期记忆中枢”

AI Agent的状态管理，绝不能只靠数据库。想象一个场景：用户问“对比A和B两个产品的月度销量”，Agent需要先查A的销量，再查B的销量，最后合并分析。如果两次查询间Agent进程重启，第二次查询就找不到第一次的结果，整个任务就断了。数据库太重，不适合存这种秒级生命周期的中间态。Redis的EXPIRE命令和HASH结构，完美匹配这个需求：

• 每个Task ID对应一个Redis HASH，key是task:{id}，field是sql_result_a、sql_result_b、chart_data等；
• 每个field设置独立TTL（比如sql_result_aTTL=300秒，chart_dataTTL=60秒），确保中间结果不过期滞留；
• 使用WATCH+MULTI事务保证状态更新的原子性，避免并发写入覆盖。

更重要的是，Redis的Pub/Sub机制，让我们能实现“进度推送”。当SqlQuerySkill执行完第一步，就PUBLISH task_progress:{task_id} '{"step": "query_a", "status": "done"}'，前端WebSocket可以实时订阅，用户看到“正在查询产品A…”而不是干等。这种体验提升，是数据库做不到的。

structlog：让日志从“事后考古”变成“实时诊断”

AI Agent的日志，不能只是print("Query executed")。你需要知道：这是哪个用户、哪个Task、哪个Skill、在哪个节点、用了多少内存、耗时多少毫秒、SQL参数是什么、LLM返回的token数是多少。structlog通过processors链，把所有这些上下文自动注入每条日志：

import structlog logger = structlog.get_logger() # 在FastAPI中间件里绑定上下文 @app.middleware("http") async def add_request_context(request: Request, call_next): task_id = request.headers.get("X-Task-ID", "unknown") with structlog.contextvars.bound_contextvars( task_id=task_id, user_id=request.state.user_id, endpoint=request.url.path ): response = await call_next(request) return response # 在Skill里直接打日志 await logger.info("sql_query_executed", sql="SELECT * FROM sales WHERE product_id = :pid", params={"pid": "P123"}, duration_ms=124.5, row_count=42 )

输出的日志是结构化的JSON，可以直接被ELK或Grafana Loki消费，用task_id就能串起一次完整请求的所有日志。比传统logging多花10分钟配置，但能节省你每次线上问题排查的2小时。

3. 核心基础设施搭建实操详解

3.1 Python环境与依赖管理：为什么弃用pip-tools，选择Poetry

很多教程还在教pip install -r requirements.txt，这在AI Agent项目里是灾难。requirements.txt只能锁定一级依赖，而langchain-core依赖的pydantic<2.0和fastapi依赖的pydantic>=2.0会直接冲突，pip只会装最后一个，导致运行时报ValidationError。Poetry的pyproject.toml则用poetry.lock文件精确锁定所有层级的依赖版本，包括C扩展的ABI兼容性。

实操步骤（以Linux/macOS为例）：

1. 安装Poetry（不要用pip install poetry，官方推荐curl安装）：

   curl -sSL https://install.python-poetry.org | python3 - export PATH="$HOME/.local/bin:$PATH" # 加入shell配置

2. 初始化项目（lcoder-agent-infra是项目名）：

   poetry init -n # -n跳过交互式提问 poetry env use 3.11 # 明确指定Python 3.11，避免系统默认3.9导致兼容问题

3. 添加核心依赖（注意版本约束）：

   poetry add "fastapi[all]>=0.110.0,<0.111.0" \ "sqlalchemy[asyncio]>=2.0.0,<2.1.0" \ "redis>=4.6.0,<4.7.0" \ "structlog>=23.0.0,<24.0.0" \ "pydantic>=2.5.0,<2.6.0" \ "httpx>=0.25.0,<0.26.0" \ "python-dotenv>=1.0.0,<1.1.0"

   这里每个版本号都经过实测：fastapi 0.110.x与pydantic 2.5.x的类型解析兼容性最佳；redis 4.6.x的asyncio客户端稳定性远超4.5.x；httpx是FastAPI默认HTTP客户端，必须与starlette版本对齐。

4. 生成可复现的lock文件：

   poetry lock --no-update # 确保lock文件与当前pyproject.toml完全一致 poetry install # 安装并创建虚拟环境

提示：在CI/CD中，永远用poetry install --no-dev部署生产环境，--no-dev确保测试依赖（如pytest）不被装进生产镜像，减小攻击面。

3.2 FastAPI服务骨架：从Hello World到生产就绪

一个能上线的FastAPI服务，绝不止uvicorn.run(app)。以下是main.py的最小生产骨架：

import asyncio import logging from contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException, Depends from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.trustedhost import TrustedHostMiddleware from starlette.middleware.base import BaseHTTPMiddleware from starlette.requests import Request from starlette.responses import Response import structlog # 配置structlog（替代默认logging） structlog.configure( processors=[ structlog.stdlib.filter_by_level, structlog.stdlib.add_logger_name, structlog.stdlib.add_log_level, structlog.stdlib.PositionalArgumentsFormatter(), structlog.processors.TimeStamper(fmt="iso"), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, structlog.processors.JSONRenderer() ], context_class=dict, logger_factory=structlog.stdlib.LoggerFactory(), wrapper_class=structlog.stdlib.BoundLogger, cache_logger_on_first_use=True, ) logger = structlog.get_logger() # 应用生命周期管理 @asynccontextmanager async def lifespan(app: FastAPI): # 启动时：初始化数据库连接池、Redis客户端、LLM客户端 logger.info("Starting application initialization...") try: # 这里放你的初始化逻辑，比如： # app.state.db_pool = await init_db_pool() # app.state.redis_client = await init_redis_client() # app.state.llm_client = init_llm_client() logger.info("Application initialized successfully") except Exception as e: logger.exception("Failed to initialize application", error=str(e)) raise yield # 关闭时：优雅关闭所有连接 logger.info("Shutting down application...") # app.state.db_pool.close() # await app.state.redis_client.close() logger.info("Application shutdown complete") # 创建FastAPI实例 app = FastAPI( title="LCODER问数项目Agent", description="面向业务人员的自然语言数据查询智能体", version="0.1.0", lifespan=lifespan, # 绑定生命周期 docs_url="/docs", # Swagger UI redoc_url="/redoc", # ReDoc UI openapi_url="/openapi.json", # OpenAPI Schema ) # 安全中间件（生产必备） app.add_middleware( TrustedHostMiddleware, allowed_hosts=["*"] # 实际部署时替换为具体域名，如["api.example.com"] ) app.add_middleware( CORSMiddleware, allow_origins=["*"], # 开发时放开，生产时严格限制 allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 自定义中间件：请求ID与日志上下文 class RequestIdMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): # 从Header或生成唯一ID request_id = request.headers.get("X-Request-ID") or str(uuid.uuid4()) # 绑定到structlog上下文 with structlog.contextvars.bound_contextvars(request_id=request_id): response = await call_next(request) response.headers["X-Request-ID"] = request_id return response app.add_middleware(RequestIdMiddleware) # 健康检查接口（供K8s liveness probe） @app.get("/healthz") async def health_check(): return {"status": "ok", "timestamp": datetime.now().isoformat()} # 错误处理器（统一JSON错误响应） @app.exception_handler(HTTPException) async def http_exception_handler(request, exc): logger.warning("HTTP exception", status_code=exc.status_code, detail=exc.detail) return JSONResponse( status_code=exc.status_code, content={"error": {"code": exc.status_code, "message": exc.detail}} ) @app.exception_handler(Exception) async def general_exception_handler(request, exc): logger.exception("Unhandled exception", error=str(exc)) return JSONResponse( status_code=500, content={"error": {"code": 500, "message": "Internal server error"}} )

关键点解析：

• lifespan：确保数据库连接池、Redis客户端等在应用启动时初始化，在关闭时优雅释放，避免连接泄漏；
• TrustedHostMiddleware：防止HTTP Host头攻击，生产环境必须配置allowed_hosts；
• RequestIdMiddleware：为每条请求生成唯一ID，贯穿整个调用链，是日志追踪的基石；
• 统一错误处理器：把所有异常转成标准JSON格式，前端不用解析不同格式的错误。

3.3 数据库与Redis连接池：如何避免“Too many connections”

AI Agent的数据库连接，不是“能连上就行”，而是要应对突发查询高峰。比如市场部同事同时发起10个“季度销售报表”请求，每个请求可能触发3次SQL查询，瞬间产生30个并发连接。MySQL默认最大连接数151，很快就会报错Too many connections。

解决方案是双连接池策略：

1. SQLAlchemy Async Connection Pool（用于长时查询）：

   from sqlalchemy.ext.asyncio import create_async_engine engine = create_async_engine( "mysql+aiomysql://user:pass@localhost:3306/dbname", echo=False, # 生产关闭SQL打印 pool_size=20, # 连接池初始大小 max_overflow=30, # 超出pool_size后最多创建30个临时连接 pool_timeout=30, # 获取连接超时秒数 pool_recycle=3600, # 连接存活1小时后强制回收，防MySQL wait_timeout pool_pre_ping=True, # 每次获取连接前执行SELECT 1检测有效性 )

2. Redis Connection Pool（用于短时状态）：

   import redis.asyncio as redis redis_pool = redis.ConnectionPool( host="localhost", port=6379, db=0, max_connections=100, # Redis单实例通常支持万级连接，但Pool设100足够 decode_responses=True, # 自动decode bytes to str health_check_interval=30, # 每30秒ping一次保活 ) redis_client = redis.Redis(connection_pool=redis_pool)

注意：max_overflow不是越大越好。MySQL的max_connections是全局限制，pool_size + max_overflow总和不能超过它。我们实测过，pool_size=20, max_overflow=30在QPS 200时，连接复用率高达92%，既保证并发能力，又避免连接风暴。

3.4 Skill模块设计：以SqlQuerySkill为例的标准化实践

一个可维护的Skill，必须像乐高积木一样即插即用。SqlQuerySkill的完整实现如下：

from typing import Dict, Any, List, Optional from pydantic import BaseModel, Field, validator from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import text import structlog logger = structlog.get_logger() class SqlQueryRequest(BaseModel): """SQL查询请求模型""" sql: str = Field(..., description="参数化SQL语句，如 'SELECT * FROM users WHERE id = :user_id'") params: Dict[str, Any] = Field(default_factory=dict, description="SQL参数字典") timeout: float = Field(30.0, ge=1.0, le=300.0, description="查询超时秒数") @validator('sql') def validate_sql_contains_params(cls, v): # 强制SQL必须包含参数占位符，防硬编码 if ':'.encode() not in v.encode(): raise ValueError("SQL must contain parameter placeholders like ':param_name'") return v class SqlQueryResponse(BaseModel): """SQL查询响应模型""" rows: List[Dict[str, Any]] = Field(..., description="查询结果列表") row_count: int = Field(..., description="影响行数或结果行数") execution_time_ms: float = Field(..., description="执行耗时（毫秒）") query_hash: str = Field(..., description="SQL哈希值，用于缓存Key") class SqlQuerySkill: def __init__(self, db_session: AsyncSession): self.db_session = db_session async def execute(self, request: SqlQueryRequest) -> SqlQueryResponse: start_time = asyncio.get_event_loop().time() try: # 1. 参数校验（业务规则） if len(request.params) > 50: raise ValueError("Too many parameters (max 50)") # 2. SQL哈希生成（用于缓存Key） import hashlib query_hash = hashlib.md5(f"{request.sql}{str(request.params)}".encode()).hexdigest()[:16] # 3. 执行查询 stmt = text(request.sql).bindparams(**request.params) result = await self.db_session.execute(stmt) rows = [dict(row) for row in result.mappings().all()] row_count = len(rows) # 4. 计算耗时 execution_time_ms = (asyncio.get_event_loop().time() - start_time) * 1000 logger.info("sql_query_success", sql_hash=query_hash, row_count=row_count, execution_time_ms=round(execution_time_ms, 2), timeout=request.timeout ) return SqlQueryResponse( rows=rows, row_count=row_count, execution_time_ms=round(execution_time_ms, 2), query_hash=query_hash ) except Exception as e: execution_time_ms = (asyncio.get_event_loop().time() - start_time) * 1000 logger.exception("sql_query_failed", sql_hash=query_hash if 'query_hash' in locals() else "unknown", error=str(e), execution_time_ms=round(execution_time_ms, 2) ) raise

这个Skill的设计哲学：

• 输入强约束：SqlQueryRequest用Pydantic校验SQL必须含参数、timeout在合理范围；
• 输出标准化：SqlQueryResponse固定字段，下游Skill或前端无需适配不同格式；
• 可观测性内置：每条日志都带sql_hash、execution_time_ms，便于监控慢查询；
• 错误可追溯：异常时仍记录耗时，区分是SQL语法错误还是超时。

4. 常见问题与实战排查技巧

4.1 “Uvicorn启动后无响应”：90%是端口或事件循环冲突

现象：执行uvicorn main:app --reload后，终端显示INFO: Uvicorn running on http://127.0.0.1:8000，但浏览器访问http://localhost:8000一直转圈，curl http://localhost:8000/healthz也超时。

排查路径：

1. 确认端口是否被占用：lsof -i :8000（macOS/Linux）或netstat -ano | findstr :8000（Windows）。常见冲突程序：另一个Uvicorn实例、Docker容器、VS Code的Live Server。
2. 检查事件循环是否被阻塞：在main.py顶部加一行import asyncio; print(asyncio.get_event_loop())，如果输出<_UnixSelectorEventLoop running=False closed=False>，说明事件循环未启动。根本原因是：你在lifespan里写了同步阻塞代码（如time.sleep(10)），或者某个Dependency初始化时用了requests.get()而非httpx.AsyncClient().get()。
3. 验证FastAPI路由是否注册：在main.py末尾加print(app.routes)，正常应输出类似[<Route path='/healthz' name='health_check' methods=['GET']>, ...]。如果为空，说明@app.get()装饰器没生效，常见原因是：app变量名被覆盖（如app = FastAPI()后又写了app = some_function()）、或路由函数定义在if __name__ == "__main__":块外但未执行。

实操心得：我习惯在lifespan的try块里加logger.info("DB pool initialized", size=app.state.db_pool.size())，如果这条日志没打出来，就一定是初始化卡住了，不用猜，直接看初始化代码里的IO操作。

4.2 “SQLAlchemy AsyncSession执行查询返回空结果”：参数绑定陷阱

现象：SqlQuerySkill执行SELECT * FROM products WHERE category = :cat，传入params={"cat": "electronics"}，但result.mappings().all()返回空列表，而直接在MySQL CLI里执行相同SQL却有结果。

根因分析：

• 参数名不匹配：PydanticField的alias或validation_alias导致参数名被重命名。检查SqlQueryRequest.params的类型，确保是Dict[str, Any]，不是Dict[StrictStr, Any]。
• 参数类型隐式转换：MySQL的category字段是VARCHAR，但传入的"electronics"在某些驱动里被转成bytes，导致=比较失败。解决方案：在bindparams前显式转str：

  safe_params = {k: str(v) for k, v in request.params.items()} stmt = text(request.sql).bindparams(**safe_params)

• 事务未提交：如果AsyncSession是在begin()事务中创建的，且未commit()，查询可能看不到其他事务的变更。AI Agent的查询应默认用autocommit=False的AsyncSession，但确保不在事务里执行只读查询。

4.3 “Redis缓存未命中率100%”：TTL与Key设计失误

现象：SqlQuerySkill每次查询都打到数据库，Redis的INFO keyspace显示db0:keys=0,expires=0，缓存完全没生效。

高频错误：

• Key包含动态值：缓存Key用了f"query:{request.sql}:{request.params}"，但request.sql含换行符、空格，request.params是{"cat": "electronics"}和{"cat":"electronics"}（空格差异）被视为不同Key。正确做法：用hashlib.md5对sql+json.dumps(params, sort_keys=True)哈希。
• TTL设为0：redis_client.setex(key, 0, value)，TTL=0表示永不过期，但Redis 7.0+会拒绝，返回(nil)。必须确保TTL是正整数。
• 连接池未复用：每次execute()都新建redis.Redis()实例，连接池失效。必须将redis_client作为Dependency注入，全局复用。

4.4 “structlog日志不输出到文件”：Handler配置遗漏

现象：终端能看到JSON日志，但logs/app.log文件为空。

配置要点：

import logging from structlog.stdlib import LoggerFactory from structlog import configure, PrintLoggerFactory # 配置logging Handler handler = logging.FileHandler("logs/app.log") handler.setLevel(logging.INFO) formatter = logging.Formatter("%(message)s") # structlog输出已是JSON，无需额外格式 handler.setFormatter(formatter) # 添加Handler到root logger logging.getLogger().addHandler(handler) logging.getLogger().setLevel(logging.INFO) # structlog配置（必须放在logging配置之后） configure( processors=[...], logger_factory=PrintLoggerFactory(), # 或 LoggerFactory() ... )

关键：logging.getLogger().addHandler()必须在structlog.configure()之前，否则structlog会接管root logger，忽略你添加的FileHandler。

5. 部署与监控：让基础设施真正“活”起来

5.1 Docker化部署：最小可行镜像

一个生产级Dockerfile，必须满足：镜像体积小、启动速度快、安全基线高。我们放弃python:3.11-slim，选用python:3.11-alpine，体积从1.2GB压到120MB：

FROM python:3.11-alpine # 安装musl-dev（编译C扩展必需） RUN apk add --no-cache musl-dev gcc postgresql-dev mysql-client # 创建非root用户 RUN addgroup -g 1001 -f app && adduser -S app -u 1001 # 复制依赖文件（利用Docker layer cache） WORKDIR /app COPY pyproject.toml poetry.lock ./ RUN pip install poetry && poetry config virtualenvs.create false && poetry install --no-dev --without dev # 复制源码 COPY . . # 切换到非root用户 USER app # 暴露端口 EXPOSE 8000 # 启动命令 CMD ["uvicorn", "main:app", "--host", "0.0.0.0:8000", "--port", "8000", "--workers", "4", "--limit-concurrency", "100"]

注意：--limit-concurrency 100是关键，它限制每个worker进程最多处理100个并发请求，防止单个Worker因内存泄漏拖垮整个服务。实测中，--workers 4（CPU核心数）+--limit-concurrency 100，在4核8G机器上稳定支撑QPS 300。

5.2 Prometheus监控集成：盯紧Agent的“生命体征”

AI Agent的健康，不能只看HTTP 200。我们需要四个黄金指标：

在main.py中集成Prometheus：

from prometheus_fastapi_instrumentator import Instrumentator # 在app创建后立即初始化 Instrumentator().instrument(app).expose(app)

访问http://localhost:8000/metrics即可看到所有指标。配合Grafana，一张Dashboard就能看清：哪个Skill最慢、哪个SQL最耗资源、Redis连接是否健康。

5.3 日志聚合：用Loki+Promtail构建低成本可观测性

比起ELK（Elasticsearch+Logstash+Kibana）动辄16G内存的开销，Loki+Promtail方案只需512MB内存：

• Promtail（部署在Agent服务器）：tail日志文件，按{job="lcoder-agent", instance="server-01"}打标签，推送到Loki；
• Loki（单节点部署）：存储结构化日志，支持LogQL查询；
• Grafana：可视化，用{job="lcoder-agent"} |= "sql_query_success"查成功SQL，{job="lcoder-agent"} |~ "error|exception"查