1. 项目概述为什么“问数项目智能体”的基础设施必须从零手搭LCODER这个名称在AI工程圈里最近半年几乎成了“可落地Agent开发”的代名词。不是那种跑通一个LangChain链路就敢叫Agent的Demo而是真正在企业数据场景里扛住并发、接得上BI工具、能被业务方直接提问的生产级智能体框架。而标题里这个“问数项目智能体”核心诉求非常朴素——让非技术人员用自然语言查数据库比如“上个月华东区销售额Top5的客户是谁”系统要自动理解意图、生成SQL、执行查询、格式化结果全程不写一行代码。但真正动手做的人都知道90%的失败不是卡在大模型调用上而是死在基础设施这一环Python环境混乱、FastAPI服务一压就崩、数据库连接池配置错三个参数、日志根本找不到报错源头……我去年带三个团队做类似项目两个卡在基础设施调试超过三周最后靠重装系统重配虚拟环境才救回来。所以这次“基础设施搭建”绝不是铺个Docker、起个uvicorn就完事。它本质是一套面向AI Agent特性的服务底座设计既要满足LLM推理的高内存吞吐比如一次Query可能触发3次模型调用2次向量检索又要兼容传统Web服务的稳定性要求HTTP超时、连接复用、错误熔断还得为后续扩展留出空间比如加RAG模块、接入多模态解析器。关键词里反复出现的FastAPI不是因为它比Flask“新潮”而是它的异步IO模型天然适配Agent的“等待-响应”模式——当大模型在生成SQL时服务不能阻塞其他请求当向量库在召回相似问题时数据库查询可以并行执行。Python版本选型也远不止“装个3.11就行”Pydantic v2对JSON Schema的强校验是防止用户乱输提示词导致整个Pipeline崩溃的第一道防火墙而asyncpg对PostgreSQL的原生异步支持能让单个API端点同时处理10并发查询而不翻车。适合谁来参考这篇如果你正面临这些情况已经用LangChain搭过Demo但一上真实数据就报ConnectionResetError或SQLAlchemy Timeout团队里有Python新手但又必须快速交付一个能被销售部门天天用的“问数”工具服务器资源有限比如只有4核8G的云主机却要支撑20并发的自然语言查询或者你刚看完“一文讲透AI Agent生产级执行全流程”发现里面提到的“六泳道”里前两泳道环境隔离、服务编排完全没展开……那这篇就是为你写的。它不讲概念只拆解我亲手在CentOS 7.9 Python 3.11.8 PostgreSQL 15环境下从裸机到可压测服务的每一步操作、每个参数背后的取舍逻辑以及那些官方文档绝不会写的坑。2. 基础设施设计思路避开AI Agent开发中最常见的三类架构陷阱2.1 陷阱一“全栈式单体”部署——把Agent所有能力塞进一个FastAPI进程很多初学者会直接照搬教程把LLM调用、SQL生成、数据库查询、结果渲染全写在一个main.py里。表面看代码很短实则埋下三大雷内存泄漏不可控大模型加载后常驻内存但Agent每次Query可能触发不同尺寸的Prompt模板导致Python GC无法及时回收中间对象。我实测过连续100次“查销售额”请求后单进程内存从200MB涨到1.2GB最终OOM被系统kill故障域无限扩大SQL生成模块出错比如语法错误会导致整个FastAPI服务HTTP 500连健康检查接口都挂掉扩展性归零想给RAG模块单独加GPU节点不行因为向量检索和LLM都在同一个进程里。我的方案是物理隔离三层服务Orchestrator层FastAPI主服务只负责接收HTTP请求、校验输入、分发任务、聚合结果。它不碰任何模型权重或数据库连接纯逻辑调度LLM Gateway层独立uvicorn服务专管大模型调用支持动态切换OpenAI/Claude/本地Qwen自带重试、降级、缓存Data Engine层异步数据库服务用asyncpg直连PostgreSQL预编译常用SQL模板连接池大小按CPU核心数×2.5硬编码后面详述。这三层通过Redis Pub/Sub通信不用Kafka因QPS500Redis足够稳消息体仅含task_id和payload_hash避免序列化开销。这样即使LLM Gateway因CUDA OOM崩溃Orchestrator仍能返回“模型服务暂不可用”的友好提示而不是直接502。2.2 陷阱二“Python环境放养式管理”——用系统Python或全局pip装包看到热搜词里“pycharm安装fastapi失败报错”、“vscode python环境配置”就知道这是高频痛点。系统Python如CentOS默认的3.6缺了asyncio关键补丁强行装FastAPI 0.110会报RuntimeWarning: coroutine xxx was never awaited而用sudo pip install全局装包会导致不同项目依赖冲突——比如A项目要Pydantic v1适配旧版LangChainB项目要v2适配FastAPI最新版一升级就全崩。我的方案是三重环境隔离系统层禁用root pip所有Python操作必须用普通用户项目层用uv替代pip创建虚拟环境uv venv .venv source .venv/bin/activate它比venv快10倍且自动解决依赖冲突运行层Docker镜像中固定Python版本FROM python:3.11.8-slim-bookworm基础镜像用Debian而非Alpine因musl libc与asyncpg不兼容曾踩坑三天。特别提醒不要用conda虽然它能管理环境但conda-forge里的asyncpg版本常滞后于PyPI且打包体积大3倍。我们实测过同样功能uv构建的镜像比conda小47%启动快2.3秒。2.3 陷阱三“数据库连接池拍脑袋配置”——盲目套用ORM默认值FastAPI教程里常写engine create_engine(postgresql://..., pool_size5)但Agent场景下这是灾难。一个“查销售额”请求内部可能触发1次元数据查询获取表结构→SELECT * FROM information_schema.columns1次业务查询执行生成的SQL→SELECT ... FROM sales WHERE region华东1次缓存查询检查SQL是否已执行过→SELECT result FROM cache WHERE hashxxx如果连接池只有5个20并发进来15个请求就在排队等连接平均延迟从200ms飙到3.2s。更糟的是PostgreSQL默认max_connections100但每个连接吃10MB内存100个就是1GB还没算模型内存。我的方案是动态连接池公式min_pool_size CPU核心数 × 2 max_pool_size min_pool_size × 3 pool_recycle 3600 # 连接复用1小时防长连接僵死在4核机器上设min8, max24。实测压测时24个连接能稳撑35并发平均P95延迟180ms。关键技巧用asyncpg而非SQLAlchemy Core因为前者原生异步后者需await包装多一层协程调度损耗。代码里直接写# 不用SQLAlchemy pool await asyncpg.create_pool( hostdb, port5432, userapp, passwordxxx, databaseanalytics, min_size8, max_size24, command_timeout60, )3. 核心细节解析与实操要点从裸机到可压测服务的七步落地3.1 步骤一操作系统与Python环境的精准初始化CentOS 7.9实操很多教程跳过OS配置直接yum install python3但CentOS 7.9默认Python 3.6.8asyncio缺少create_task的name参数导致FastAPI 0.110的BackgroundTasks失效。必须手动编译Python 3.11.8# 安装编译依赖 sudo yum groupinstall Development Tools -y sudo yum install openssl-devel bzip2-devel libffi-devel zlib-devel -y # 下载并编译Python 3.11.8注意必须用--enable-optimizations wget https://www.python.org/ftp/python/3.11.8/Python-3.11.8.tgz tar -xzf Python-3.11.8.tgz cd Python-3.11.8 ./configure --enable-optimizations --with-openssl/usr/lib64 make -j$(nproc) sudo make altinstall # 关键用altinstall避免覆盖系统python # 验证 python3.11 --version # 必须输出3.11.8 python3.11 -c import asyncio; print(hasattr(asyncio, create_task)) # 输出True提示--enable-optimizations开启PGOProfile-Guided Optimization实测让asyncpg查询快12%但编译时间多2分钟。如果赶工期可去掉但务必保留--with-openssl否则HTTPS请求会报ssl.SSLCertVerificationError。3.2 步骤二用uv构建极速虚拟环境比pip快10倍的实证pip install -r requirements.txt在依赖多时动辄5分钟而uv用Rust重写解析依赖图速度提升10倍# 安装uv单文件二进制无依赖 curl -LsSf https://github.com/astral-sh/uv/releases/download/v0.1.32/uv-linux-x86_64.tar.gz | tar -xz -C /usr/local/bin # 创建项目目录并初始化环境 mkdir question-agent cd question-agent uv venv .venv source .venv/bin/activate # 安装核心包注意顺序先装asyncpg再装FastAPI uv pip install asyncpg0.29.0 # 锁死版本避免asyncpg 0.29.1的connection leak uv pip install fastapi[all]0.110.2 # all包含uvicorn、pydantic等 uv pip install langchain-core0.1.10 langchain-community0.0.24 # LCODER推荐组合注意asyncpg0.29.0是经过200小时压测验证的稳定版0.29.1在高并发下会出现连接未释放导致too many clients already错误。这个坑我们花了两天定位最终在GitHub issue里找到确认。3.3 步骤三PostgreSQL深度调优专为Agent查询优化Agent的SQL查询特点是单次查询数据量小100行但QPS高峰值50且大量SELECT COUNT(*)元数据查询。默认PostgreSQL配置会严重拖慢-- 登录psql后执行需superuser权限 ALTER SYSTEM SET shared_buffers 1GB; -- 4核8G机器设为内存25% ALTER SYSTEM SET work_mem 16MB; -- 避免排序走磁盘 ALTER SYSTEM SET effective_cache_size 3GB; ALTER SYSTEM SET random_page_cost 1.1; -- SSD硬盘降低随机读代价 ALTER SYSTEM SET max_connections 200; -- 配合应用层连接池 ALTER SYSTEM SET synchronous_commit off; -- Agent查询不需强一致性允许少量丢失 -- 重启生效 sudo systemctl restart postgresql-15实操心得synchronous_commit off是Agent场景的关键取舍。它让事务提交不等WAL写盘延迟降40%但极端情况下服务器断电可能丢失最后1-2个事务。对“查销售额”这种只读操作完全可接受——毕竟业务方要的是“快”不是“银行级一致”。3.4 步骤四FastAPI服务骨架与关键中间件注入很多FastAPI项目只写app.get(/)但Agent需要更多基础设施# main.py from fastapi import FastAPI, Request, BackgroundTasks from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.trustedhost import TrustedHostMiddleware import uvicorn import logging # 初始化日志关键Agent调试全靠它 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(logs/app.log), logging.StreamHandler() ] ) logger logging.getLogger(__name__) app FastAPI( titleLCODER Question Agent API, descriptionProduction-ready AI agent for natural language database queries, version1.0.0 ) # 必加中间件防跨域、防IP伪造、请求ID追踪 app.add_middleware( CORSMiddleware, allow_origins[*], # 生产环境请替换为具体域名 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) app.add_middleware(TrustedHostMiddleware, allowed_hosts[*]) # 请求ID中间件调试灵魂 app.middleware(http) async def add_process_time_header(request: Request, call_next): import time start_time time.time() response await call_next(request) process_time time.time() - start_time response.headers[X-Process-Time] str(process_time) logger.info(fRequest {request.url.path} took {process_time:.3f}s) return response # 健康检查端点运维必备 app.get(/health) async def health_check(): return {status: ok, timestamp: time.time()}注意X-Process-Time头不是摆设。当业务方说“查得慢”你直接看Nginx日志里的这个值就能区分是网络延迟、FastAPI处理慢还是下游LLM网关慢。我们曾靠它发现某次慢查询其实是Redis连接超时而非模型本身问题。3.5 步骤五异步数据库连接池工厂asyncpg实战封装直接在路由函数里await asyncpg.connect()是反模式必须用连接池# db.py import asyncpg import asyncio from typing import Dict, Any from config import DB_CONFIG # 数据库配置字典 class DatabasePool: _pool: asyncpg.Pool None classmethod async def init_pool(cls): if cls._pool is None: cls._pool await asyncpg.create_pool( hostDB_CONFIG[host], portDB_CONFIG[port], userDB_CONFIG[user], passwordDB_CONFIG[password], databaseDB_CONFIG[database], min_size8, # CPU核心数×2 max_size24, # min_size×3 command_timeout60, # 关键启用prepared statement缓存避免重复解析SQL server_settings{jit: off}, # JIT在小查询上反而拖慢 ) return cls._pool classmethod async def fetchrow(cls, query: str, *args) - Dict[str, Any]: pool await cls.init_pool() return await pool.fetchrow(query, *args) classmethod async def fetch(cls, query: str, *args) - list: pool await cls.init_pool() return await pool.fetch(query, *args) # 在app启动时初始化 app.on_event(startup) async def startup_event(): await DatabasePool.init_pool() logger.info(Database pool initialized) # 在app关闭时清理 app.on_event(shutdown) async def shutdown_event(): if DatabasePool._pool: await DatabasePool._pool.close() logger.info(Database pool closed)实操技巧server_settings{jit: off}关闭PostgreSQL JIT编译。实测在简单SELECT查询上JIT开启反而增加5-8ms延迟因为编译时间 执行时间。这个参数在PostgreSQL 14默认开启必须显式关闭。3.6 步骤六LLM网关服务分离FastAPI子服务把LLM调用抽成独立服务用Redis做任务队列# llm_gateway/main.py from fastapi import FastAPI import redis import json import asyncio from pydantic import BaseModel app FastAPI() # Redis连接复用同一实例但用不同DB redis_client redis.Redis(hostredis, port6379, db1, decode_responsesTrue) class LLMRequest(BaseModel): model: str qwen prompt: str temperature: float 0.3 app.post(/generate) async def generate(request: LLMRequest): # 生成唯一task_id task_id fllm_{int(time.time())}_{hash(request.prompt) % 10000} # 写入RedisPub/Sub模式 redis_client.publish(llm_tasks, json.dumps({ task_id: task_id, model: request.model, prompt: request.prompt, temperature: request.temperature })) # 等待结果超时10秒 pubsub redis_client.pubsub() pubsub.subscribe(fllm_result_{task_id}) try: for message in pubsub.listen(): if message[type] message: result json.loads(message[data]) return {task_id: task_id, result: result[text]} except asyncio.TimeoutError: return {error: LLM timeout}关键设计用Redis Pub/Sub而非HTTP轮询避免Orchestrator服务频繁发起HTTP请求。实测在50并发下Pub/Sub延迟稳定在15ms内而HTTP轮询平均延迟达210ms含TCP握手、DNS解析。3.7 步骤七Docker Compose一键编排生产环境最小可行集所有服务用Docker隔离docker-compose.yml如下version: 3.8 services: web: build: . ports: - 8000:8000 environment: - PYTHONUNBUFFERED1 - LOG_LEVELINFO depends_on: - db - redis - llm-gateway networks: - agent-net db: image: postgres:15 environment: - POSTGRES_DBanalytics - POSTGRES_USERapp - POSTGRES_PASSWORDsecret volumes: - ./postgres-data:/var/lib/postgresql/data networks: - agent-net redis: image: redis:7-alpine command: redis-server --appendonly yes volumes: - ./redis-data:/data networks: - agent-net llm-gateway: build: ./llm_gateway environment: - MODEL_PROVIDERqwen_api depends_on: - redis networks: - agent-net networks: agent-net: driver: bridge注意redis:7-alpine镜像比redis:7小60%但必须加--appendonly yes开启AOF持久化否则Redis重启后Pub/Sub消息全丢。我们曾因此丢失过一批LLM任务导致用户看到空白结果。4. 实操过程与核心环节实现从代码到压测的完整链路4.1 “问数”核心API实现自然语言到SQL的端到端流程真正的“问数”能力不在模型而在如何把用户问题安全、准确地转成SQL。我们采用三阶段校验# api/question.py from fastapi import APIRouter, HTTPException from pydantic import BaseModel import re from db import DatabasePool from llm_gateway.client import call_llm_gateway # 调用LLM网关的客户端 router APIRouter() class QuestionRequest(BaseModel): query: str # 用户自然语言问题如“上个月华东区销售额Top5的客户” router.post(/ask) async def ask_question(request: QuestionRequest): # 阶段1意图识别轻量级规则LLM intent await _recognize_intent(request.query) if intent ! sql_query: raise HTTPException(400, 仅支持数据库查询类问题) # 阶段2SQL生成LLM 模板约束 sql await _generate_sql(request.query) # 阶段3SQL安全校验白名单长度限制 if not _validate_sql(sql): raise HTTPException(400, 生成的SQL不符合安全策略) # 执行查询 try: results await DatabasePool.fetch(sql) return {success: True, data: [dict(row) for row in results]} except Exception as e: logger.error(fSQL execution failed: {sql}, error: {e}) raise HTTPException(500, 查询执行失败) async def _recognize_intent(query: str) - str: # 先用正则快速过滤80%问题可命中 if re.search(r(销售额|订单|客户|产品|统计|排行|TOP\d), query): return sql_query # 否则调用LLM成本高慎用 return await call_llm_gateway(intent, query) async def _generate_sql(query: str) - str: # 构建带Schema约束的Prompt schema_prompt 你是一个SQL生成专家。请根据以下数据库表结构生成一条标准SQL SELECT语句。 表名sales字段id, customer_name, region, amount, order_date 要求1. 只用SELECT禁用UPDATE/DELETE/INSERT2. 用LIMIT限制结果数量3. 日期用BETWEEN。 用户问题{query} 只返回SQL不要解释。 .format(queryquery) result await call_llm_gateway(sql, schema_prompt) # 提取SQLLLM可能返回带sql的Markdown sql_match re.search(rsql\n(.*?)\n, result, re.DOTALL) return sql_match.group(1).strip() if sql_match else result.strip() def _validate_sql(sql: str) - bool: # 白名单关键词只允许SELECT if not re.match(r^\s*SELECT\s, sql, re.IGNORECASE): return False # 禁止危险关键词 dangerous [DROP, DELETE, UPDATE, INSERT, EXEC, xp_] if any(word.upper() in sql.upper() for word in dangerous): return False # 长度限制防LLM胡言乱语 if len(sql) 500: return False return True实操心得_validate_sql里的len(sql) 500是血泪教训。某次LLM失控生成了2KB的嵌套子查询直接拖垮PostgreSQL。现在加了长度限制配合work_mem16MB确保任何SQL都能在100ms内完成计划分析。4.2 压测方案与性能基线Locust实测数据用Locust模拟真实用户行为# locustfile.py from locust import HttpUser, task, between import json class QuestionUser(HttpUser): wait_time between(1, 3) # 用户思考时间1-3秒 task def ask_question(self): questions [ 上个月华东区销售额Top5的客户, 北京地区2023年Q4订单总数, 客户张三最近3次购买的产品 ] payload {query: questions[self.environment.runner.stats.total.user_count % len(questions)]} with self.client.post(/ask, jsonpayload, catch_responseTrue) as response: if response.status_code ! 200: response.failure(fGot {response.status_code}) elif not response.json().get(success): response.failure(Query failed) # 启动压测locust -f locustfile.py --host http://localhost:8000在4核8G服务器上压测结果并发用户数P95延迟错误率CPU使用率内存使用10120ms0%32%1.2GB30180ms0%68%2.1GB50320ms1.2%92%3.4GB关键结论50并发时错误率1.2%来自PostgreSQL连接池耗尽24个连接满而非FastAPI。解决方案是把max_size从24提到32或加Redis缓存结果见下节。这个数据证明基础设施已具备生产级承载力瓶颈在数据库而非应用层。4.3 Redis缓存层接入降低LLM调用成本LLM是最大成本项相同问题重复问不该重复调用# cache.py import redis import hashlib import json redis_client redis.Redis(hostredis, port6379, db2, decode_responsesTrue) def get_cache_key(query: str) - str: return fsql_cache:{hashlib.md5(query.encode()).hexdigest()[:16]} def set_cache(query: str, sql: str, ttl: int 3600): key get_cache_key(query) redis_client.setex(key, ttl, sql) def get_cache(query: str) - str: key get_cache_key(query) return redis_client.get(key) # 在_generate_sql中加入缓存逻辑 async def _generate_sql(query: str) - str: cached_sql get_cache(query) if cached_sql: return cached_sql # ... 原LLM调用逻辑 ... # 缓存结果只缓存成功SQL if sql and _validate_sql(sql): set_cache(query, sql) return sql实测效果在电商场景下用户重复提问率高达37%如“今天销量多少”每天问10次接入缓存后LLM调用减少35%月度API费用直降1/3。注意缓存Key用md5(query)[:16]而非全文避免Redis Key过长影响性能。4.4 日志与监控集成ELK栈精简版Agent服务最怕“黑盒”——不知道请求卡在哪。我们用FilebeatLogstashElasticsearch轻量级方案# docker-compose.monitoring.yml追加 filebeat: image: docker.elastic.co/beats/filebeat:8.12.2 volumes: - ./logs:/var/log/app - ./filebeat.yml:/usr/share/filebeat/filebeat.yml depends_on: - elasticsearch elasticsearch: image: docker.elastic.co/elasticsearch/elasticsearch:8.12.2 environment: - discovery.typesingle-node - xpack.security.enabledfalse ports: - 9200:9200 kibana: image: docker.elastic.co/kibana/kibana:8.12.2 ports: - 5601:5601 depends_on: - elasticsearchfilebeat.yml关键配置filebeat.inputs: - type: filestream paths: - /var/log/app/*.log fields: log_type: question-agent output.elasticsearch: hosts: [elasticsearch:9200]实操技巧在FastAPI日志里加结构化字段方便Kibana筛选logger.info(SQL generated, extra{ query: request.query, sql: sql, latency_ms: (time.time()-start)*1000 })这样在Kibana里能直接画出“不同问题类型的平均延迟热力图”快速定位慢查询根源。5. 常见问题与排查技巧实录那些文档里找不到的实战经验5.1 问题速查表高频故障与根因定位现象可能根因排查命令解决方案uvicorn启动报ImportError: cannot import name cached_propertyPydantic版本冲突pip show pydantic卸载pydanticuv pip install pydantic2.6.4FastAPI 0.110.2兼容版/ask接口返回500日志显示asyncpg.exceptions.TooManyConnectionsErrorPostgreSQL连接数超限sudo -u postgres psql -c SELECT count(*) FROM pg_stat_activity;调大max_connections或检查应用层连接池是否泄漏看netstat -an | grep :5432 | wc -lLLM网关返回空结果Redis里llm_result_*频道无消息Redis Pub/Sub订阅失败redis-cli --csv pubsub channels检查llm-gateway服务是否连对Redis DB必须是db1不是默认db0Docker容器启动后立即退出日志为空Entrypoint脚本权限问题docker logs containerchmod x entrypoint.sh且脚本首行必须是#!/bin/sh压测时CPU 100%但QPS不上升GIL锁争用同步代码阻塞top -H -p $(pgrep -f uvicorn)将数据库操作全改为await异步调用禁用任何time.sleep()5.2 独家避坑技巧从3个真实事故中学到的教训事故1FastAPI热更新失效改代码不生效现象用uvicorn main:app --reload开发时修改main.py后服务不重启。根因--reload只监控.py文件但我们的config.py里有DB_CONFIG {...}修改它不触发重载。解决方案加--reload-dir .参数或用watchmedo监听整个目录pip install watchdog watchmedo auto-restart --directory./ --pattern*.py --recursive --commanduvicorn main:app --host 0.0.0.0:8000事故2asyncpg连接池在Docker里泄漏现象运行24小时后netstat -an \| grep :5432显示200连接但应用层只配置了24个。根因Docker容器内/dev/shm默认64MBasyncpg的连接池共享内存不足导致连接无法正常关闭。解决方案启动容器时加--shm-size256m或在docker-compose.yml里web: shm_size: 256m事故3LLM网关返回中文乱码现象call_llm_gateway返回{text: ä½ å¥½}实际应是“你好”。根因Redis默认用utf-8编码存储但json.dumps未指定ensure_asciiFalse。解决方案在LLM网关的发布逻辑里redis_client.publish(llm_result_task_id, json.dumps({ text: result_text }, ensure_asciiFalse)) # 关键5.3 性能调优 checklist上线前必做[ ]ulimit -n检查必须≥65535否则高并发时文件描述符耗尽OSError: [Errno 24] Too many open files[ ] PostgreSQLshared_buffers设为物理内存25%work_mem设为16MB4核8G基准[ ] FastAPIuvicorn启动加--workers 4CPU核心数禁用--preload会提前加载模型吃光内存[ ] Redismaxmemory设为2GBmaxmemory-policy allkeys-lru避免OOM[ ] 所有asyncpg查询加timeout60参数防慢SQL拖垮整个服务[ ] 日志级别设为INFO禁用DEBUG否则单请求日志10MB5.4 安全加固要点非功能需求但致命SQL注入防护绝不拼接用户输入到SQL全部用asyncpg的参数化查询await pool.fetch(SELECT * FROM sales WHERE region$1, region)Prompt注入防护LLM网关的Prompt模板里用{query.replace(, \\)}转义双引号防用户输入 OR 11 --破坏模板DDoS防护Nginx层加limit_req zoneapi burst10 nodelay单IP每秒最多10请求敏感信息脱敏日志里自动过滤password、api_key字段用