企业官网建设流程全解析

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”，而是聚焦在：如何用Python+FastAPI构建一个真正服务于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.py（Uvicorn配置）、router.py（定义/v1/ask等路径）、middleware.py（统一日志、鉴权、限流）；
• src/agent/core/：Agent主干逻辑，含orchestrator.py（调用LangChain或自研编排引擎）、memory.py（对话历史管理）；
• src/tools/：所有可调用工具，每个工具都是独立类，实现ToolInterface抽象基类，强制定义name、description、args_schema；
• src/adapters/llm/：大模型适配层，如QwenAdapter、GLMAdapter，统一处理stream=True的SSE响应、token统计、fallback策略。

这种分层带来的直接好处是：当客户要求把后端从Qwen切换到GLM时，只需替换adapters/llm/下的一个文件，http/和core/目录一行代码都不动。去年我们给一家保险客户升级模型，从切换配置到全量回归测试，只用了47分钟——因为基础设施没耦合任何具体模型实现。

2.3 Python环境：为什么放弃venv而选择uv + pyproject.toml

“Python安装”“pycharm安装fastapi失败报错”这类热搜词，暴露出一个残酷现实：Python环境管理仍是最大痛点。virtualenv+pip的组合，在AI项目中会引发三重灾难：①pip install -r requirements.txt无法保证依赖版本完全一致（requests>=2.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”靠谱一万倍。

我们全程采用uv（Rust写的超快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 = [ "fastapi==0.115.0", "uvicorn[standard]==0.32.0", "sqlalchemy==2.0.34", "psycopg2-binary==2.9.9", "langchain-core==0.3.21", ] [project.optional-dependencies] dev = ["pytest==8.3.3", "black==24.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_NAME="lcoder-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( title="LCODER Ask Agent API", version="0.1.0", docs_url="/docs" if os.getenv("ENV") == "dev" else None, redoc_url=None, 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( app=create_app(), host="0.0.0.0", port=8000, workers=4, # CPU核心数*2，非绝对，需压测调整 loop="uvloop", # 替代默认asyncio，提升IO性能 http="httptools", # 替代默认h11，解析HTTP更快 reload=False, # 生产禁用热重载 log_level="info", access_log=True, timeout_keep_alive=5, # HTTP长连接保持时间 timeout_graceful_shutdown=30, # 进程优雅退出等待时间 ) if __name__ == "__main__": config = get_uvicorn_config() server = Server(config) server.run()

关键点解析：

• workers=4：不是盲目设为CPU核数。我们用ab -n 1000 -c 100 http://localhost:8000/v1/ask压测，发现当workers从2升到4时，TPS从320升到580；再升到6时TPS反降至520（进程切换开销增大）。最终选定4。
• loop="uvloop"：实测在高并发SQL查询场景下，uvloop比asyncio快18%，因为它是用Cython重写的事件循环。
• http="httptools"：httptools是Instagram开源的HTTP解析器，比h11快2.3倍，对Agent频繁的JSON请求体解析至关重要。
• timeout_graceful_shutdown=30：当收到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 2>&1 & echo $! > /var/run/lcoder-agent.pid

提示：nohup+&是临时方案，生产环境必须用systemd。我们提供lcoder-agent.service模板，其中RestartSec=5确保进程崩溃后5秒内重启，MemoryLimit=2G防止内存泄漏拖垮整机。

3.3 大模型适配层：统一处理流式响应与Token计费

FastAPI原生不支持Server-Sent Events（SSE）的流式响应，而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 f"data: {json.dumps({'delta': chunk})}\n\n" token_count += llm_adapter.count_tokens(chunk) except Exception as e: logger.error(f"LLM stream error: {e}") yield f"data: {json.dumps({'error': str(e)})}\n\n" finally: # 记录本次调用总Token数 logger.info(f"LLM call finished. Total tokens: {token_count}") return StreamingResponse( stream_generator(), media_type="text/event-stream", headers={"X-Token-Count": str(token_count)}, # 响应头透传 )

这个设计解决了三个实际问题：

1. 前端可实时渲染：Vue3前端用EventSource监听/v1/ask，每收到data: {"delta": "今天"}就追加到DOM，实现打字机效果；
2. 计费有据可依：X-Token-Count响应头供网关层记录，按token数结算费用，避免“按次收费”引发的纠纷；
3. 错误可追溯：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注入防护；② 查询超时熔断；③ 大结果集截断；④ 执行失败自动降级。我们用sqlalchemy+tenacity实现：

# 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_size=10, max_overflow=20, pool_timeout=30, pool_recycle=3600, ) @contextmanager def get_connection(self): conn = self.engine.connect() try: yield conn finally: conn.close() @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), retry=retry_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(f"Query returned {len(rows)} rows, truncating to 1000") rows = rows[:1000] return [dict(row) for row in rows] # 使用示例 engine = SQLEngine("postgresql://user:pass@db: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_recycle=3600：强制连接每小时重建，解决PostgreSQL的idle in transaction连接泄漏；
• len(rows) > 1000截断：业务约定单次查询最多返回1000行，超出则告警并截断，防止Agent生成“查询结果共128476行”这种无效回答。

注意：tenacity的retry_if_exception_type必须精确到具体异常类。我们曾用Exception导致KeyboardInterrupt也被重试，Ctrl+C无法退出进程。现在只重试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.04（GLIBC 2.31）。pip install psycopg2-binary不会报错，但运行时动态链接失败。

排查步骤：

1. 用py-spy top --pid $(pgrep -f "src.infra.http.main")查看进程正在加载的so库；
2. 发现psycopg2/_psycopg.cpython-310-x86_64-linux-gnu.so被加载；
3. 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.2s，py-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_size=5, max_size=20, command_timeout=60, ) 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提升到89，P95延迟从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(f"SQL execution failed: {e}", exc_info=True) 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(f"Unhandled exception in {request.method} {request.url.path}", exc_info=True) raise

效果：现在所有未处理异常都会记录完整stack trace，exc_info=True确保日志包含变量值，方便快速定位params里传入了错误的表名。

实操心得：永远不要在工具类里except Exception: pass。我们代码审查红线：任何except后面必须跟logger.error或raise，禁止静默吞异常。

4.4 生产环境部署：systemd服务配置与健康检查

现象：客户用nohup启动，但进程意外退出后无人知晓，监控告警缺失。

标准systemd配置（/etc/systemd/system/lcoder-agent.service）：

[Unit] Description=LCODER Ask Agent Service After=network.target [Service] Type=simple User=lcoder Group=lcoder WorkingDirectory=/opt/lcoder-ask-agent Environment="PATH=/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 Restart=always RestartSec=5 MemoryLimit=2G CPUQuota=200% StandardOutput=journal StandardError=journal SyslogIdentifier=lcoder-agent [Install] WantedBy=multi-user.target

关键参数说明：

• Restart=always：进程退出必重启，RestartSec=5间隔5秒；
• MemoryLimit=2G：内存超2G自动OOM Killer，避免拖垮整机；
• CPUQuota=200%：限制最多使用2个CPU核心（100%=1核），防止单个Agent吃光CPU；
• SyslogIdentifier=lcoder-agent：所有日志打上lcoder-agent标签，journalctl -u lcoder-agent即可过滤。

健康检查端点（/healthz）：

@app.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_code=503, detail=f"DB unreachable: {e}") # 检查LLM服务可用性 try: await llm_adapter.astream("test") except Exception as e: raise HTTPException(status_code=503, detail=f"LLM 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(host="localhost", port=8125, prefix="lcoder.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的默认host="localhost"，在Docker里映射--add-host=statsd: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 > 1000，5秒定位慢查询；用trace_id关联HTTP请求日志、LLM调用日志、SQL日志，形成完整调用链。

注意：params字段必须脱敏！我们用正则re.sub(r"'[^']*'", "'***'", sql)在日志中隐藏敏感值，避免password='123456'泄露。

5.3 错误分类看板：用日志关键词自动聚类Agent失败原因

Agent失败不是随机的，而是集中在几类模式：SQL语法错误、表不存在、LLM返回格式错误、网络超时。我们用日志中的error字段做关键词聚类：

我们写了一个小脚本，每天凌晨扫描/var/log/lcoder-agent.log，用grep -oE 'relation "[^"]+" does not exist'提取错误，统计TOP10，邮件发送给数据团队。上线两周后，“表不存在”错误下降76%，因为数据团队根据报告补全了32张缺失表的元数据。

最后分享一个小技巧：在src/agent/core/orchestrator.py里，所有工具调用都包装一层try/except，捕获异常后主动添加error_category字段到日志。这样分类更准，不依赖日志文本匹配。