最近我在搞一个 AI Agent 的项目，想把它跑到生产周边环境去。较大家都说要“较高并发”，我就懵逼：到底怎么做到？于是我把一堆命令、配置、代码放在一起，想给你们做个“教程”。不过我得说一句——这不是一本正经的技术手段手册，而是我的草稿。写得很烂，你看着笑笑就行。

并发到底是啥？

并发，就是同一时间段里有很更多申请同时也跑。对 AI Agent 往往不是单纯的网络层面而是整个链路：用户申请 → 路由 → FastAPI → Celery → LLM API / DB / 工具调用 → 返回最终还是结果是。 不如... 如果链路里任意一步卡住就会造成雪崩。

为哪些单机加CPU不行？

很更多人以为更多核就良好了 可是：

• LLM 调用投入成本较高，往往受供应商限速；
• 数据库连接池耗尽； psql -c "SHOW max_connections;"
• Redis 内存溢出； maxmemory 1gbmaxmemory-policy allkeys-lru
• 工具调用阻塞引起线程占满。

基础设施准备

Nginx 安装：

sudo apt install -y nginx
sudo systemctl enable nginx
sudo systemctl start nginx

PostgreSQL 安装：

sudo apt install -y postgresql postgresql-contrib
sudo systemctl enable postgresql
sudo systemctl start postgresql
# 创建数据库和用户
sudo -u postgres psql -c "CREATE DATABASE agentdb;"
sudo -u postgres psql -c "CREATE USER agentuser WITH PASSWORD 'agentpassword';"
sudo -u postgres psql -c "GRANT ALL PRIVILEGES ON DATABASE agentdb TO agentuser;"

Redis 安装：

sudo apt install -y redis-server
sudo systemctl enable redis-server
sudo systemctl start redis-server
# 简洁检查一下有没有可用
redis-cli ping   # 应当返回 PONG，如果没有就跑下面的命令来沉重启一下吧。

Nginx + Uvicorn + Gunicorn

划水。 Nginx 做入口限流与反向代理，后端用 Uvicorn+Gunicorn 处理 Python 申请。

server {
    listen 80;
    server_name your-domain.com;
    limit_req_zone $binary_remote_addr zone=agent_limit:10m rate=20r/s;
    limit_req zone=agent_limit burst=40 nodelay;
    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_connect_timeout 30s;
        proxy_read_timeout 120s;
    }
}

Gunicorn 配置：

调整一下。 bind = '0.0.0.0:8000' workers = or 4) * 2 + 1 # 粗糙估算更多进程数，可根据坚硬件调试。 workerclass = 'uvicorn.workers.UvicornWorker' timeout = 180 # 超时设置较大一点，让 LLM 有机会完成。 keepalive = 30 # KeepAlive 超时时间段。 maxrequests = 20000 # 避免内存泄漏时崩溃。 errorlog = '-' accesslog = '-' loglevel = 'info' workertmpdir = '/tmp/gunicorn' # 临时文件目录。 sys.path.insert # 加载项目路径。 reload = False # 开发周边环境能够开启炎热加载。

bash

gunicorn app.main:app -c gunicorn_config.py --daemon

FastAPI 主体代码

不如... from fastapi import FastAPI, HTTPException, Request from pydantic import BaseModel import asyncio import httpx

app = FastAPI

class AgentRequest: us 也是醉了... er_id: str query: str

@app.post async def handleagent: try: result = await runagent return 也是没谁了。 {"status":"ok","answer":result} except Exception as e: raise HTTPException)

到位。 async def runagent: # 简化：直接调用 LLM 接口， 实际项目请加缓存/限流/工具调用等逻辑 async with httpx.AsyncClient as client: resp = await client.post( url='https://api.openai.com/v1/chat/completions', headers={'Authorization': f'Bearer YOURAPIKEY'}, json={ 'model':'gpt-4o-mini', 'messages':, 'temperature':0, }) resp.raisefor_status data=resp.json return data

Caching 与限流实现粗糙版示例

杀疯了！ import redis import hashlib import json

redis_client = redis.Redis，何不...

def makecachekey: raw=json.dumps.encode return hashlib.md5.hexdigest，往白了说...

格局小了。 def getcached: key=f'cache:{makecachekey}' return redisclient.get

滑动窗口限流示例

def isallowed:
    now=int)
    pipeline=redisclient.pipeline
    rkey=f'rate:{key}'
    pipeline.zremrangebyscore,now-time_window)
    pipeline.zadd
    pipeline.zcard

bash pipelines.execute count=pipeline return count<=max_req

采用方式：

划水。 python if not is_allowed: raise HTTPException

Celery 异步任务 & 队列

from celery import Celery

KTV你。 celeryapp=Celery celeryapp.conf.update( taskserializer='json', resultserializer='json', acceptcontent=, timezone='Asia/Shanghai', enableutc=False, )

@celeryapp.task def runagenttask: try: res=runllmlogic # 用上面 FastAPI 的 runllm_logic 函数或自己实现 return {'status':'success','data':res} except Exception as exc: self.retry

启动 Celery Worker：

白嫖。 bash celery -A celery_app worker --loglevel info --concurrency=8 --queues agent_high,agent_normal

AIOHTTP 用来提交异步任务的接口

@app.post async def submitasync: task=runagentasync.delay # runagen 一针见血。 tasync 是 Celery 中定义的异步函数名 return {"taskid":task.id,"status":"submitted"}

梳理梳理。 @app.get async def gettask: res=runagentasync.AsyncResult if res.state=='SUCCESS': return {"status":"done","result":res.result} else: return {"status":res.state,"progress":res.info}

Docker Compose 一键部署

version:"3" services: api: build:. containername:"ai-agent-api" ports: envfile: dependson: celery: build:. containername:"ai-agent-celery" command:"celery -A celeryapp worker --loglevel info 当你.… --concurrency=8" envfile: dependson: redis: image:"redis:7" containername:"ai-agent-redis" pg: image:"postgres:16" container_name:"ai-agent-postgres" ... volumes:{ pgdata:/var/lib/postgresql/data }

运行：

太顶了。 bash docker compose up --build -d docker compose logs -f api docker compose logs -f celery

压测 Agent 服务

wrk 示例： bash wrk -t4 -c100 -d60s http://127\.0\.0\.1\:8000/agent，躺平...

hey 示例： bash hey \ -n10000 \ -c200 \ -mPOST \ -H"Content-Type: application/json" \ 我无法认同... -d '{"user_id":"bench","query":"请阐述哪些是 AI Agent"}' \ http://127\.0\.0\.1\:8000/agent

压测时注意查看指标： Promeus 的 /metrics 接口能够暴露 "request_count", "latency". 若出现较更多缓慢申请， 完善一下。 一定要排查 LLM 调用或 DB 等。

提议采用 k6 或 Locust 做更细粒度的压力测试， 在我看来... 但这里先保留简洁脚本即可。

"如果你发觉全部申请都排队变成秒级延迟， 那说明你的模型 我的看法是... 接口被限速了或者 Redis 没有足够内存。再往下看…"

"PostgreSQL 最较大连接数最良好不要较高于 DB 容器允许的最较大值，否则会报错。能够考虑 PgBouncer 或者直接把 pool_size 约束在50左右。"​，内卷。

"如果你还没安装 Promeus 和 Grafana， 能够随便跑一个 docker 镜像，然后把 metrics 路径挂进去。"​，等..….

AIOHTTP+Celery+Redis+Postgres 是最常见组合； 将心比心... 但关键在于每一层都要合理约束和缓存，否则瓶颈只会转移到别处。

Aggressive scaling 并非仅有解法；优先考虑链路优化、 异步化、分布式缓存，再做水平扩容，大体上...。

MVP 阶段能够先跑较小规模测试， 如果出现雪崩现象，再逐步提升资源条件； ICU你。 千万别一次性投较大容量服务器然后再抬头问自己为哪些缓慢。

"投入成本控制" 在 AI 场景尤其十分沉关键：Token 数量直接决定账单较大较小。一定要做 token‑cost 可视化，也别忘了记录每次申请消耗了更多更少个 token。

---

©2026 AI Lab | 本文仅作学习了解交流之用，请勿用于商业活动用途。如有侵权，请联系删除。​ 📚️✏️💻🤖 "嘿， 我写这篇文章的时候真实的是想帮朋友搞个 Demo，但后来发觉自己的代码写得太烂，于是就顺手把它们拼凑成了一个“完整方案”。如果你读完后觉得良好像被灌了一较大袋狗屎一样，那就是我的工作岗位成果啦！"​

---