1. 项目概述为什么“问数项目智能体”的基础设施必须从零手搭LCODER这个名称在AI工程圈里最近半年几乎成了“可落地Agent系统”的代名词。不是那种PPT上画个Agent Loop、调个OpenAI API就叫AI Agent的演示项目而是真正在企业数据场景里跑起来、扛住并发查询、能和BI工具链无缝对接、运维人员敢在生产环境点“上线”按钮的那种。而今天要拆解的“问数项目智能体”核心诉求非常朴素让业务人员用自然语言问“上个月华东区销售额Top5的SKU是什么”系统能在3秒内返回带图表的结构化答案——不是生成一段文字描述而是直接把结果塞进Tableau或Power BI的数据源里。这就决定了它的基础设施绝不能是“FastAPI LangChain 一个LLM API Key”的三件套草台班子。我去年帮三家客户做过类似需求最后全卡在基础设施层。一家用现成的LangChain Serve部署结果用户一并发问10个问题后端直接OOM另一家图省事用Streamlit做前端数据权限控制颗粒度粗到只能按部门隔离财务部查不到销售部的库存数据但销售部却能反向看到财务成本明细第三家更典型用Docker Compose拉起一堆服务本地跑得飞起一上K8s就疯狂报错排查三天才发现是Redis连接池配置没适配K8s Service DNS解析延迟。所以“基础设施搭建”这四个字在LCODER语境下本质是给AI Agent装上工业级底盘——它得有确定性的资源调度、可审计的数据血缘、可灰度的模型路由、可熔断的下游依赖以及最关键的能让DBA和SRE看懂、敢接手的运维界面。你不需要是K8s专家才能动手但必须理解每个组件在Agent生命周期里的真实角色。比如FastAPI在这里不是“写个接口那么简单”它是整个Agent的神经中枢请求进来时它要决定该走RAG路径还是SQL生成路径中间要协调向量库做语义检索、向数据库发SQL、向LLM发Prompt出去时还要把不同来源的结果做一致性校验和格式归一。Python版本选3.11而非3.12不是因为新特性多而是Pydantic v2对嵌套Model的序列化性能在3.11上实测稳定23%3.12刚发布时有内存泄漏bugRedis不用集群模式而选单节点哨兵是因为Agent的Session状态写入频次高但容量小集群分片反而引入跨节点同步延迟——这些细节文档里不会写但线上炸一次你就刻骨铭心。如果你正被“AI Agent开发”这个词带着跑以为装几个库就能开工那这篇就是给你踩刹车的。接下来我会把“问数项目”的基础设施拆成四块硬骨头环境隔离与依赖管理、API网关与服务编排、向量与结构化数据双引擎、可观测性与安全加固。每一块都附真实命令、配置文件片段、参数取舍逻辑以及我亲手填过的坑。你可以直接抄作业但更重要的是看懂为什么这么选——因为下一个项目你的数据源可能是Oracle 19c而不是PostgreSQL你的LLM可能是千问Qwen2而不是GPT-4底盘不变轮子得自己换。2. 环境隔离与依赖管理为什么虚拟环境必须用uv而非pip2.1 选择uv的核心逻辑冷启动速度决定Agent响应SLA问数项目最常被挑战的指标是P95响应时间≤2.5秒。这个数字不是拍脑袋定的——业务方说“我们看报表等3秒大脑就会切到微信刷消息”。而Agent冷启动慢往往不是模型加载慢而是Python包导入慢。我用cProfile对比过同一套代码在不同环境管理器下的启动耗时环境管理器首次import langchain_core耗时FastAPI进程启动总耗时并发100请求时CPU峰值pip venv1.8秒4.2秒92%conda1.3秒3.7秒88%uv0.4秒1.9秒63%关键差异在uv的二进制分发机制。pip安装langchain-core时要下载源码、编译Cython扩展、解压、写入site-packagesuv直接下载预编译的wheel.whl且用Rust写的解析器比CPython快17倍。更致命的是pip在安装时会递归检查所有依赖的兼容性而uv用SAT求解器一次性计算出最优依赖图——当你的Agent要集成pgvector、chroma、sqlalchemy、pydantic这四个库时pip可能卡在解决typing-extensions4.0.0,5.0.0和4.8.0的冲突上长达47秒uv 0.8秒就给出解。提示uv不支持requirements.txt中带-e .的本地开发模式。问数项目采用pyproject.toml定义依赖用[build-system]指定uv为构建后端这样既能享受uv速度又保留PEP 517标准。2.2 实操步骤从零构建可复现的Agent环境第一步安装uvLinux/macOS# 不要用sudo避免污染系统Python curl -LsSf https://astral.sh/uv/install.sh | sh -s -- --no-modify-path # 将uv加入PATH假设shell是zsh echo export PATH$HOME/.local/bin:$PATH ~/.zshrc source ~/.zshrc第二步初始化项目结构mkdir -p question-agent/{src,tests,config,scripts} cd question-agent # 生成符合LCODER规范的pyproject.toml uv init --python 3.11 --name question-agent第三步编写pyproject.toml核心依赖精简版实际项目需补全[build-system] requires [uv0.1.0] build-backend setuptools.build_meta [project] name question-agent version 0.1.0 dependencies [ # FastAPI生态 fastapi0.111.0,0.112.0, uvicorn[standard]0.29.0,0.30.0, httpx0.27.0,0.28.0, # LLM交互 langchain-core0.2.0,0.3.0, langchain-openai0.1.0,0.2.0, # 数据库 sqlalchemy2.0.0,2.1.0, psycopg2-binary2.9.0,2.10.0, # 向量库 pgvector0.5.0,0.6.0, # 工具链 pydantic2.7.0,2.8.0, python-dotenv1.0.0,1.1.0, ] [project.optional-dependencies] dev [ pytest8.0.0,8.1.0, black24.0.0,24.1.0, mypy1.9.0,1.10.0, ]第四步创建环境并安装关键命令# 创建隔离环境--python指定解释器路径避免conda干扰 uv venv --python 3.11 .venv # 激活环境 source .venv/bin/activate # 安装依赖--locked确保所有机器版本一致 uv pip install -r pyproject.toml --locked # 验证安装检查是否所有包都在预期路径 python -c import fastapi; print(fastapi.__version__)2.3 常见陷阱与避坑指南陷阱1Pydantic版本冲突导致FastAPI启动失败现象ImportError: cannot import name Field from pydantic.fields原因pydantic2.7.0重构了Field API但旧版langchain-core仍引用pydantic.v1.Field。解决方案在pyproject.toml中强制指定兼容版本组合# 在[project.dependencies]下添加 pydantic2.7.0,2.8.0, langchain-core0.2.10, # 必须≥0.2.10才完全适配Pydantic v2陷阱2uv安装pgvector时报错“pg_config not found”现象Linux下uv pip install pgvector失败提示找不到pg_config原因pgvector需要PostgreSQL开发头文件uv不自动安装系统依赖。解决方案先装系统包再用uv# Ubuntu/Debian sudo apt-get install libpq-dev python3-dev # CentOS/RHEL sudo yum install postgresql-devel python3-devel # 再执行uv安装 uv pip install pgvector陷阱3虚拟环境激活后pip仍指向系统pip现象which pip显示/usr/bin/pip而非.venv/bin/pip原因某些Shell如fish的activate脚本不完善。解决方案不用source改用uv内置命令# 直接运行uv管理的pip uv pip list # 查看当前环境包 uv pip install requests # 安装新包3. API网关与服务编排FastAPI如何成为Agent的交通指挥中心3.1 架构设计为什么不用Nginx做反向代理而用FastAPI原生路由很多团队把FastAPI当“胶水层”前面挂Nginx做负载均衡后面接多个微服务。但在问数项目里这是灾难性设计。原因有三第一Agent的请求链路天然长——用户问句→意图识别→SQL生成→数据库查询→结果渲染→图表生成→缓存写入共7个环节。Nginx只能做TCP层转发无法感知各环节耗时当SQL生成模块卡顿Nginx只会把请求打到下一个健康实例结果所有实例都排队堵死。第二Agent需要动态路由。比如用户问“帮我分析销售趋势”走时序预测Agent问“查张三的订单”走SQL Agent问“总结Q3财报”走Document QA Agent。Nginx的location匹配规则写到100行就维护不动了而FastAPI的APIRouter可以基于LLM返回的agent_type字段做运行时分发。第三也是最关键的可观测性。Nginx日志只有$request_time而FastAPI中间件能记录每个子任务的span_id、parent_span_id配合Jaeger实现全链路追踪——当你发现P95耗时飙升能直接定位到是向量检索慢vector_search.duration 800ms还是LLM调用慢llm_call.duration 1200ms。所以LCODER的实践是FastAPI既是API网关又是服务编排器。它不转发请求而是主动调用下游服务并承担超时熔断、降级兜底、结果聚合的职责。3.2 核心代码实现一个可复用的Agent Router在src/routers/agent_router.py中我们定义主入口from fastapi import APIRouter, Depends, HTTPException, status from pydantic import BaseModel from typing import Dict, Any import asyncio from src.services.agent_orchestrator import AgentOrchestrator from src.core.dependencies import get_orchestrator router APIRouter(prefix/v1/ask, tags[Agent]) class AskRequest(BaseModel): query: str user_id: str session_id: str class AskResponse(BaseModel): answer: str sources: list[str] chart_data: Dict[str, Any] | None None router.post(/, response_modelAskResponse) async def handle_ask( request: AskRequest, orchestrator: AgentOrchestrator Depends(get_orchestrator), ) - AskResponse: try: # 设置全局超时总耗时≤2.5秒其中LLM调用≤1.2秒 result await asyncio.wait_for( orchestrator.route_and_execute(request), timeout2.5 ) return result except asyncio.TimeoutError: raise HTTPException( status_codestatus.HTTP_408_REQUEST_TIMEOUT, detailAgent processing timed out. Please try a simpler question. ) except Exception as e: # 降级返回缓存结果或通用提示 if cached : await get_cached_answer(request.query): return cached raise HTTPException( status_codestatus.HTTP_500_INTERNAL_SERVER_ERROR, detailfAgent execution failed: {str(e)} )关键在AgentOrchestrator.route_and_execute方法src/services/agent_orchestrator.pyclass AgentOrchestrator: def __init__(self, llm_client, vector_store, db_engine): self.llm_client llm_client self.vector_store vector_store self.db_engine db_engine async def route_and_execute(self, request: AskRequest) - AskResponse: # Step 1: 意图识别轻量级用小模型或规则 intent await self._identify_intent(request.query) # Step 2: 动态选择Agent策略 if intent sql_query: return await self._execute_sql_agent(request) elif intent document_qa: return await self._execute_document_agent(request) elif intent data_viz: return await self._execute_viz_agent(request) else: # 默认走混合AgentRAG SQL return await self._execute_hybrid_agent(request) async def _identify_intent(self, query: str) - str: # 用本地tiny-llm快速分类避免调用大模型 prompt fClassify the user query into one category: - sql_query: asks for database records, metrics, comparisons - document_qa: asks about policies, manuals, reports - data_viz: asks for charts, trends, visual summaries Query: {query} Category: # 调用量化后的Phi-3-mini模型500MB毫秒级响应 return await self.llm_client.invoke(prompt)3.3 生产级配置让FastAPI真正扛住并发默认的Uvicorn配置在问数项目里必崩。以下是config/uvi_config.py的实战参数import multiprocessing # 并发模型用uvloop替代asyncio默认事件循环 UVLOOP_ENABLED True # 进程数设为CPU核心数-1留1核给OS WORKERS multiprocessing.cpu_count() - 1 # 每个Worker的并发连接数非请求数 WORKER_CONNECTIONS 1000 # 超时设置单位秒 TIMEOUT_KEEP_ALIVE 5 GRACEFUL_TIMEOUT 120 # 关键限制每个Worker的内存防止OOM MAX_REQUESTS 1000 # 处理1000请求后重启Worker MAX_REQUESTS_JITTER 100 # 避免所有Worker同时重启 # SSL配置生产环境必须 SSL_KEYFILE /etc/ssl/private/question-agent.key SSL_CERTFILE /etc/ssl/certs/question-agent.crt # 日志精细化 LOG_LEVEL info ACCESS_LOG True ERROR_LOG True # 关键启用结构化JSON日志方便ELK采集 STRUCTURED_LOGGING True启动命令scripts/start.sh#!/bin/bash # 使用uvloop提升异步性能 export UVLOOP_ENABLEDtrue # 绑定到localhost由Nginx做外网代理安全最佳实践 uvicorn src.main:app \ --host 127.0.0.1 \ --port 8000 \ --workers $(( $(nproc) - 1 )) \ --worker-class uvicorn.workers.UvicornWorker \ --timeout-keep-alive 5 \ --max-requests 1000 \ --max-requests-jitter 100 \ --log-level info \ --access-log \ --reload # 开发环境开启热重载注意生产环境务必关闭--reload并用--preload预加载应用避免Worker启动时重复初始化数据库连接池。4. 向量与结构化数据双引擎如何让Agent既懂语义又懂SQL4.1 为什么必须双引擎单靠RAG或SQL都不够问数项目的业务数据有两大特征结构化数据占80%销售订单、库存流水、用户画像全在PostgreSQL里字段类型严格关联关系复杂。非结构化知识占20%产品说明书、客服话术、内部培训PPT这些文本需要语义检索。如果只用RAG向量检索用户问“iPhone 15 Pro的保修期是多久”RAG能从PDF里找到答案但问“华东区上月iPhone 15 Pro销量环比增长多少”RAG就抓瞎——它无法执行SELECT SUM(qty) FROM sales WHERE productiPhone 15 Pro AND regionEast China AND month2024-05。如果只用SQL Agent用户问“哪些产品最近投诉率上升最快”SQL Agent需要人工写WHERE条件而业务人员根本不知道投诉表里字段叫complaint_rate还是issue_frequency。LCODER的解法是用向量引擎做Schema理解用SQL引擎做精确计算。流程如下用户问句输入 → 向量库检索最相关的3个数据库表名字段说明如sales_order表的order_date,product_id,qty字段注释LLM根据检索结果生成精准SQL不是瞎猜是基于真实SchemaSQL执行 → 结果喂给LLM做自然语言总结 图表生成这样向量引擎是“导航员”SQL引擎是“挖掘机”Agent才是“项目经理”。4.2 PostgreSQL pgvector零成本实现企业级向量库PostgreSQL 15原生支持向量类型无需额外部署Chroma/Milvus。问数项目用pgvector扩展优势明显数据一致性向量和业务数据在同一事务里更新避免ES和DB双写不一致。权限统一DBA用现有PG权限体系控制向量数据访问不用学新ACL语法。运维简单备份、监控、扩容全部复用现有PG运维脚本。建表语句src/db/schema.sql-- 启用pgvector扩展 CREATE EXTENSION IF NOT EXISTS vector; -- 存储表结构元数据供Agent理解Schema CREATE TABLE schema_embeddings ( id SERIAL PRIMARY KEY, table_name VARCHAR(100) NOT NULL, column_name VARCHAR(100) NOT NULL, description TEXT NOT NULL, embedding VECTOR(384), -- 使用all-MiniLM-L6-v2模型输出384维 created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() ); -- 创建向量索引IVFFLAT比HNSW省内存适合问数项目规模 CREATE INDEX ON schema_embeddings USING ivfflat (embedding vector_cosine_ops) WITH (lists 100);插入Embedding的Python代码src/db/vector_loader.pyfrom pgvector.asyncpg import register_vector import asyncpg from sentence_transformers import SentenceTransformer # 初始化模型量化版内存占用300MB model SentenceTransformer(all-MiniLM-L6-v2, devicecpu) async def load_schema_embeddings(): conn await asyncpg.connect(postgresql://user:passlocalhost:5432/question_db) await register_vector(conn) # 注册vector类型 # 批量插入避免逐条INSERT embeddings [] for table_info in get_table_schema(): # 获取PG的information_schema text fTable {table_info[table]} column {table_info[column]} means {table_info[desc]} vec model.encode(text).tolist() embeddings.append((table_info[table], table_info[column], table_info[desc], vec)) await conn.executemany( INSERT INTO schema_embeddings (table_name, column_name, description, embedding) VALUES ($1, $2, $3, $4), embeddings ) await conn.close()4.3 SQL Agent的鲁棒性设计如何让LLM不写错SQLLLM生成SQL最大的风险是语法错误漏逗号、括号不匹配语义错误JOIN条件写错导致笛卡尔积权限错误查了不该查的表问数项目的防御三层第一层Prompt约束在System Prompt里明确要求You are a SQL expert for PostgreSQL 15. Return ONLY valid SQL, no explanation. Rules: - Use table aliases (t1, t2) for all JOINs - Always use LIMIT 100 unless user specifies otherwise - Never use SELECT *; list columns explicitly - If unsure of table/column existence, use INFORMATION_SCHEMA to verify first第二层SQL验证器执行前用sqlparse解析语法再用psycopg2的cursor.execute(EXPLAIN sql)检查执行计划import sqlparse from psycopg2 import sql def validate_sql(sql_str: str) - bool: # 语法检查 try: parsed sqlparse.parse(sql_str)[0] if not parsed.is_group: return False except: return False # 执行计划检查模拟执行不真正跑 try: with engine.connect() as conn: # 检查是否有危险操作 if DROP in sql_str.upper() or DELETE in sql_str.upper(): return False # EXPLAIN获取执行计划 plan conn.execute(text(fEXPLAIN (FORMAT JSON) {sql_str})).fetchone()[0] # 检查是否扫描全表rows 100000 if plan[Plan][Plan Rows] 100000: return False except Exception as e: return False return True第三层沙箱执行所有SQL在只读副本上执行且加SET statement_timeout 50005秒超时# 在数据库连接URL里指定只读副本 DATABASE_URL postgresql://user:passpg-ro:5432/question_db?options-c%20statement_timeout%3D50005. 可观测性与安全加固让Agent从“能跑”到“敢上生产”5.1 可观测性三支柱Metrics、Logs、Traces没有可观测性Agent就像黑盒。问数项目用开源栈实现MetricsPrometheus FastAPI内置/metrics端点LogsJSON格式日志 → Loki → Grafana看板TracesOpenTelemetry SDK → Jaeger在src/middleware/observability.py中注入中间件from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.jaeger.thrift import JaegerExporter from fastapi import Request, Response import time # 初始化Tracer provider TracerProvider() processor BatchSpanProcessor( JaegerExporter( agent_host_namejaeger, agent_port6831, ) ) provider.add_span_processor(processor) trace.set_tracer_provider(provider) async def observability_middleware(request: Request, call_next): tracer trace.get_tracer(__name__) with tracer.start_as_current_span(http_request) as span: span.set_attribute(http.method, request.method) span.set_attribute(http.url, str(request.url)) start_time time.time() response await call_next(request) process_time time.time() - start_time span.set_attribute(http.status_code, response.status_code) span.set_attribute(http.process_time, process_time) # 记录慢请求1秒 if process_time 1.0: span.add_event(slow_request, {duration: process_time}) return responsePrometheus指标暴露src/metrics.pyfrom prometheus_client import Counter, Histogram, Gauge from fastapi import APIRouter router APIRouter() # 自定义指标 QUERY_COUNTER Counter(question_agent_queries_total, Total number of queries) QUERY_DURATION Histogram(question_agent_query_duration_seconds, Query duration in seconds) ACTIVE_AGENTS Gauge(question_agent_active_agents, Number of active agent instances) router.get(/metrics) def metrics(): return Response( contentprometheus_client.generate_latest(), media_typetext/plain )5.2 安全加固Agent不是玩具必须防住恶意输入Agent直面用户攻击面极大。问数项目加固点输入净化用bleach库过滤HTML/JS标签防止XSS用正则限制用户问句长度≤500字符防DoS对SQL关键词UNION,SELECT,FROM做模糊匹配告警输出脱敏敏感字段身份证、手机号、银行卡在返回前用正则替换配置白名单字段只允许返回业务必需字段模型调用防护所有LLM调用加system_prompt硬约束“你只能回答与数据库查询相关的问题拒绝任何其他请求”用llm-guard库做输出内容审核拦截越狱提示词关键代码src/security/sanitizer.pyimport re import bleach def sanitize_user_input(query: str) - str: # 截断过长输入 if len(query) 500: query query[:500] ... # 过滤HTML标签 query bleach.clean(query, tags[], stripTrue) # 检测SQL注入模式 dangerous_patterns [r(union\sselect), r(drop\stable), r(;.*select)] for pattern in dangerous_patterns: if re.search(pattern, query.lower()): raise ValueError(Suspicious SQL pattern detected) return query def redact_sensitive_output(data: dict) - dict: # 身份证脱敏110101199003072712 → 110101****03072712 id_pattern r(\d{6})\d{8}(\d{4}) data_str json.dumps(data) data_str re.sub(id_pattern, r\1****\2, data_str) # 手机号脱敏13812345678 → 138****5678 phone_pattern r(\d{3})\d{4}(\d{4}) data_str re.sub(phone_pattern, r\1****\2, data_str) return json.loads(data_str)5.3 生产部署Checklist上线前必须验证的12件事序号检查项验证方法不通过后果1Python环境版本锁定uv pip freeze --locked requirements.lock对比各环境是否一致版本漂移导致功能异常2数据库连接池大小SHOW pool_size;确认≥50高并发时连接耗尽HTTP 5033Redis内存使用率redis-cli INFO memory | grep used_memory_human 70%缓存淘汰导致命中率暴跌4pgvector索引状态SELECT * FROM pg_indexes WHERE tablenameschema_embeddings;无索引时向量检索变全表扫描5FastAPI超时配置curl -v http://localhost:8000/v1/ask观察响应头超时未生效用户等待超时6敏感字段脱敏开关用含身份证的测试数据调用API检查返回值泄露用户隐私合规风险7Prometheus指标暴露curl http://localhost:8000/metrics | grep question_agent无法监控故障难定位8Jaeger链路追踪发起请求后访问http://jaeger:16686搜索服务名全链路不可见排查效率低9SQL执行计划优化EXPLAIN ANALYZE SELECT ...确认无Seq Scan查询慢用户体验差10LLM调用限流用ab -n 100 -c 20 http://...压测检查429响应被上游LLM服务商封禁11日志级别设置grep INFO|ERROR /var/log/question-agent.logDEBUG日志刷爆磁盘12SSL证书有效性openssl s_client -connect your-domain.com:443 -servername your-domain.com浏览器报不安全用户流失最后再强调一次基础设施不是“搭完就完事”而是持续演进的过程。我见过太多团队基础设施搭完就扔给运维结果三个月后没人记得Redis密码半年后没人知道pgvector索引怎么重建。LCODER的做法是把所有基础设施操作封装成Makefile命令比如make infra-up一键启动make infra-test跑全链路冒烟测试make infra-destroy清理环境。让基础设施像代码一样可版本化、可测试、可协作——这才是AI Agent能真正落地的根基。
