1. 项目概述为什么“问数项目智能体”的基础设施必须从零手搭你手上正要启动一个叫“问数项目”的AI Agent——它得能听懂业务人员用大白话提的问题比如“上个月华东区销售额TOP3的产品是什么”然后自动查数据库、生成图表、再用自然语言把结论说清楚。这不是调个API就能搞定的事它背后是一整套协同运转的基础设施数据接入层得稳如磐石推理调度层得毫秒响应状态管理得不丢不乱接口暴露得安全可控。标题里那个“LCODER之AI Agent开发实战2”不是随便编号的它是整个系列里承上启下的关键一环——第一讲讲清楚了Agent的决策逻辑和技能编排这一讲就直奔命门把纸面上的架构图变成跑在你本地或服务器上、能真实扛住并发请求、能随时调试、能快速迭代的活系统。我带过十几支团队落地AI Agent项目踩过最深的坑90%都出在基础设施这一步。有人图快直接套用现成框架结果三个月后发现日志埋点全断、缓存策略无法定制、权限模型和公司现有体系对不上也有人迷信“云原生全家桶”硬上K8sIstioPrometheus结果连本地调试都要配半天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.Queueconcurrent.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 structlogFastAPI不只是“快”是“确定性快”FastAPI被选中核心在于它把“开发时确定性”和“运行时确定性”做到了极致。举个例子当你定义一个接口app.post(/query)参数是request: SqlQueryRequestFastAPI会在启动时就完成三件事静态类型检查用Pydantic解析SqlQueryRequest如果字段类型不匹配比如传了字符串当int直接在请求到达业务逻辑前就返回422错误并附带精确到字段的错误信息OpenAPI Schema生成自动生成符合规范的JSON SchemaSwagger UI能直接渲染出可交互的调试界面前端同学不用等后端写完文档就能联调异步执行路径固化所有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 HASHkey是task:{id}field是sql_result_a、sql_result_b、chart_data等每个field设置独立TTL比如sql_result_aTTL300秒chart_dataTTL60秒确保中间结果不过期滞留使用WATCHMULTI事务保证状态更新的原子性避免并发写入覆盖。更重要的是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_idtask_id, user_idrequest.state.user_id, endpointrequest.url.path ): response await call_next(request) return response # 在Skill里直接打日志 await logger.info(sql_query_executed, sqlSELECT * FROM sales WHERE product_id :pid, params{pid: P123}, duration_ms124.5, row_count42 )输出的日志是结构化的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依赖的pydantic2.0和fastapi依赖的pydantic2.0会直接冲突pip只会装最后一个导致运行时报ValidationError。Poetry的pyproject.toml则用poetry.lock文件精确锁定所有层级的依赖版本包括C扩展的ABI兼容性。实操步骤以Linux/macOS为例安装Poetry不要用pip install poetry官方推荐curl安装curl -sSL https://install.python-poetry.org | python3 - export PATH$HOME/.local/bin:$PATH # 加入shell配置初始化项目lcoder-agent-infra是项目名poetry init -n # -n跳过交互式提问 poetry env use 3.11 # 明确指定Python 3.11避免系统默认3.9导致兼容问题添加核心依赖注意版本约束poetry add fastapi[all]0.110.0,0.111.0 \ sqlalchemy[asyncio]2.0.0,2.1.0 \ redis4.6.0,4.7.0 \ structlog23.0.0,24.0.0 \ pydantic2.5.0,2.6.0 \ httpx0.25.0,0.26.0 \ python-dotenv1.0.0,1.1.0这里每个版本号都经过实测fastapi 0.110.x与pydantic 2.5.x的类型解析兼容性最佳redis 4.6.x的asyncio客户端稳定性远超4.5.xhttpx是FastAPI默认HTTP客户端必须与starlette版本对齐。生成可复现的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(fmtiso), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, structlog.processors.JSONRenderer() ], context_classdict, logger_factorystructlog.stdlib.LoggerFactory(), wrapper_classstructlog.stdlib.BoundLogger, cache_logger_on_first_useTrue, ) 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, errorstr(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( titleLCODER问数项目Agent, description面向业务人员的自然语言数据查询智能体, version0.1.0, lifespanlifespan, # 绑定生命周期 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_credentialsTrue, 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_idrequest_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_codeexc.status_code, detailexc.detail) return JSONResponse( status_codeexc.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, errorstr(exc)) return JSONResponse( status_code500, content{error: {code: 500, message: Internal server error}} )关键点解析lifespan确保数据库连接池、Redis客户端等在应用启动时初始化在关闭时优雅释放避免连接泄漏TrustedHostMiddleware防止HTTP Host头攻击生产环境必须配置allowed_hostsRequestIdMiddleware为每条请求生成唯一ID贯穿整个调用链是日志追踪的基石统一错误处理器把所有异常转成标准JSON格式前端不用解析不同格式的错误。3.3 数据库与Redis连接池如何避免“Too many connections”AI Agent的数据库连接不是“能连上就行”而是要应对突发查询高峰。比如市场部同事同时发起10个“季度销售报表”请求每个请求可能触发3次SQL查询瞬间产生30个并发连接。MySQL默认最大连接数151很快就会报错Too many connections。解决方案是双连接池策略SQLAlchemy Async Connection Pool用于长时查询from sqlalchemy.ext.asyncio import create_async_engine engine create_async_engine( mysqlaiomysql://user:passlocalhost:3306/dbname, echoFalse, # 生产关闭SQL打印 pool_size20, # 连接池初始大小 max_overflow30, # 超出pool_size后最多创建30个临时连接 pool_timeout30, # 获取连接超时秒数 pool_recycle3600, # 连接存活1小时后强制回收防MySQL wait_timeout pool_pre_pingTrue, # 每次获取连接前执行SELECT 1检测有效性 )Redis Connection Pool用于短时状态import redis.asyncio as redis redis_pool redis.ConnectionPool( hostlocalhost, port6379, db0, max_connections100, # Redis单实例通常支持万级连接但Pool设100足够 decode_responsesTrue, # 自动decode bytes to str health_check_interval30, # 每30秒ping一次保活 ) redis_client redis.Redis(connection_poolredis_pool)注意max_overflow不是越大越好。MySQL的max_connections是全局限制pool_size max_overflow总和不能超过它。我们实测过pool_size20, max_overflow30在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_factorydict, descriptionSQL参数字典) timeout: float Field(30.0, ge1.0, le300.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(..., descriptionSQL哈希值用于缓存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_hashquery_hash, row_countrow_count, execution_time_msround(execution_time_ms, 2), timeoutrequest.timeout ) return SqlQueryResponse( rowsrows, row_countrow_count, execution_time_msround(execution_time_ms, 2), query_hashquery_hash ) except Exception as e: execution_time_ms (asyncio.get_event_loop().time() - start_time) * 1000 logger.exception(sql_query_failed, sql_hashquery_hash if query_hash in locals() else unknown, errorstr(e), execution_time_msround(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也超时。排查路径确认端口是否被占用lsof -i :8000macOS/Linux或netstat -ano | findstr :8000Windows。常见冲突程序另一个Uvicorn实例、Docker容器、VS Code的Live Server。检查事件循环是否被阻塞在main.py顶部加一行import asyncio; print(asyncio.get_event_loop())如果输出_UnixSelectorEventLoop runningFalse closedFalse说明事件循环未启动。根本原因是你在lifespan里写了同步阻塞代码如time.sleep(10)或者某个Dependency初始化时用了requests.get()而非httpx.AsyncClient().get()。验证FastAPI路由是否注册在main.py末尾加print(app.routes)正常应输出类似[Route path/healthz namehealth_check methods[GET], ...]。如果为空说明app.get()装饰器没生效常见原因是app变量名被覆盖如app FastAPI()后又写了app some_function()、或路由函数定义在if __name__ __main__:块外但未执行。实操心得我习惯在lifespan的try块里加logger.info(DB pool initialized, sizeapp.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前显式转strsafe_params {k: str(v) for k, v in request.params.items()} stmt text(request.sql).bindparams(**safe_params)事务未提交如果AsyncSession是在begin()事务中创建的且未commit()查询可能看不到其他事务的变更。AI Agent的查询应默认用autocommitFalse的AsyncSession但确保不在事务里执行只读查询。4.3 “Redis缓存未命中率100%”TTL与Key设计失误现象SqlQuerySkill每次查询都打到数据库Redis的INFO keyspace显示db0:keys0,expires0缓存完全没生效。高频错误Key包含动态值缓存Key用了fquery:{request.sql}:{request.params}但request.sql含换行符、空格request.params是{cat: electronics}和{cat:electronics}空格差异被视为不同Key。正确做法用hashlib.md5对sqljson.dumps(params, sort_keysTrue)哈希。TTL设为0redis_client.setex(key, 0, value)TTL0表示永不过期但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_factoryPrintLoggerFactory(), # 或 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压到120MBFROM 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 4CPU核心数--limit-concurrency 100在4核8G机器上稳定支撑QPS 300。5.2 Prometheus监控集成盯紧Agent的“生命体征”AI Agent的健康不能只看HTTP 200。我们需要四个黄金指标指标名类型查询示例业务意义fastapi_http_requests_total{status_code~5..}[1h]Counterrate(fastapi_http_requests_total{status_code~5..}[5m])5xx错误率超过0.1%需告警sql_query_execution_seconds_sum / sql_query_execution_seconds_countHistogramhistogram_quantile(0.95, sum(rate(sql_query_execution_seconds_bucket[1h])) by (le))SQL查询P95耗时超过2s需优化redis_connected_clientsGaugeredis_connected_clients{addr~redis:6379}Redis连接数持续80需扩容process_resident_memory_bytesGaugeprocess_resident_memory_bytes{joblcoder-agent}进程内存突增50%可能内存泄漏在main.py中集成Prometheusfrom prometheus_fastapi_instrumentator import Instrumentator # 在app创建后立即初始化 Instrumentator().instrument(app).expose(app)访问http://localhost:8000/metrics即可看到所有指标。配合Grafana一张Dashboard就能看清哪个Skill最慢、哪个SQL最耗资源、Redis连接是否健康。5.3 日志聚合用LokiPromtail构建低成本可观测性比起ELKElasticsearchLogstashKibana动辄16G内存的开销LokiPromtail方案只需512MB内存Promtail部署在Agent服务器tail日志文件按{joblcoder-agent, instanceserver-01}打标签推送到LokiLoki单节点部署存储结构化日志支持LogQL查询Grafana可视化用{joblcoder-agent} | sql_query_success查成功SQL{joblcoder-agent} |~ error|exception查