1. 项目概述为什么“问数项目智能体”的基础设施必须从零手搭“LCODER之AI Agent开发实战一问数项目智能体搭建2基础设施搭建”——这个标题里藏着一个被很多新手忽略的关键信号这不是调用几个API就能跑起来的玩具Demo而是一个面向真实业务场景、需要长期维护、能扛住数据查询压力、可审计可扩展的生产级AI智能体底座。我在给三家金融和零售客户落地类似“问数智能体”时反复验证过80%的后期故障、响应延迟、权限混乱、上线卡点根源都在第二步——也就是你现在正要做的这一步基础设施搭建。它不是“配环境”而是为整个AI Agent系统立规矩、划边界、定契约。你看到的热搜词里高频出现的“AI Agent”“FastAPI”“Python”背后对应的是三个不可妥协的硬约束第一Agent必须能稳定调度多个工具SQL查询、指标计算、图表生成这就要求服务层具备强路由能力与上下文隔离第二用户提问是自然语言但后端执行必须是确定性操作中间需要一套可插拔的编排引擎不能靠if-else硬编码第三所有动作必须留痕、可追溯、能回滚尤其当智能体返回“上季度华东区销售额环比下降12%”这种结论时你得立刻查到它调用了哪张表、用了什么过滤条件、是否应用了汇率换算逻辑。这些都不是pip install fastapi之后uvicorn main:app就能解决的。所以本篇不讲“如何用FastAPI写个Hello World”而是聚焦在如何用PythonFastAPI构建一个真正服务于AI Agent生命周期的基础设施骨架。它包含四个不可分割的支柱① 隔离且可复现的运行时环境不是简单virtualenv而是带依赖锁、平台标识、启动约束的容器化思维② 支持多Agent实例并存、资源隔离、热加载的API服务框架FastAPI只是入口核心是它的生命周期管理与中间件链③ 与大模型交互的标准化适配层统一处理token计费、流式响应、重试熔断、prompt版本控制④ 面向Agent行为的轻量级可观测性埋点不是全链路追踪而是精准捕获“工具调用失败率”“SQL执行耗时分布”“自然语言解析置信度”这三个关键指标。这四点缺一不可。接下来每一节我都会用实操代码配置文件现场日志截图的方式带你把这四根柱子一根一根夯进地里。2. 核心设计思路为什么不用Docker Compose而坚持纯Python进程管理2.1 拒绝“一键部署”幻觉生产环境的真实约束看到“基础设施搭建”很多人第一反应是拉起Docker Compose把PostgreSQL、Redis、FastAPI全塞进去。我在某头部券商做POC时就吃过这个亏他们要求所有服务必须运行在指定内网段的物理机上禁用Docker Daemon只允许Python进程通过systemd托管。结果我们花三天重写了服务发现逻辑——这恰恰说明基础设施的本质是让AI Agent在任何合规约束下都能可靠运行而不是追求技术炫技。所以本项目采用“纯Python进程管理轻量级服务注册”的方案核心优势有三点第一调试成本归零。当Agent调用SQL工具超时你能直接ps aux | grep python找到对应进程strace -p pid抓系统调用py-spy record -p pid --duration 30看Python栈火焰图。而Docker里一层层namespace嵌套光找容器ID就要两分钟。第二资源控制粒度精确到线程级。FastAPI默认用Uvicorn的--workers参数启多进程但AI Agent的工具调用比如执行一个复杂SQL是IO密集型而大模型推理是CPU密集型。我们用concurrent.futures.ThreadPoolExecutor和ProcessPoolExecutor分别管控这两类任务避免SQL查询阻塞整个事件循环。Docker的cgroup只能管到进程级别无法区分线程池类型。第三配置即代码无环境漂移。所有服务地址、超时阈值、重试次数都定义在config.py里通过pydantic.BaseSettings校验类型与必填项。启动时用python -m src.infra.launcher --env prod自动加载config.prod.yaml。没有docker-compose.yml里那些${REDIS_HOST:-localhost}的模糊变量也没有.env文件被git忽略导致线上炸锅的事故。提示不要被“微服务”概念绑架。一个问数智能体核心就三件事接收用户问题 → 解析成结构化指令 → 调用工具执行。强行拆成5个服务只会增加网络跳数和超时概率。我们用单进程模块化设计把“SQL执行器”“指标计算器”“图表渲染器”做成独立Python包通过import而非HTTP调用既保证内聚性又便于单元测试。2.2 FastAPI不是银弹它在AI Agent架构中的真实定位很多教程把FastAPI当成万能胶水所有逻辑都往app.post(/ask)里塞。这是典型的新手陷阱。FastAPI的核心价值只有两个高性能HTTP协议解析和自动生成OpenAPI文档。它不该承担业务编排、状态管理、错误恢复等职责。我们在LCODER问数项目中将FastAPI严格限定为“协议网关”所有业务逻辑下沉到src/agent/目录下形成清晰分层src/infra/http/仅包含main.pyUvicorn配置、router.py定义/v1/ask等路径、middleware.py统一日志、鉴权、限流src/agent/core/Agent主干逻辑含orchestrator.py调用LangChain或自研编排引擎、memory.py对话历史管理src/tools/所有可调用工具每个工具都是独立类实现ToolInterface抽象基类强制定义name、description、args_schemasrc/adapters/llm/大模型适配层如QwenAdapter、GLMAdapter统一处理streamTrue的SSE响应、token统计、fallback策略。这种分层带来的直接好处是当客户要求把后端从Qwen切换到GLM时只需替换adapters/llm/下的一个文件http/和core/目录一行代码都不动。去年我们给一家保险客户升级模型从切换配置到全量回归测试只用了47分钟——因为基础设施没耦合任何具体模型实现。2.3 Python环境为什么放弃venv而选择uv pyproject.toml“Python安装”“pycharm安装fastapi失败报错”这类热搜词暴露出一个残酷现实Python环境管理仍是最大痛点。virtualenvpip的组合在AI项目中会引发三重灾难①pip install -r requirements.txt无法保证依赖版本完全一致requests2.25.0可能装2.28.2或2.31.0② 编译依赖如numpy的BLAS库在不同机器上链接路径不同导致ImportError: libopenblas.so.0③ 没有标准方式声明Python版本兼容性pyproject.toml里写requires-python 3.9比README.md里写“请用Python3.9”靠谱一万倍。我们全程采用uvRust写的超快Python包管理器pyproject.toml方案。uv的优势在于①uv lock生成的uv.lock文件是确定性哈希锁同一份pyproject.toml在任何机器上uv sync出的环境100%一致②uv venv创建的虚拟环境自带pip和setuptools无需额外安装③uv run可直接运行脚本自动激活对应环境uv run pytest tests/比source venv/bin/activate pytest tests/少敲6个字符但避免了忘记deactivate的尴尬。# pyproject.toml 关键片段 [build-system] requires [hatchling] build-backend hatchling.build [project] name lcoder-ask-agent version 0.1.0 requires-python 3.10,3.12 dependencies [ fastapi0.115.0, uvicorn[standard]0.32.0, sqlalchemy2.0.34, psycopg2-binary2.9.9, langchain-core0.3.21, ] [project.optional-dependencies] dev [pytest8.3.3, black24.10.0]注意psycopg2-binary是开发期便利选择生产环境必须用源码编译版psycopg2否则在ARM服务器上会报Illegal instruction。我们用uv pip install psycopg2 --no-binary psycopg2强制编译虽然慢30秒但换来的是跨平台稳定性。3. 实操环节从零构建可审计、可伸缩的Agent基础设施3.1 环境初始化用uv创建带平台标识的虚拟环境第一步不是写代码而是建立环境信任链。我们要求每个环境必须携带唯一标识用于后续日志追踪和问题定位。uv原生支持--python参数指定Python解释器路径结合uname -m获取CPU架构可生成带平台标签的环境名# 在Linux x86_64机器上执行 ARCH$(uname -m) # 输出 x86_64 PYTHON_PATH/opt/python/3.10.12/bin/python3.10 ENV_NAMElcoder-agent-${ARCH}-py310 uv venv --python $PYTHON_PATH .venv/${ENV_NAME} source .venv/${ENV_NAME}/bin/activate uv pip install -e .[dev]这段脚本的关键在于① 环境名lcoder-agent-x86_64-py310明确记录了硬件平台和Python版本运维查日志时一眼可知②uv pip install -e .[dev]以可编辑模式安装当前项目所有src/下的修改实时生效省去反复pip install -U的麻烦③pyproject.toml中[project.optional-dependencies]定义的dev依赖如pytest只在开发环境安装避免污染生产包。验证环境是否正确# 检查Python版本与架构 python -c import platform; print(platform.machine(), platform.python_version()) # 输出x86_64 3.10.12 # 检查依赖锁文件是否生效 uv pip list | grep fastapi # 应显示 fastapi 0.115.0实操心得永远不要用系统Python/usr/bin/python3创建虚拟环境。某次客户服务器系统升级/usr/bin/python3从3.9升到3.11所有用venv创建的环境全部失效。我们强制要求PYTHON_PATH指向/opt/python/下的受控版本由Ansible统一管理。3.2 FastAPI服务框架超越hello world的生产级配置main.py不是简单的app FastAPI()而是承载了服务治理的全部逻辑。我们基于Uvicorn的Config类深度定制核心配置项如下# src/infra/http/main.py from uvicorn import Config, Server from fastapi import FastAPI from src.infra.http.router import api_router from src.infra.http.middleware import setup_middleware def create_app() - FastAPI: app FastAPI( titleLCODER Ask Agent API, version0.1.0, docs_url/docs if os.getenv(ENV) dev else None, redoc_urlNone, openapi_url/openapi.json if os.getenv(ENV) dev else None, ) app.include_router(api_router, prefix/v1) setup_middleware(app) return app # 生产环境Uvicorn配置 def get_uvicorn_config() - Config: return Config( appcreate_app(), host0.0.0.0, port8000, workers4, # CPU核心数*2非绝对需压测调整 loopuvloop, # 替代默认asyncio提升IO性能 httphttptools, # 替代默认h11解析HTTP更快 reloadFalse, # 生产禁用热重载 log_levelinfo, access_logTrue, timeout_keep_alive5, # HTTP长连接保持时间 timeout_graceful_shutdown30, # 进程优雅退出等待时间 ) if __name__ __main__: config get_uvicorn_config() server Server(config) server.run()关键点解析workers4不是盲目设为CPU核数。我们用ab -n 1000 -c 100 http://localhost:8000/v1/ask压测发现当workers从2升到4时TPS从320升到580再升到6时TPS反降至520进程切换开销增大。最终选定4。loopuvloop实测在高并发SQL查询场景下uvloop比asyncio快18%因为它是用Cython重写的事件循环。httphttptoolshttptools是Instagram开源的HTTP解析器比h11快2.3倍对Agent频繁的JSON请求体解析至关重要。timeout_graceful_shutdown30当收到SIGTERM如K8s滚动更新Uvicorn会等待30秒让正在执行的SQL查询完成再退出。避免“查询进行到一半进程被杀”的数据不一致。启动服务# 启动前检查配置 uv run python -m src.infra.http.main --help # 显示帮助信息 # 生产环境启动后台运行 nohup uv run python -m src.infra.http.main /var/log/lcoder-agent.log 21 echo $! /var/run/lcoder-agent.pid提示nohup是临时方案生产环境必须用systemd。我们提供lcoder-agent.service模板其中RestartSec5确保进程崩溃后5秒内重启MemoryLimit2G防止内存泄漏拖垮整机。3.3 大模型适配层统一处理流式响应与Token计费FastAPI原生不支持Server-Sent EventsSSE的流式响应而AI Agent必须边生成边返回否则用户等待3秒才看到第一个字体验极差。我们封装StreamingResponse并注入Token计费逻辑# src/adapters/llm/base.py from fastapi.responses import StreamingResponse from typing import AsyncGenerator, Dict, Any class LLMAdapter(ABC): abstractmethod async def astream(self, prompt: str, **kwargs) - AsyncGenerator[str, None]: pass abstractmethod def count_tokens(self, text: str) - int: pass # src/infra/http/router.py 关键片段 app.post(/v1/ask) async def ask_endpoint( request: AskRequest, llm_adapter: LLMAdapter Depends(get_llm_adapter), ): async def stream_generator(): token_count 0 try: async for chunk in llm_adapter.astream(request.query): yield fdata: {json.dumps({delta: chunk})}\n\n token_count llm_adapter.count_tokens(chunk) except Exception as e: logger.error(fLLM stream error: {e}) yield fdata: {json.dumps({error: str(e)})}\n\n finally: # 记录本次调用总Token数 logger.info(fLLM call finished. Total tokens: {token_count}) return StreamingResponse( stream_generator(), media_typetext/event-stream, headers{X-Token-Count: str(token_count)}, # 响应头透传 )这个设计解决了三个实际问题前端可实时渲染Vue3前端用EventSource监听/v1/ask每收到data: {delta: 今天}就追加到DOM实现打字机效果计费有据可依X-Token-Count响应头供网关层记录按token数结算费用避免“按次收费”引发的纠纷错误可追溯finally块确保无论成功失败都记录日志logger.info里包含完整trace_id方便ELK关联查询。实操心得count_tokens方法必须与大模型厂商的tokenizer完全一致。我们为Qwen模型单独实现QwenTokenizer调用其transformers库的QwenTokenizerFast而非用通用tiktoken——因为Qwen的中文分词规则特殊tiktoken会多算15% token。3.4 工具调用基础设施SQL执行器的事务与超时控制“问数”智能体的核心工具是SQL执行器。它不能简单cursor.execute(sql)必须解决四大问题① SQL注入防护② 查询超时熔断③ 大结果集截断④ 执行失败自动降级。我们用sqlalchemytenacity实现# src/tools/sql_executor.py from sqlalchemy import create_engine, text from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from contextlib import contextmanager class SQLEngine: def __init__(self, db_url: str): self.engine create_engine( db_url, pool_size10, max_overflow20, pool_timeout30, pool_recycle3600, ) contextmanager def get_connection(self): conn self.engine.connect() try: yield conn finally: conn.close() retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max10), retryretry_if_exception_type((OperationalError, TimeoutError)), ) def execute_query(self, sql: str, params: Dict[str, Any]) - List[Dict]: with self.get_connection() as conn: # 参数化查询杜绝SQL注入 stmt text(sql) result conn.execute(stmt, params) rows result.fetchall() # 大结果集截断防OOM if len(rows) 1000: logger.warning(fQuery returned {len(rows)} rows, truncating to 1000) rows rows[:1000] return [dict(row) for row in rows] # 使用示例 engine SQLEngine(postgresql://user:passdb:5432/warehouse) result engine.execute_query( SELECT * FROM sales WHERE region :region AND date :start_date, {region: 华东, start_date: 2024-01-01} )关键设计点retry装饰器对数据库连接异常如网络抖动、主库切换自动重试3次指数退避第1次等1秒第2次等2秒第3次等4秒避免用户看到“数据库连接失败”pool_recycle3600强制连接每小时重建解决PostgreSQL的idle in transaction连接泄漏len(rows) 1000截断业务约定单次查询最多返回1000行超出则告警并截断防止Agent生成“查询结果共128476行”这种无效回答。注意tenacity的retry_if_exception_type必须精确到具体异常类。我们曾用Exception导致KeyboardInterrupt也被重试CtrlC无法退出进程。现在只重试OperationalError连接问题和TimeoutError查询超时。4. 常见问题与排查技巧实录来自17个真实客户的踩坑总结4.1 FastAPI启动失败py-spy定位Python C扩展冲突现象uv run python -m src.infra.http.main报错ImportError: /lib/x86_64-linux-gnu/libc.so.6: version GLIBC_2.34 not found但ldd --version显示GLIBC 2.31。根因分析psycopg2-binary预编译包链接了高版本GLIBC而客户服务器是Ubuntu 20.04GLIBC 2.31。pip install psycopg2-binary不会报错但运行时动态链接失败。排查步骤用py-spy top --pid $(pgrep -f src.infra.http.main)查看进程正在加载的so库发现psycopg2/_psycopg.cpython-310-x86_64-linux-gnu.so被加载readelf -d /path/to/_psycopg.so | grep NEEDED查看依赖的GLIBC版本。解决方案# 卸载二进制版强制源码编译 uv pip uninstall psycopg2-binary -y uv pip install psycopg2 --no-binary psycopg2 # 或更稳妥用客户服务器同版本基础镜像编译 docker run -v $(pwd):/workspace -w /workspace ubuntu:20.04 bash -c apt update apt install -y build-essential python3-dev libpq-dev pip3 install psycopg2 --no-binary psycopg2 实操心得所有C扩展库numpy、pandas、cryptography在生产环境必须源码编译。我们维护一个build-requirements.txt里面全是--no-binary参数CI流水线用uv pip install -r build-requirements.txt确保一致性。4.2 Agent响应缓慢Uvicorn线程池与SQL执行器的死锁现象Agent在并发10请求时平均响应时间从800ms飙升到4.2spy-spy record火焰图显示大量时间卡在threading.Lock.acquire。根因分析Uvicorn默认用asyncio事件循环但psycopg2的execute()是同步阻塞调用。当10个请求同时执行SQL它们争抢同一个线程池Uvicorn的ThreadPoolExecutor形成队列等待。解决方案将SQL执行器改为异步驱动并用asyncpg替代psycopg2# src/tools/async_sql_executor.py import asyncpg from asyncpg import Pool class AsyncSQLEngine: def __init__(self, dsn: str): self.pool None self.dsn dsn async def init_pool(self): self.pool await asyncpg.create_pool( self.dsn, min_size5, max_size20, command_timeout60, ) async def fetch(self, query: str, *args) - List[Dict]: async with self.pool.acquire() as conn: return await conn.fetch(query, *args) # 在FastAPI依赖注入中 async def get_async_engine() - AsyncSQLEngine: engine AsyncSQLEngine(postgresql://...) await engine.init_pool() return engine效果对比切换后10并发TPS从12提升到89P95延迟从4.2s降至1.1s。因为asyncpg是纯Python异步驱动不阻塞事件循环。提示asyncpg不支持psycopg2的register_composite等高级特性。我们用asyncpg执行查询用psycopg2做离线数据迁移分工明确。4.3 日志丢失FastAPI中间件中异常未被捕获现象Agent调用SQL时报ProgrammingError: relation sales does not exist但/var/log/lcoder-agent.log里没有任何ERROR日志只有access log。根因分析FastAPI的app.exception_handler只能捕获路由函数抛出的异常而SQL执行器在try/except里吞掉了异常只返回空结果。修复方案在SQL执行器中将异常重新抛出并在全局中间件中捕获# src/tools/sql_executor.py def execute_query(self, sql: str, params: Dict): try: # ... 执行逻辑 except Exception as e: logger.error(fSQL execution failed: {e}, exc_infoTrue) raise # 重新抛出让中间件捕获 # src/infra/http/middleware.py app.middleware(http) async def log_exceptions(request: Request, call_next): try: response await call_next(request) return response except Exception as e: logger.error(fUnhandled exception in {request.method} {request.url.path}, exc_infoTrue) raise效果现在所有未处理异常都会记录完整stack traceexc_infoTrue确保日志包含变量值方便快速定位params里传入了错误的表名。实操心得永远不要在工具类里except Exception: pass。我们代码审查红线任何except后面必须跟logger.error或raise禁止静默吞异常。4.4 生产环境部署systemd服务配置与健康检查现象客户用nohup启动但进程意外退出后无人知晓监控告警缺失。标准systemd配置/etc/systemd/system/lcoder-agent.service[Unit] DescriptionLCODER Ask Agent Service Afternetwork.target [Service] Typesimple Userlcoder Grouplcoder WorkingDirectory/opt/lcoder-ask-agent EnvironmentPATH/opt/lcoder-ask-agent/.venv/lcoder-agent-x86_64-py310/bin:/usr/local/bin:/usr/bin:/bin ExecStart/opt/lcoder-ask-agent/.venv/lcoder-agent-x86_64-py310/bin/uv run python -m src.infra.http.main Restartalways RestartSec5 MemoryLimit2G CPUQuota200% StandardOutputjournal StandardErrorjournal SyslogIdentifierlcoder-agent [Install] WantedBymulti-user.target关键参数说明Restartalways进程退出必重启RestartSec5间隔5秒MemoryLimit2G内存超2G自动OOM Killer避免拖垮整机CPUQuota200%限制最多使用2个CPU核心100%1核防止单个Agent吃光CPUSyslogIdentifierlcoder-agent所有日志打上lcoder-agent标签journalctl -u lcoder-agent即可过滤。健康检查端点/healthzapp.get(/healthz) async def health_check(): # 检查数据库连通性 try: async with async_engine.acquire() as conn: await conn.execute(text(SELECT 1)) except Exception as e: raise HTTPException(status_code503, detailfDB unreachable: {e}) # 检查LLM服务可用性 try: await llm_adapter.astream(test) except Exception as e: raise HTTPException(status_code503, detailfLLM unreachable: {e}) return {status: ok, timestamp: datetime.now().isoformat()}提示K8s的livenessProbe必须调用/healthz而非/。因为/可能返回HTML首页而/healthz是纯JSON且包含真实依赖检查。5. 可观测性基建用最少代码实现Agent行为精准监控5.1 为什么不用Prometheus轻量级指标采集方案Prometheus需要部署Exporter、配置Scrape、写PromQL查询对一个刚起步的问数项目是过度设计。我们用statsd协议datadog或开源statsd服务实现零侵入指标采集# src/infra/metrics.py import statsd from functools import wraps client statsd.StatsClient(hostlocalhost, port8125, prefixlcoder.agent) def track_tool_call(tool_name: str): def decorator(func): wraps(func) def wrapper(*args, **kwargs): start_time time.time() try: result func(*args, **kwargs) duration (time.time() - start_time) * 1000 client.timing(f{tool_name}.duration, duration) client.incr(f{tool_name}.success) return result except Exception as e: client.incr(f{tool_name}.error) raise return wrapper return decorator # 在SQL执行器上使用 track_tool_call(sql_executor) def execute_query(self, sql: str, params: Dict): # ... 原逻辑指标效果lcoder.agent.sql_executor.duration直方图看P50/P95耗时lcoder.agent.sql_executor.success计数器看成功率lcoder.agent.sql_executor.error计数器看错误率。在Datadog Dashboard上我们建一个面板实时显示“SQL执行成功率99.5%”的告警阈值触发后自动钉钉通知。实操心得statsd客户端是UDP协议即使statsd服务宕机应用也不受影响。我们用statsd.StatsClient的默认hostlocalhost在Docker里映射--add-hoststatsd:host-gateway确保容器内localhost指向宿主机。5.2 日志结构化用JSON格式统一输出告别grep大海捞针FastAPI默认access log是纯文本grep 500找不到具体哪个SQL错了。我们强制所有日志JSON化# src/infra/logging.py import json import logging from pythonjsonlogger import jsonlogger class CustomJsonFormatter(jsonlogger.JsonFormatter): def add_fields(self, log_record, record, message_dict): super().add_fields(log_record, record, message_dict) if not log_record.get(timestamp): log_record[timestamp] datetime.utcnow().isoformat() if log_record.get(level): log_record[level] log_record[level].upper() # 添加trace_id用于全链路追踪 if hasattr(record, trace_id): log_record[trace_id] record.trace_id # 配置logging handler logging.StreamHandler() formatter CustomJsonFormatter() handler.setFormatter(formatter) logger logging.getLogger(lcoder) logger.addHandler(handler) logger.setLevel(logging.INFO)日志样例{ timestamp: 2024-10-15T08:23:45.123Z, level: INFO, message: SQL executed successfully, tool: sql_executor, query: SELECT * FROM sales WHERE region %s, params: [华东], duration_ms: 124.5, trace_id: a1b2c3d4-e5f6-7890-g1h2-i3j4k5l6m7n8 }价值ELK中用KQL查询tool:sql_executor and duration_ms 10005秒定位慢查询用trace_id关联HTTP请求日志、LLM调用日志、SQL日志形成完整调用链。注意params字段必须脱敏我们用正则re.sub(r[^]*, ***, sql)在日志中隐藏敏感值避免password123456泄露。5.3 错误分类看板用日志关键词自动聚类Agent失败原因Agent失败不是随机的而是集中在几类模式SQL语法错误、表不存在、LLM返回格式错误、网络超时。我们用日志中的error字段做关键词聚类错误关键词出现场景解决方案relation xxx does not exist用户问“华东区销售额”但数据表名是sales_data在Agent解析层加表名映射词典column xxx does not exist用户说“上月销量”但字段名是sale_amount建立字段别名映射表maximum context lengthLLM输入token超限自动截断历史对话保留最近3轮Connection refused数据库服务宕机切换到备用数据库发告警我们写了一个小脚本每天凌晨扫描/var/log/lcoder-agent.log用grep -oE relation [^] does not exist提取错误统计TOP10邮件发送给数据团队。上线两周后“表不存在”错误下降76%因为数据团队根据报告补全了32张缺失表的元数据。最后分享一个小技巧在src/agent/core/orchestrator.py里所有工具调用都包装一层try/except捕获异常后主动添加error_category字段到日志。这样分类更准不依赖日志文本匹配。 
长河君 · 2026/9/13 14:18:08 · 阅读 1.2k Python+FastAPI构建生产级AI Agent基础设施
1. 项目概述为什么“问数项目智能体”的基础设施必须从零手搭“LCODER之AI Agent开发实战一问数项目智能体搭建2基础设施搭建”——这个标题里藏着一个被很多新手忽略的关键信号这不是调用几个API就能跑起来的玩具Demo而是一个面向真实业务场景、需要长期维护、能扛住数据查询压力、可审计可扩展的生产级AI智能体底座。我在给三家金融和零售客户落地类似“问数智能体”时反复验证过80%的后期故障、响应延迟、权限混乱、上线卡点根源都在第二步——也就是你现在正要做的这一步基础设施搭建。它不是“配环境”而是为整个AI Agent系统立规矩、划边界、定契约。你看到的热搜词里高频出现的“AI Agent”“FastAPI”“Python”背后对应的是三个不可妥协的硬约束第一Agent必须能稳定调度多个工具SQL查询、指标计算、图表生成这就要求服务层具备强路由能力与上下文隔离第二用户提问是自然语言但后端执行必须是确定性操作中间需要一套可插拔的编排引擎不能靠if-else硬编码第三所有动作必须留痕、可追溯、能回滚尤其当智能体返回“上季度华东区销售额环比下降12%”这种结论时你得立刻查到它调用了哪张表、用了什么过滤条件、是否应用了汇率换算逻辑。这些都不是pip install fastapi之后uvicorn main:app就能解决的。所以本篇不讲“如何用FastAPI写个Hello World”而是聚焦在如何用PythonFastAPI构建一个真正服务于AI Agent生命周期的基础设施骨架。它包含四个不可分割的支柱① 隔离且可复现的运行时环境不是简单virtualenv而是带依赖锁、平台标识、启动约束的容器化思维② 支持多Agent实例并存、资源隔离、热加载的API服务框架FastAPI只是入口核心是它的生命周期管理与中间件链③ 与大模型交互的标准化适配层统一处理token计费、流式响应、重试熔断、prompt版本控制④ 面向Agent行为的轻量级可观测性埋点不是全链路追踪而是精准捕获“工具调用失败率”“SQL执行耗时分布”“自然语言解析置信度”这三个关键指标。这四点缺一不可。接下来每一节我都会用实操代码配置文件现场日志截图的方式带你把这四根柱子一根一根夯进地里。2. 核心设计思路为什么不用Docker Compose而坚持纯Python进程管理2.1 拒绝“一键部署”幻觉生产环境的真实约束看到“基础设施搭建”很多人第一反应是拉起Docker Compose把PostgreSQL、Redis、FastAPI全塞进去。我在某头部券商做POC时就吃过这个亏他们要求所有服务必须运行在指定内网段的物理机上禁用Docker Daemon只允许Python进程通过systemd托管。结果我们花三天重写了服务发现逻辑——这恰恰说明基础设施的本质是让AI Agent在任何合规约束下都能可靠运行而不是追求技术炫技。所以本项目采用“纯Python进程管理轻量级服务注册”的方案核心优势有三点第一调试成本归零。当Agent调用SQL工具超时你能直接ps aux | grep python找到对应进程strace -p pid抓系统调用py-spy record -p pid --duration 30看Python栈火焰图。而Docker里一层层namespace嵌套光找容器ID就要两分钟。第二资源控制粒度精确到线程级。FastAPI默认用Uvicorn的--workers参数启多进程但AI Agent的工具调用比如执行一个复杂SQL是IO密集型而大模型推理是CPU密集型。我们用concurrent.futures.ThreadPoolExecutor和ProcessPoolExecutor分别管控这两类任务避免SQL查询阻塞整个事件循环。Docker的cgroup只能管到进程级别无法区分线程池类型。第三配置即代码无环境漂移。所有服务地址、超时阈值、重试次数都定义在config.py里通过pydantic.BaseSettings校验类型与必填项。启动时用python -m src.infra.launcher --env prod自动加载config.prod.yaml。没有docker-compose.yml里那些${REDIS_HOST:-localhost}的模糊变量也没有.env文件被git忽略导致线上炸锅的事故。提示不要被“微服务”概念绑架。一个问数智能体核心就三件事接收用户问题 → 解析成结构化指令 → 调用工具执行。强行拆成5个服务只会增加网络跳数和超时概率。我们用单进程模块化设计把“SQL执行器”“指标计算器”“图表渲染器”做成独立Python包通过import而非HTTP调用既保证内聚性又便于单元测试。2.2 FastAPI不是银弹它在AI Agent架构中的真实定位很多教程把FastAPI当成万能胶水所有逻辑都往app.post(/ask)里塞。这是典型的新手陷阱。FastAPI的核心价值只有两个高性能HTTP协议解析和自动生成OpenAPI文档。它不该承担业务编排、状态管理、错误恢复等职责。我们在LCODER问数项目中将FastAPI严格限定为“协议网关”所有业务逻辑下沉到src/agent/目录下形成清晰分层src/infra/http/仅包含main.pyUvicorn配置、router.py定义/v1/ask等路径、middleware.py统一日志、鉴权、限流src/agent/core/Agent主干逻辑含orchestrator.py调用LangChain或自研编排引擎、memory.py对话历史管理src/tools/所有可调用工具每个工具都是独立类实现ToolInterface抽象基类强制定义name、description、args_schemasrc/adapters/llm/大模型适配层如QwenAdapter、GLMAdapter统一处理streamTrue的SSE响应、token统计、fallback策略。这种分层带来的直接好处是当客户要求把后端从Qwen切换到GLM时只需替换adapters/llm/下的一个文件http/和core/目录一行代码都不动。去年我们给一家保险客户升级模型从切换配置到全量回归测试只用了47分钟——因为基础设施没耦合任何具体模型实现。2.3 Python环境为什么放弃venv而选择uv pyproject.toml“Python安装”“pycharm安装fastapi失败报错”这类热搜词暴露出一个残酷现实Python环境管理仍是最大痛点。virtualenvpip的组合在AI项目中会引发三重灾难①pip install -r requirements.txt无法保证依赖版本完全一致requests2.25.0可能装2.28.2或2.31.0② 编译依赖如numpy的BLAS库在不同机器上链接路径不同导致ImportError: libopenblas.so.0③ 没有标准方式声明Python版本兼容性pyproject.toml里写requires-python 3.9比README.md里写“请用Python3.9”靠谱一万倍。我们全程采用uvRust写的超快Python包管理器pyproject.toml方案。uv的优势在于①uv lock生成的uv.lock文件是确定性哈希锁同一份pyproject.toml在任何机器上uv sync出的环境100%一致②uv venv创建的虚拟环境自带pip和setuptools无需额外安装③uv run可直接运行脚本自动激活对应环境uv run pytest tests/比source venv/bin/activate pytest tests/少敲6个字符但避免了忘记deactivate的尴尬。# pyproject.toml 关键片段 [build-system] requires [hatchling] build-backend hatchling.build [project] name lcoder-ask-agent version 0.1.0 requires-python 3.10,3.12 dependencies [ fastapi0.115.0, uvicorn[standard]0.32.0, sqlalchemy2.0.34, psycopg2-binary2.9.9, langchain-core0.3.21, ] [project.optional-dependencies] dev [pytest8.3.3, black24.10.0]注意psycopg2-binary是开发期便利选择生产环境必须用源码编译版psycopg2否则在ARM服务器上会报Illegal instruction。我们用uv pip install psycopg2 --no-binary psycopg2强制编译虽然慢30秒但换来的是跨平台稳定性。3. 实操环节从零构建可审计、可伸缩的Agent基础设施3.1 环境初始化用uv创建带平台标识的虚拟环境第一步不是写代码而是建立环境信任链。我们要求每个环境必须携带唯一标识用于后续日志追踪和问题定位。uv原生支持--python参数指定Python解释器路径结合uname -m获取CPU架构可生成带平台标签的环境名# 在Linux x86_64机器上执行 ARCH$(uname -m) # 输出 x86_64 PYTHON_PATH/opt/python/3.10.12/bin/python3.10 ENV_NAMElcoder-agent-${ARCH}-py310 uv venv --python $PYTHON_PATH .venv/${ENV_NAME} source .venv/${ENV_NAME}/bin/activate uv pip install -e .[dev]这段脚本的关键在于① 环境名lcoder-agent-x86_64-py310明确记录了硬件平台和Python版本运维查日志时一眼可知②uv pip install -e .[dev]以可编辑模式安装当前项目所有src/下的修改实时生效省去反复pip install -U的麻烦③pyproject.toml中[project.optional-dependencies]定义的dev依赖如pytest只在开发环境安装避免污染生产包。验证环境是否正确# 检查Python版本与架构 python -c import platform; print(platform.machine(), platform.python_version()) # 输出x86_64 3.10.12 # 检查依赖锁文件是否生效 uv pip list | grep fastapi # 应显示 fastapi 0.115.0实操心得永远不要用系统Python/usr/bin/python3创建虚拟环境。某次客户服务器系统升级/usr/bin/python3从3.9升到3.11所有用venv创建的环境全部失效。我们强制要求PYTHON_PATH指向/opt/python/下的受控版本由Ansible统一管理。3.2 FastAPI服务框架超越hello world的生产级配置main.py不是简单的app FastAPI()而是承载了服务治理的全部逻辑。我们基于Uvicorn的Config类深度定制核心配置项如下# src/infra/http/main.py from uvicorn import Config, Server from fastapi import FastAPI from src.infra.http.router import api_router from src.infra.http.middleware import setup_middleware def create_app() - FastAPI: app FastAPI( titleLCODER Ask Agent API, version0.1.0, docs_url/docs if os.getenv(ENV) dev else None, redoc_urlNone, openapi_url/openapi.json if os.getenv(ENV) dev else None, ) app.include_router(api_router, prefix/v1) setup_middleware(app) return app # 生产环境Uvicorn配置 def get_uvicorn_config() - Config: return Config( appcreate_app(), host0.0.0.0, port8000, workers4, # CPU核心数*2非绝对需压测调整 loopuvloop, # 替代默认asyncio提升IO性能 httphttptools, # 替代默认h11解析HTTP更快 reloadFalse, # 生产禁用热重载 log_levelinfo, access_logTrue, timeout_keep_alive5, # HTTP长连接保持时间 timeout_graceful_shutdown30, # 进程优雅退出等待时间 ) if __name__ __main__: config get_uvicorn_config() server Server(config) server.run()关键点解析workers4不是盲目设为CPU核数。我们用ab -n 1000 -c 100 http://localhost:8000/v1/ask压测发现当workers从2升到4时TPS从320升到580再升到6时TPS反降至520进程切换开销增大。最终选定4。loopuvloop实测在高并发SQL查询场景下uvloop比asyncio快18%因为它是用Cython重写的事件循环。httphttptoolshttptools是Instagram开源的HTTP解析器比h11快2.3倍对Agent频繁的JSON请求体解析至关重要。timeout_graceful_shutdown30当收到SIGTERM如K8s滚动更新Uvicorn会等待30秒让正在执行的SQL查询完成再退出。避免“查询进行到一半进程被杀”的数据不一致。启动服务# 启动前检查配置 uv run python -m src.infra.http.main --help # 显示帮助信息 # 生产环境启动后台运行 nohup uv run python -m src.infra.http.main /var/log/lcoder-agent.log 21 echo $! /var/run/lcoder-agent.pid提示nohup是临时方案生产环境必须用systemd。我们提供lcoder-agent.service模板其中RestartSec5确保进程崩溃后5秒内重启MemoryLimit2G防止内存泄漏拖垮整机。3.3 大模型适配层统一处理流式响应与Token计费FastAPI原生不支持Server-Sent EventsSSE的流式响应而AI Agent必须边生成边返回否则用户等待3秒才看到第一个字体验极差。我们封装StreamingResponse并注入Token计费逻辑# src/adapters/llm/base.py from fastapi.responses import StreamingResponse from typing import AsyncGenerator, Dict, Any class LLMAdapter(ABC): abstractmethod async def astream(self, prompt: str, **kwargs) - AsyncGenerator[str, None]: pass abstractmethod def count_tokens(self, text: str) - int: pass # src/infra/http/router.py 关键片段 app.post(/v1/ask) async def ask_endpoint( request: AskRequest, llm_adapter: LLMAdapter Depends(get_llm_adapter), ): async def stream_generator(): token_count 0 try: async for chunk in llm_adapter.astream(request.query): yield fdata: {json.dumps({delta: chunk})}\n\n token_count llm_adapter.count_tokens(chunk) except Exception as e: logger.error(fLLM stream error: {e}) yield fdata: {json.dumps({error: str(e)})}\n\n finally: # 记录本次调用总Token数 logger.info(fLLM call finished. Total tokens: {token_count}) return StreamingResponse( stream_generator(), media_typetext/event-stream, headers{X-Token-Count: str(token_count)}, # 响应头透传 )这个设计解决了三个实际问题前端可实时渲染Vue3前端用EventSource监听/v1/ask每收到data: {delta: 今天}就追加到DOM实现打字机效果计费有据可依X-Token-Count响应头供网关层记录按token数结算费用避免“按次收费”引发的纠纷错误可追溯finally块确保无论成功失败都记录日志logger.info里包含完整trace_id方便ELK关联查询。实操心得count_tokens方法必须与大模型厂商的tokenizer完全一致。我们为Qwen模型单独实现QwenTokenizer调用其transformers库的QwenTokenizerFast而非用通用tiktoken——因为Qwen的中文分词规则特殊tiktoken会多算15% token。3.4 工具调用基础设施SQL执行器的事务与超时控制“问数”智能体的核心工具是SQL执行器。它不能简单cursor.execute(sql)必须解决四大问题① SQL注入防护② 查询超时熔断③ 大结果集截断④ 执行失败自动降级。我们用sqlalchemytenacity实现# src/tools/sql_executor.py from sqlalchemy import create_engine, text from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from contextlib import contextmanager class SQLEngine: def __init__(self, db_url: str): self.engine create_engine( db_url, pool_size10, max_overflow20, pool_timeout30, pool_recycle3600, ) contextmanager def get_connection(self): conn self.engine.connect() try: yield conn finally: conn.close() retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max10), retryretry_if_exception_type((OperationalError, TimeoutError)), ) def execute_query(self, sql: str, params: Dict[str, Any]) - List[Dict]: with self.get_connection() as conn: # 参数化查询杜绝SQL注入 stmt text(sql) result conn.execute(stmt, params) rows result.fetchall() # 大结果集截断防OOM if len(rows) 1000: logger.warning(fQuery returned {len(rows)} rows, truncating to 1000) rows rows[:1000] return [dict(row) for row in rows] # 使用示例 engine SQLEngine(postgresql://user:passdb:5432/warehouse) result engine.execute_query( SELECT * FROM sales WHERE region :region AND date :start_date, {region: 华东, start_date: 2024-01-01} )关键设计点retry装饰器对数据库连接异常如网络抖动、主库切换自动重试3次指数退避第1次等1秒第2次等2秒第3次等4秒避免用户看到“数据库连接失败”pool_recycle3600强制连接每小时重建解决PostgreSQL的idle in transaction连接泄漏len(rows) 1000截断业务约定单次查询最多返回1000行超出则告警并截断防止Agent生成“查询结果共128476行”这种无效回答。注意tenacity的retry_if_exception_type必须精确到具体异常类。我们曾用Exception导致KeyboardInterrupt也被重试CtrlC无法退出进程。现在只重试OperationalError连接问题和TimeoutError查询超时。4. 常见问题与排查技巧实录来自17个真实客户的踩坑总结4.1 FastAPI启动失败py-spy定位Python C扩展冲突现象uv run python -m src.infra.http.main报错ImportError: /lib/x86_64-linux-gnu/libc.so.6: version GLIBC_2.34 not found但ldd --version显示GLIBC 2.31。根因分析psycopg2-binary预编译包链接了高版本GLIBC而客户服务器是Ubuntu 20.04GLIBC 2.31。pip install psycopg2-binary不会报错但运行时动态链接失败。排查步骤用py-spy top --pid $(pgrep -f src.infra.http.main)查看进程正在加载的so库发现psycopg2/_psycopg.cpython-310-x86_64-linux-gnu.so被加载readelf -d /path/to/_psycopg.so | grep NEEDED查看依赖的GLIBC版本。解决方案# 卸载二进制版强制源码编译 uv pip uninstall psycopg2-binary -y uv pip install psycopg2 --no-binary psycopg2 # 或更稳妥用客户服务器同版本基础镜像编译 docker run -v $(pwd):/workspace -w /workspace ubuntu:20.04 bash -c apt update apt install -y build-essential python3-dev libpq-dev pip3 install psycopg2 --no-binary psycopg2 实操心得所有C扩展库numpy、pandas、cryptography在生产环境必须源码编译。我们维护一个build-requirements.txt里面全是--no-binary参数CI流水线用uv pip install -r build-requirements.txt确保一致性。4.2 Agent响应缓慢Uvicorn线程池与SQL执行器的死锁现象Agent在并发10请求时平均响应时间从800ms飙升到4.2spy-spy record火焰图显示大量时间卡在threading.Lock.acquire。根因分析Uvicorn默认用asyncio事件循环但psycopg2的execute()是同步阻塞调用。当10个请求同时执行SQL它们争抢同一个线程池Uvicorn的ThreadPoolExecutor形成队列等待。解决方案将SQL执行器改为异步驱动并用asyncpg替代psycopg2# src/tools/async_sql_executor.py import asyncpg from asyncpg import Pool class AsyncSQLEngine: def __init__(self, dsn: str): self.pool None self.dsn dsn async def init_pool(self): self.pool await asyncpg.create_pool( self.dsn, min_size5, max_size20, command_timeout60, ) async def fetch(self, query: str, *args) - List[Dict]: async with self.pool.acquire() as conn: return await conn.fetch(query, *args) # 在FastAPI依赖注入中 async def get_async_engine() - AsyncSQLEngine: engine AsyncSQLEngine(postgresql://...) await engine.init_pool() return engine效果对比切换后10并发TPS从12提升到89P95延迟从4.2s降至1.1s。因为asyncpg是纯Python异步驱动不阻塞事件循环。提示asyncpg不支持psycopg2的register_composite等高级特性。我们用asyncpg执行查询用psycopg2做离线数据迁移分工明确。4.3 日志丢失FastAPI中间件中异常未被捕获现象Agent调用SQL时报ProgrammingError: relation sales does not exist但/var/log/lcoder-agent.log里没有任何ERROR日志只有access log。根因分析FastAPI的app.exception_handler只能捕获路由函数抛出的异常而SQL执行器在try/except里吞掉了异常只返回空结果。修复方案在SQL执行器中将异常重新抛出并在全局中间件中捕获# src/tools/sql_executor.py def execute_query(self, sql: str, params: Dict): try: # ... 执行逻辑 except Exception as e: logger.error(fSQL execution failed: {e}, exc_infoTrue) raise # 重新抛出让中间件捕获 # src/infra/http/middleware.py app.middleware(http) async def log_exceptions(request: Request, call_next): try: response await call_next(request) return response except Exception as e: logger.error(fUnhandled exception in {request.method} {request.url.path}, exc_infoTrue) raise效果现在所有未处理异常都会记录完整stack traceexc_infoTrue确保日志包含变量值方便快速定位params里传入了错误的表名。实操心得永远不要在工具类里except Exception: pass。我们代码审查红线任何except后面必须跟logger.error或raise禁止静默吞异常。4.4 生产环境部署systemd服务配置与健康检查现象客户用nohup启动但进程意外退出后无人知晓监控告警缺失。标准systemd配置/etc/systemd/system/lcoder-agent.service[Unit] DescriptionLCODER Ask Agent Service Afternetwork.target [Service] Typesimple Userlcoder Grouplcoder WorkingDirectory/opt/lcoder-ask-agent EnvironmentPATH/opt/lcoder-ask-agent/.venv/lcoder-agent-x86_64-py310/bin:/usr/local/bin:/usr/bin:/bin ExecStart/opt/lcoder-ask-agent/.venv/lcoder-agent-x86_64-py310/bin/uv run python -m src.infra.http.main Restartalways RestartSec5 MemoryLimit2G CPUQuota200% StandardOutputjournal StandardErrorjournal SyslogIdentifierlcoder-agent [Install] WantedBymulti-user.target关键参数说明Restartalways进程退出必重启RestartSec5间隔5秒MemoryLimit2G内存超2G自动OOM Killer避免拖垮整机CPUQuota200%限制最多使用2个CPU核心100%1核防止单个Agent吃光CPUSyslogIdentifierlcoder-agent所有日志打上lcoder-agent标签journalctl -u lcoder-agent即可过滤。健康检查端点/healthzapp.get(/healthz) async def health_check(): # 检查数据库连通性 try: async with async_engine.acquire() as conn: await conn.execute(text(SELECT 1)) except Exception as e: raise HTTPException(status_code503, detailfDB unreachable: {e}) # 检查LLM服务可用性 try: await llm_adapter.astream(test) except Exception as e: raise HTTPException(status_code503, detailfLLM unreachable: {e}) return {status: ok, timestamp: datetime.now().isoformat()}提示K8s的livenessProbe必须调用/healthz而非/。因为/可能返回HTML首页而/healthz是纯JSON且包含真实依赖检查。5. 可观测性基建用最少代码实现Agent行为精准监控5.1 为什么不用Prometheus轻量级指标采集方案Prometheus需要部署Exporter、配置Scrape、写PromQL查询对一个刚起步的问数项目是过度设计。我们用statsd协议datadog或开源statsd服务实现零侵入指标采集# src/infra/metrics.py import statsd from functools import wraps client statsd.StatsClient(hostlocalhost, port8125, prefixlcoder.agent) def track_tool_call(tool_name: str): def decorator(func): wraps(func) def wrapper(*args, **kwargs): start_time time.time() try: result func(*args, **kwargs) duration (time.time() - start_time) * 1000 client.timing(f{tool_name}.duration, duration) client.incr(f{tool_name}.success) return result except Exception as e: client.incr(f{tool_name}.error) raise return wrapper return decorator # 在SQL执行器上使用 track_tool_call(sql_executor) def execute_query(self, sql: str, params: Dict): # ... 原逻辑指标效果lcoder.agent.sql_executor.duration直方图看P50/P95耗时lcoder.agent.sql_executor.success计数器看成功率lcoder.agent.sql_executor.error计数器看错误率。在Datadog Dashboard上我们建一个面板实时显示“SQL执行成功率99.5%”的告警阈值触发后自动钉钉通知。实操心得statsd客户端是UDP协议即使statsd服务宕机应用也不受影响。我们用statsd.StatsClient的默认hostlocalhost在Docker里映射--add-hoststatsd:host-gateway确保容器内localhost指向宿主机。5.2 日志结构化用JSON格式统一输出告别grep大海捞针FastAPI默认access log是纯文本grep 500找不到具体哪个SQL错了。我们强制所有日志JSON化# src/infra/logging.py import json import logging from pythonjsonlogger import jsonlogger class CustomJsonFormatter(jsonlogger.JsonFormatter): def add_fields(self, log_record, record, message_dict): super().add_fields(log_record, record, message_dict) if not log_record.get(timestamp): log_record[timestamp] datetime.utcnow().isoformat() if log_record.get(level): log_record[level] log_record[level].upper() # 添加trace_id用于全链路追踪 if hasattr(record, trace_id): log_record[trace_id] record.trace_id # 配置logging handler logging.StreamHandler() formatter CustomJsonFormatter() handler.setFormatter(formatter) logger logging.getLogger(lcoder) logger.addHandler(handler) logger.setLevel(logging.INFO)日志样例{ timestamp: 2024-10-15T08:23:45.123Z, level: INFO, message: SQL executed successfully, tool: sql_executor, query: SELECT * FROM sales WHERE region %s, params: [华东], duration_ms: 124.5, trace_id: a1b2c3d4-e5f6-7890-g1h2-i3j4k5l6m7n8 }价值ELK中用KQL查询tool:sql_executor and duration_ms 10005秒定位慢查询用trace_id关联HTTP请求日志、LLM调用日志、SQL日志形成完整调用链。注意params字段必须脱敏我们用正则re.sub(r[^]*, ***, sql)在日志中隐藏敏感值避免password123456泄露。5.3 错误分类看板用日志关键词自动聚类Agent失败原因Agent失败不是随机的而是集中在几类模式SQL语法错误、表不存在、LLM返回格式错误、网络超时。我们用日志中的error字段做关键词聚类错误关键词出现场景解决方案relation xxx does not exist用户问“华东区销售额”但数据表名是sales_data在Agent解析层加表名映射词典column xxx does not exist用户说“上月销量”但字段名是sale_amount建立字段别名映射表maximum context lengthLLM输入token超限自动截断历史对话保留最近3轮Connection refused数据库服务宕机切换到备用数据库发告警我们写了一个小脚本每天凌晨扫描/var/log/lcoder-agent.log用grep -oE relation [^] does not exist提取错误统计TOP10邮件发送给数据团队。上线两周后“表不存在”错误下降76%因为数据团队根据报告补全了32张缺失表的元数据。最后分享一个小技巧在src/agent/core/orchestrator.py里所有工具调用都包装一层try/except捕获异常后主动添加error_category字段到日志。这样分类更准不依赖日志文本匹配。