AI Agent 高并发解决方案｜附完整命令

在大模型应用从 Demo 走向生产的过程中，AI Agent 的高并发能力往往会成为系统能否稳定上线的关键。很多团队在早期只关注 Prompt、模型效果、工具调用能力，但当用户量上来之后，就会遇到一系列典型问题：请求排队严重、响应时间变长、模型 API 被限流、Agent 执行链路过长、工具调用阻塞、数据库连接耗尽、任务失败重试导致雪崩等。

本文将从架构设计、并发控制、异步任务、缓存、限流、队列、服务部署、容器化和压测等方面，系统讲解一套适用于生产环境的 AI Agent 高并发解决方案，并附上完整命令示例，方便直接落地实践。

一、为什么 AI Agent 更容易遇到高并发问题？

普通的大模型问答应用通常是：

用户输入 -> 调用 LLM -> 返回结果

而 AI Agent 的链路往往更加复杂：

用户输入
  -> 意图理解
  -> 任务规划
  -> 调用工具
  -> 查询数据库
  -> 调用第三方接口
  -> 再次调用 LLM
  -> 汇总结果
  -> 返回用户

也就是说，一个 Agent 请求背后可能包含多次 LLM 调用、多次工具调用、多次数据库访问。假设一个用户请求平均需要调用 3 次大模型接口、2 次数据库查询、1 次外部 API，那么 1000 个并发请求可能瞬间放大成：

3000 次 LLM 请求
2000 次数据库访问
1000 次外部接口调用

这就是 AI Agent 系统的核心难点：表面并发不高，实际后端压力被执行链路放大了数倍。

因此，高并发 Agent 系统不能只靠“加服务器”解决，而要从架构层面做分层、削峰、限流、缓存、异步化和弹性伸缩。

二、整体架构设计

推荐的 AI Agent 高并发架构如下：

客户端
  |
  v
Nginx / API Gateway
  |
  v
FastAPI / Node.js Agent 服务
  |
  |---- Redis：缓存、分布式锁、限流、会话状态
  |
  |---- PostgreSQL / MySQL：业务数据存储
  |
  |---- Milvus / Qdrant / Weaviate：向量数据库
  |
  |---- Celery / Dramatiq / RQ：异步任务队列
  |
  |---- Kafka / RabbitMQ / Redis Stream：消息队列
  |
  |---- LLM Provider：OpenAI、Claude、Gemini、本地模型
  |
  v
监控系统：Prometheus + Grafana + Loki

核心思路是：

1. 入口层限流：防止流量瞬间打爆服务。
2. 应用层异步化：避免阻塞式调用拖垮线程。
3. 任务队列削峰：长耗时 Agent 任务进入队列。
4. 缓存热点结果：减少重复 LLM 和数据库查询。
5. 连接池管理：避免数据库、Redis、HTTP 连接耗尽。
6. 模型调用并发控制：遵守模型 API 限额。
7. 横向扩容：通过容器和负载均衡增加处理能力。
8. 监控告警：及时发现延迟、失败率和资源瓶颈。

三、环境准备

下面以 Ubuntu 22.04 服务器为例，演示一套基于：

• Python 3.11
• FastAPI
• Uvicorn
• Gunicorn
• Redis
• Celery
• PostgreSQL
• Nginx
• Docker
• Prometheus + Grafana

的高并发 Agent 服务部署方案。

1. 更新系统

sudo apt update && sudo apt upgrade -y

2. 安装基础工具

sudo apt install -y \
  curl \
  wget \
  git \
  vim \
  htop \
  unzip \
  build-essential \
  software-properties-common

3. 安装 Python 3.11

sudo apt install -y python3.11 python3.11-venv python3.11-dev python3-pip

查看版本：

python3.11 --version
pip3 --version

四、创建 Agent 项目

1. 创建项目目录

mkdir -p ~/ai-agent-high-concurrency
cd ~/ai-agent-high-concurrency

2. 创建虚拟环境

python3.11 -m venv .venv
source .venv/bin/activate

3. 安装依赖

pip install --upgrade pip
pip install \
  fastapi \
  uvicorn \
  gunicorn \
  httpx \
  redis \
  celery \
  pydantic \
  pydantic-settings \
  asyncpg \
  sqlalchemy \
  psycopg2-binary \
  python-dotenv \
  prometheus-client

生成依赖文件：

pip freeze > requirements.txt

五、核心代码结构

推荐项目结构：

ai-agent-high-concurrency/
├── app/
│   ├── main.py
│   ├── config.py
│   ├── limiter.py
│   ├── agent.py
│   ├── cache.py
│   ├── tasks.py
│   └── database.py
├── requirements.txt
├── .env
└── gunicorn_conf.py

创建目录：

mkdir -p app
touch app/main.py app/config.py app/limiter.py app/agent.py app/cache.py app/tasks.py app/database.py
touch .env gunicorn_conf.py

六、配置环境变量

编辑 .env：

vim .env

写入：

APP_NAME=AI_AGENT_SERVICE
ENV=production

REDIS_HOST=127.0.0.1
REDIS_PORT=6379
REDIS_DB=0

POSTGRES_HOST=127.0.0.1
POSTGRES_PORT=5432
POSTGRES_DB=agentdb
POSTGRES_USER=agentuser
POSTGRES_PASSWORD=agentpassword

LLM_API_KEY=your_api_key_here
LLM_BASE_URL=https://api.openai.com/v1
LLM_MODEL=gpt-4o-mini

MAX_LLM_CONCURRENCY=50
REQUEST_RATE_LIMIT=100

配置读取文件 app/config.py：

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    APP_NAME: str = "AI_AGENT_SERVICE"
    ENV: str = "production"

    REDIS_HOST: str = "127.0.0.1"
    REDIS_PORT: int = 6379
    REDIS_DB: int = 0

    POSTGRES_HOST: str = "127.0.0.1"
    POSTGRES_PORT: int = 5432
    POSTGRES_DB: str = "agentdb"
    POSTGRES_USER: str = "agentuser"
    POSTGRES_PASSWORD: str = "agentpassword"

    LLM_API_KEY: str
    LLM_BASE_URL: str = "https://api.openai.com/v1"
    LLM_MODEL: str = "gpt-4o-mini"

    MAX_LLM_CONCURRENCY: int = 50
    REQUEST_RATE_LIMIT: int = 100

    class Config:
        env_file = ".env"


settings = Settings()

七、安装 Redis 与 PostgreSQL

1. 安装 Redis

sudo apt install -y redis-server

启动并设置开机自启：

sudo systemctl enable redis-server
sudo systemctl start redis-server

检查 Redis 状态：

sudo systemctl status redis-server

测试连接：

redis-cli ping

如果返回：

PONG

说明 Redis 正常。

2. 安装 PostgreSQL

sudo apt install -y postgresql postgresql-contrib

启动 PostgreSQL：

sudo systemctl enable postgresql
sudo systemctl start postgresql

创建数据库和用户：

sudo -u postgres psql

进入 PostgreSQL 后执行：

CREATE DATABASE agentdb;
CREATE USER agentuser WITH PASSWORD 'agentpassword';
GRANT ALL PRIVILEGES ON DATABASE agentdb TO agentuser;
\q

测试连接：

psql -h 127.0.0.1 -U agentuser -d agentdb

八、应用层限流方案

高并发系统必须具备限流能力。否则突发流量会直接打爆模型接口、数据库或第三方服务。

常见限流策略包括：

这里使用 Redis 实现简单的滑动窗口限流。

app/limiter.py：

import time
import redis
from app.config import settings


redis_client = redis.Redis(
    host=settings.REDIS_HOST,
    port=settings.REDIS_PORT,
    db=settings.REDIS_DB,
    decode_responses=True,
)


def is_allowed(key: str, limit: int, window: int = 60) -> bool:
    """
    Redis 滑动窗口限流
    key: 限流对象，例如 IP、用户 ID
    limit: 时间窗口内最大请求数
    window: 时间窗口，单位秒
    """
    now = time.time()
    pipeline = redis_client.pipeline()

    redis_key = f"rate_limit:{key}"

    pipeline.zremrangebyscore(redis_key, 0, now - window)
    pipeline.zadd(redis_key, {str(now): now})
    pipeline.zcard(redis_key)
    pipeline.expire(redis_key, window)

    _, _, count, _ = pipeline.execute()

    return count <= limit

九、缓存重复请求，降低 LLM 成本

对于相同问题、相同上下文、相同工具参数的请求，可以使用 Redis 缓存结果，避免重复调用大模型。

app/cache.py：

import hashlib
import json
import redis
from app.config import settings


redis_client = redis.Redis(
    host=settings.REDIS_HOST,
    port=settings.REDIS_PORT,
    db=settings.REDIS_DB,
    decode_responses=True,
)


def make_cache_key(payload: dict) -> str:
    raw = json.dumps(payload, sort_keys=True, ensure_ascii=False)
    digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()
    return f"agent_cache:{digest}"


def get_cache(payload: dict):
    key = make_cache_key(payload)
    return redis_client.get(key)


def set_cache(payload: dict, value: str, ttl: int = 300):
    key = make_cache_key(payload)
    redis_client.setex(key, ttl, value)

缓存建议：

1. 短 TTL 缓存：适合普通问答，TTL 可设为 1～10 分钟。
2. 长 TTL 缓存：适合知识库检索结果、工具元数据。
3. 不要缓存敏感数据：如用户隐私、交易信息。
4. 缓存 Key 要包含上下文版本：避免旧知识污染新结果。

十、控制 LLM 并发，避免 API 限流

模型供应商通常会限制：

• 每分钟请求数 RPM
• 每分钟 Token 数 TPM
• 并发连接数
• 单次请求最大 Token

如果应用不控制并发，可能出现大量 429 Too Many Requests。

在 Python 中可以使用 asyncio.Semaphore 控制模型调用并发。

app/agent.py：

import asyncio
import httpx
from app.config import settings


llm_semaphore = asyncio.Semaphore(settings.MAX_LLM_CONCURRENCY)


async def call_llm(prompt: str) -> str:
    async with llm_semaphore:
        async with httpx.AsyncClient(timeout=60) as client:
            response = await client.post(
                f"{settings.LLM_BASE_URL}/chat/completions",
                headers={
                    "Authorization": f"Bearer {settings.LLM_API_KEY}",
                    "Content-Type": "application/json",
                },
                json={
                    "model": settings.LLM_MODEL,
                    "messages": [
                        {"role": "system", "content": "你是一个可靠的 AI Agent。"},
                        {"role": "user", "content": prompt},
                    ],
                    "temperature": 0.2,
                },
            )
            response.raise_for_status()
            data = response.json()
            return data["choices"][0]["message"]["content"]


async def run_agent(user_input: str) -> str:
    """
    简化版 Agent 执行逻辑。
    生产环境可在这里加入工具调用、RAG、任务规划等模块。
    """
    planning_prompt = f"""
你是一个任务规划 Agent，请分析用户需求并给出执行计划。
用户需求：{user_input}
"""

    plan = await call_llm(planning_prompt)

    final_prompt = f"""
根据以下执行计划，生成最终答复。

用户需求：
{user_input}

执行计划：
{plan}
"""

    result = await call_llm(final_prompt)

    return result

注意：这里每次调用都新建 httpx.AsyncClient，为了示例简单。生产环境建议在应用生命周期内复用 HTTP 连接池，这样可以减少 TCP 握手和 TLS 握手成本。

十一、FastAPI 主服务

app/main.py：

from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel
from app.agent import run_agent
from app.limiter import is_allowed
from app.cache import get_cache, set_cache
from app.config import settings
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
from fastapi.responses import Response
import time


app = FastAPI(title=settings.APP_NAME)

REQUEST_COUNT = Counter(
    "agent_request_count",
    "Total Agent requests",
    ["endpoint", "status"]
)

REQUEST_LATENCY = Histogram(
    "agent_request_latency_seconds",
    "Agent request latency",
    ["endpoint"]
)


class AgentRequest(BaseModel):
    user_id: str
    query: str


@app.get("/health")
async def health():
    return {"status": "ok"}


@app.get("/metrics")
async def metrics():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)


@app.post("/agent")
async def agent_api(req: AgentRequest, request: Request):
    start_time = time.time()

    client_ip = request.client.host
    limit_key = f"{req.user_id}:{client_ip}"

    if not is_allowed(limit_key, settings.REQUEST_RATE_LIMIT, 60):
        REQUEST_COUNT.labels(endpoint="/agent", status="limited").inc()
        raise HTTPException(status_code=429, detail="Too Many Requests")

    payload = req.model_dump()
    cached = get_cache(payload)

    if cached:
        REQUEST_COUNT.labels(endpoint="/agent", status="cache_hit").inc()
        return {
            "source": "cache",
            "answer": cached,
        }

    try:
        answer = await run_agent(req.query)
        set_cache(payload, answer, ttl=300)

        REQUEST_COUNT.labels(endpoint="/agent", status="success").inc()
        REQUEST_LATENCY.labels(endpoint="/agent").observe(time.time() - start_time)

        return {
            "source": "llm",
            "answer": answer,
        }

    except Exception as e:
        REQUEST_COUNT.labels(endpoint="/agent", status="error").inc()
        raise HTTPException(status_code=500, detail=str(e))

启动测试：

uvicorn app.main:app --host 0.0.0.0 --port 8000

测试接口：

curl -X POST "http://127.0.0.1:8000/agent" \
  -H "Content-Type: application/json" \
  -d '{"user_id":"u001","query":"请帮我制定一个三天的北京旅行计划"}'

十二、使用 Gunicorn 提升并发能力

生产环境不建议直接使用单进程 Uvicorn。可以使用 Gunicorn 管理多个 Worker。

创建 gunicorn_conf.py：

bind = "0.0.0.0:8000"

workers = 4
worker_class = "uvicorn.workers.UvicornWorker"

timeout = 120
keepalive = 30

max_requests = 5000
max_requests_jitter = 500

accesslog = "-"
errorlog = "-"
loglevel = "info"

启动：

gunicorn app.main:app -c gunicorn_conf.py

Worker 数量一般可按如下公式估算：

workers = CPU 核心数 * 2 + 1

但对于 AI Agent 服务而言，由于大量时间消耗在网络 I/O，例如 LLM API、数据库、工具调用，所以可以适当增加 Worker 或使用异步模型。不过要注意，Worker 越多不代表越好，过多 Worker 会导致上下文切换增加、内存占用上升、连接池压力变大。

十三、长任务异步化：Celery 队列方案

如果 Agent 任务耗时超过 10 秒，建议不要让 HTTP 请求一直等待，而是采用异步任务模式：

用户提交任务 -> 返回 task_id
后台执行 Agent -> 用户轮询结果 / WebSocket 推送

1. 配置 Celery

app/tasks.py：

from celery import Celery
import asyncio
import redis
from app.agent import run_agent
from app.config import settings


celery_app = Celery(
    "agent_tasks",
    broker=f"redis://{settings.REDIS_HOST}:{settings.REDIS_PORT}/{settings.REDIS_DB}",
    backend=f"redis://{settings.REDIS_HOST}:{settings.REDIS_PORT}/{settings.REDIS_DB}",
)


celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="Asia/Shanghai",
    enable_utc=False,
    worker_concurrency=8,
    task_acks_late=True,
    task_reject_on_worker_lost=True,
)


@celery_app.task(name="run_agent_task")
def run_agent_task(user_input: str):
    return asyncio.run(run_agent(user_input))

启动 Celery Worker：

celery -A app.tasks.celery_app worker \
  --loglevel=info \
  --concurrency=8

如果希望多队列：

celery -A app.tasks.celery_app worker \
  --loglevel=info \
  --concurrency=8 \
  -Q agent_high,agent_normal,agent_low

2. 在 FastAPI 中提交异步任务

可在 app/main.py 中增加接口：

from app.tasks import run_agent_task

新增接口：

@app.post("/agent/async")
async def agent_async_api(req: AgentRequest):
    task = run_agent_task.delay(req.query)
    return {
        "task_id": task.id,
        "status": "submitted",
    }


@app.get("/agent/task/{task_id}")
async def get_task_result(task_id: str):
    task = run_agent_task.AsyncResult(task_id)

    return {
        "task_id": task_id,
        "status": task.status,
        "result": task.result if task.ready() else None,
    }

测试异步任务提交：

curl -X POST "http://127.0.0.1:8000/agent/async" \
  -H "Content-Type: application/json" \
  -d '{"user_id":"u001","query":"请生成一份市场调研报告"}'

查询任务结果：

curl "http://127.0.0.1:8000/agent/task/你的task_id"

十四、Nginx 反向代理与入口限流

安装 Nginx：

sudo apt install -y nginx

创建配置：

sudo vim /etc/nginx/sites-available/ai-agent

写入：

limit_req_zone $binary_remote_addr zone=agent_limit:10m rate=20r/s;

server {
    listen 80;
    server_name your-domain.com;

    client_max_body_size 10m;

    location / {
        limit_req zone=agent_limit burst=40 nodelay;

        proxy_pass http://127.0.0.1:8000;

        proxy_http_version 1.1;
        proxy_set_header Host $host;
        proxy_set_header Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        proxy_connect_timeout 30s;
        proxy_send_timeout 120s;
        proxy_read_timeout 120s;
    }
}

启用配置：

sudo ln -s /etc/nginx/sites-available/ai-agent /etc/nginx/sites-enabled/
sudo nginx -t
sudo systemctl reload nginx

如果已有默认站点冲突：

sudo rm -f /etc/nginx/sites-enabled/default
sudo nginx -t
sudo systemctl reload nginx

十五、使用 systemd 托管服务

创建 Gunicorn 服务：

sudo vim /etc/systemd/system/ai-agent.service

写入：

[Unit]
Description=AI Agent FastAPI Service
After=network.target redis-server.service postgresql.service

[Service]
User=ubuntu
Group=ubuntu
WorkingDirectory=/home/ubuntu/ai-agent-high-concurrency
Environment="PATH=/home/ubuntu/ai-agent-high-concurrency/.venv/bin"
ExecStart=/home/ubuntu/ai-agent-high-concurrency/.venv/bin/gunicorn app.main:app -c gunicorn_conf.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

启动服务：

sudo systemctl daemon-reload
sudo systemctl enable ai-agent
sudo systemctl start ai-agent

查看状态：

sudo systemctl status ai-agent

查看日志：

journalctl -u ai-agent -f

创建 Celery 服务：

sudo vim /etc/systemd/system/ai-agent-celery.service

写入：

[Unit]
Description=AI Agent Celery Worker
After=network.target redis-server.service

[Service]
User=ubuntu
Group=ubuntu
WorkingDirectory=/home/ubuntu/ai-agent-high-concurrency
Environment="PATH=/home/ubuntu/ai-agent-high-concurrency/.venv/bin"
ExecStart=/home/ubuntu/ai-agent-high-concurrency/.venv/bin/celery -A app.tasks.celery_app worker --loglevel=info --concurrency=8
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

启动：

sudo systemctl daemon-reload
sudo systemctl enable ai-agent-celery
sudo systemctl start ai-agent-celery

查看日志：

journalctl -u ai-agent-celery -f

十六、Docker Compose 一键部署

如果希望更方便部署，可以使用 Docker Compose。

创建 Dockerfile：

vim Dockerfile

写入：

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .

RUN pip install --no-cache-dir --upgrade pip \
    && pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["gunicorn", "app.main:app", "-c", "gunicorn_conf.py"]

创建 docker-compose.yml：

vim docker-compose.yml

写入：

version: "3.9"

services:
  api:
    build: .
    container_name: ai-agent-api
    ports:
      - "8000:8000"
    env_file:
      - .env
    depends_on:
      - redis
      - postgres
    restart: always

  celery:
    build: .
    container_name: ai-agent-celery
    command: celery -A app.tasks.celery_app worker --loglevel=info --concurrency=8
    env_file:
      - .env
    depends_on:
      - redis
    restart: always

  redis:
    image: redis:7
    container_name: ai-agent-redis
    ports:
      - "6379:6379"
    restart: always

  postgres:
    image: postgres:16
    container_name: ai-agent-postgres
    environment:
      POSTGRES_DB: agentdb
      POSTGRES_USER: agentuser
      POSTGRES_PASSWORD: agentpassword
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
    restart: always

volumes:
  postgres_data:

启动：

docker compose up -d --build

查看容器：

docker compose ps

查看日志：

docker compose logs -f api
docker compose logs -f celery

停止：

docker compose down

删除数据卷：

docker compose down -v

十七、压测 Agent 服务

高并发方案上线前必须压测。推荐使用 wrk 或 hey。

1. 安装 wrk

sudo apt install -y wrk

创建压测脚本：

vim post.lua

写入：

wrk.method = "POST"
wrk.headers["Content-Type"] = "application/json"
wrk.body = '{"user_id":"bench_user","query":"请解释什么是 AI Agent 高并发架构"}'

执行压测：

wrk -t4 -c100 -d60s -s post.lua http://127.0.0.1:8000/agent

参数说明：

-t4：4 个线程
-c100：100 个并发连接
-d60s：持续 60 秒

2. 使用 hey 压测

安装：

wget https://hey-release.s3.us-east-2.amazonaws.com/hey_linux_amd64
chmod +x hey_linux_amd64
sudo mv hey_linux_amd64 /usr/local/bin/hey

执行：

hey -n 1000 -c 100 \
  -m POST \
  -H "Content-Type: application/json" \
  -d '{"user_id":"bench_user","query":"请解释什么是 AI Agent 高并发架构"}' \
  http://127.0.0.1:8000/agent

压测时重点关注：

1. 平均响应时间。
2. P95、P99 延迟。
3. QPS。
4. 错误率。
5. Redis CPU 和内存。
6. PostgreSQL 连接数。
7. LLM API 429 次数。
8. Celery 队列积压长度。
9. Nginx 连接数。
10. 应用服务内存是否持续上涨。

十八、监控与告警

1. 查看 Prometheus 指标

访问：

http://服务器IP:8000/metrics

可以看到类似指标：

agent_request_count_total
agent_request_latency_seconds_bucket
agent_request_latency_seconds_sum
agent_request_latency_seconds_count

2. Docker 启动 Prometheus 和 Grafana

创建 monitoring/docker-compose.yml：

mkdir -p monitoring
vim monitoring/docker-compose.yml

写入：

version: "3.9"

services:
  prometheus:
    image: prom/prometheus
    container_name: prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    restart: always

  grafana:
    image: grafana/grafana
    container_name: grafana
    ports:
      - "3000:3000"
    restart: always

创建 Prometheus 配置：

vim monitoring/prometheus.yml

写入：

global:
  scrape_interval: 15s

scrape_configs:
  - job_name: "ai-agent"
    static_configs:
      - targets: ["host.docker.internal:8000"]

如果是在 Linux Docker 环境中，host.docker.internal 可能不可用，可以改成宿主机 IP，例如：

targets: ["172.17.0.1:8000"]

启动监控：

cd monitoring
docker compose up -d

访问：

Prometheus：http://服务器IP:9090
Grafana：http://服务器IP:3000

Grafana 默认账号密码通常是：

admin / admin

十九、关键优化策略总结

1. 请求分级

不要所有请求都走同一条通道。可以分为：

普通问答：同步返回
复杂任务：异步队列
批量任务：低优先级队列
VIP 用户：高优先级队列

2. Agent 步数限制

Agent 容易陷入循环，因此必须限制最大执行步数：

max_steps = 5
max_tool_calls = 10
max_tokens = 4096
timeout = 60s

3. 工具调用超时

每个工具都要设置超时：

数据库查询超时：3 秒
外部 API 超时：5 秒
搜索工具超时：10 秒
LLM 调用超时：60 秒

4. 熔断与降级

当外部模型或工具不可用时，需要降级：

GPT-4 不可用 -> 切换 GPT-4o-mini
主向量库不可用 -> 使用关键词检索
实时搜索失败 -> 返回已有知识库答案
异步任务积压 -> 暂停低优先级任务

5. Token 成本控制

高并发下 Token 成本会迅速放大，应做：

1. 压缩历史对话。
2. 控制上下文窗口。
3. 对检索文档做摘要。
4. 减少无意义工具调用。
5. 对短问题使用小模型。
6. 对复杂任务才使用强模型。

6. 数据库连接池

数据库连接池不能无限放大。建议：

每个 Worker 连接池大小：5～20
总连接数 < 数据库最大连接数
使用 PgBouncer 管理 PostgreSQL 连接

安装 PgBouncer：

sudo apt install -y pgbouncer

查看 PostgreSQL 最大连接数：

sudo -u postgres psql -c "SHOW max_connections;"

7. Redis 内存策略

Redis 用于缓存和限流时，必须设置内存上限和淘汰策略：

sudo vim /etc/redis/redis.conf

设置：

maxmemory 1gb
maxmemory-policy allkeys-lru

重启 Redis：

sudo systemctl restart redis-server

二十、推荐生产参数

以下是一组中小规模 AI Agent 服务的推荐参数：

二十一、最终落地清单

上线前建议逐项检查：

• [x] Nginx 已配置入口限流。
• [x] FastAPI 已配置用户级限流。
• [x] Redis 缓存已启用。
• [x] LLM 并发已通过 Semaphore 控制。
• [x] 长任务已接入 Celery。
• [x] Gunicorn 多 Worker 已启用。
• [x] 数据库连接池已设置上限。
• [x] 外部工具调用已设置超时。
• [x] Agent 最大执行步数已限制。
• [x] 已配置健康检查接口。
• [x] 已暴露 Prometheus 指标。
• [x] 已完成压测。
• [x] 已配置 systemd 或 Docker Compose 守护。
• [x] 已准备降级和熔断策略。

结语

AI Agent 的高并发问题，本质上不是单点性能问题，而是一个完整的工程系统问题。真正稳定的 Agent 服务，需要同时处理模型调用、工具调用、数据库访问、缓存、队列、限流、监控和成本控制。

如果只是简单增加服务器，短期可能能缓解压力，但无法解决执行链路放大、模型限流、任务阻塞和雪崩重试等问题。更合理的做法是：

入口限流 + 异步队列 + 缓存复用 + 并发控制 + 横向扩容 + 监控告警 + 熔断降级

对于大多数生产环境，这套方案已经可以支撑从几十并发到数千并发的 AI Agent 服务。当然，最终参数仍需要根据业务场景、模型供应商限额、Agent 复杂度和服务器配置进行压测调优。