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

在企业落地 AI 工具的过程中，很多团队都会遇到同一个问题：模型能力看起来很强，但一旦接入真实业务流量，高并发场景下就会出现响应慢、请求堆积、接口超时、成本飙升，甚至服务不可用。

尤其是基于大语言模型的 AI 工具，例如智能客服、知识库问答、AI 写作、代码助手、数据分析 Agent、企业内部 Copilot 等，它们和传统 Web 系统相比，有几个非常明显的差异：

• 单次请求耗时更长；
• Token 消耗不可控；
• 上游模型接口存在速率限制；
• 推理服务资源消耗高；
• 流式响应连接占用时间长；
• 多轮对话需要上下文存储；
• 用户高峰期请求会突然集中爆发。

因此，AI 工具的高并发架构不能简单照搬传统 CRUD 系统的设计，而需要围绕 异步化、限流、缓存、队列、模型网关、任务调度、弹性扩容、可观测性 等能力进行整体设计。

本文将从架构设计、服务拆分、并发控制、缓存策略、消息队列、Nginx 反向代理、Docker 部署、Redis 限流、FastAPI 示例、Celery 异步任务、压测命令等多个方面，给出一套可落地的 AI 工具高并发解决方案，并附上完整命令。

一、AI工具高并发的核心挑战

在设计方案之前，必须先理解 AI 工具为什么容易在高并发下出现问题。

1. 单次请求耗时长

传统接口可能几十毫秒到几百毫秒返回，但 AI 请求通常需要：

• 解析用户输入；
• 检索知识库；
• 构造 Prompt；
• 调用大模型；
• 等待模型生成；
• 流式返回；
• 保存对话记录。

一个完整链路可能耗时 2 秒、5 秒，甚至 30 秒以上。

如果 1000 个用户同时访问，每个请求平均占用连接 10 秒，那么服务端需要同时维持大量长连接，对 Web 服务、网关、线程池、数据库连接池都会产生压力。

2. 上游模型接口有限流

如果使用 OpenAI、Claude、通义千问、智谱、DeepSeek、火山方舟等模型 API，通常会有：

• QPS 限制；
• TPM 限制，即每分钟 Token 数；
• RPM 限制，即每分钟请求数；
• 并发连接数限制；
• 账号级别额度限制。

当用户请求超过上游模型限制时，如果没有本地限流和队列机制，系统会出现大量错误，例如：

429 Too Many Requests
Rate limit exceeded
Quota exceeded
Upstream timeout

3. Token 成本难以控制

AI 工具的请求成本不只取决于请求次数，还取决于输入和输出 Token 数。

高并发场景下，如果不控制上下文长度、不做缓存、不做 Prompt 压缩，成本会快速放大。

例如：

• 1000 次请求；
• 每次输入 3000 tokens；
• 每次输出 1000 tokens；
• 总共就是 400 万 tokens。

如果再叠加多轮对话和 RAG 检索内容，成本会更高。

4. 流式响应占用连接资源

AI 应用常用 Server-Sent Events，即 SSE，或者 WebSocket 实现流式输出。

流式输出体验很好，但问题是：连接保持时间长。

传统接口请求返回后连接释放，而流式接口可能保持十几秒甚至几十秒。并发用户越多，服务器需要维持的连接越多。

因此必须合理配置：

• Nginx 超时时间；
• 后端服务 worker 数；
• keepalive；
• 最大连接数；
• 文件描述符；
• 操作系统网络参数。

5. 任务类型复杂

AI 工具不全是即时问答，还可能包含：

• 文档解析；
• 向量化入库；
• 批量总结；
• 长文本生成；
• 图片生成；
• 音频转写；
• 视频理解；
• 自动报表生成。

这些任务不适合直接在 HTTP 请求中同步完成，应该进入异步任务队列，由后台 worker 消费。

二、整体架构设计

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

用户
  │
  ▼
CDN / WAF
  │
  ▼
Nginx / API Gateway
  │
  ├── 静态资源服务
  │
  ├── 鉴权 / 限流 / 熔断
  │
  ▼
AI 应用服务 FastAPI / Node.js / Go
  │
  ├── Redis 缓存
  │
  ├── Redis 限流
  │
  ├── PostgreSQL / MySQL
  │
  ├── Milvus / Qdrant / pgvector
  │
  ├── Celery / RabbitMQ / Kafka
  │
  ▼
模型网关 Model Gateway
  │
  ├── OpenAI
  │
  ├── DeepSeek
  │
  ├── Qwen
  │
  ├── Claude
  │
  ├── 本地 vLLM / Ollama / TGI
  ▼
大模型服务

这套架构的关键思想是：

1. 入口层统一限流：避免恶意流量和突发流量打爆后端；
2. 应用层轻量化：不要让 Web 请求承担重任务；
3. 缓存优先：相同问题、相同知识库、相同 Prompt 尽量命中缓存；
4. 异步队列削峰填谷：长任务放入队列；
5. 模型网关统一调度：多模型、多供应商统一路由；
6. 可观测性贯穿全链路：监控请求耗时、错误率、Token 消耗和队列堆积。

三、推荐技术选型

下面是一套比较通用的技术组合：

如果是中小型团队，建议先用：

Nginx + FastAPI + Redis + PostgreSQL + Celery + Docker Compose

这套组合足够支撑大多数 AI 工具的早期和中期并发需求。

四、服务器基础优化命令

在部署高并发 AI 服务前，建议先调整 Linux 系统参数。

1. 查看系统版本

cat /etc/os-release
uname -a

2. 查看 CPU 和内存

lscpu
free -h
df -h

3. 调整文件描述符限制

AI 流式接口会占用大量连接，因此需要提高文件描述符上限。

ulimit -n

临时设置：

ulimit -n 65535

永久设置：

sudo vim /etc/security/limits.conf

追加：

* soft nofile 65535
* hard nofile 65535
root soft nofile 65535
root hard nofile 65535

修改 systemd 配置：

sudo vim /etc/systemd/system.conf

添加或修改：

DefaultLimitNOFILE=65535

重新加载：

sudo systemctl daemon-reexec

4. 优化内核网络参数

编辑配置文件：

sudo vim /etc/sysctl.conf

追加：

net.core.somaxconn = 65535
net.core.netdev_max_backlog = 65535
net.ipv4.tcp_max_syn_backlog = 65535
net.ipv4.ip_local_port_range = 1024 65000
net.ipv4.tcp_tw_reuse = 1
net.ipv4.tcp_fin_timeout = 15
net.ipv4.tcp_keepalive_time = 600
net.ipv4.tcp_keepalive_intvl = 30
net.ipv4.tcp_keepalive_probes = 5

使配置生效：

sudo sysctl -p

查看是否生效：

sysctl net.core.somaxconn
sysctl net.ipv4.tcp_fin_timeout

五、安装 Docker 与 Docker Compose

推荐用 Docker 部署，方便扩容、迁移和统一环境。

1. 安装 Docker

Ubuntu 系统命令如下：

sudo apt update
sudo apt install -y ca-certificates curl gnupg lsb-release

添加 Docker GPG key：

sudo mkdir -p /etc/apt/keyrings
curl -fsSL https://download.docker.com/linux/ubuntu/gpg | \
sudo gpg --dearmor -o /etc/apt/keyrings/docker.gpg

添加 Docker 软件源：

echo \
"deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.gpg] \
https://download.docker.com/linux/ubuntu \
$(lsb_release -cs) stable" | \
sudo tee /etc/apt/sources.list.d/docker.list > /dev/null

安装 Docker：

sudo apt update
sudo apt install -y docker-ce docker-ce-cli containerd.io docker-compose-plugin

启动 Docker：

sudo systemctl enable docker
sudo systemctl start docker

查看版本：

docker version
docker compose version

将当前用户加入 docker 组：

sudo usermod -aG docker $USER
newgrp docker

六、项目目录结构

创建项目目录：

mkdir -p ai-high-concurrency/{app,nginx,worker}
cd ai-high-concurrency

推荐目录结构：

ai-high-concurrency
├── app
│   ├── main.py
│   ├── requirements.txt
│   └── Dockerfile
├── worker
│   ├── tasks.py
│   ├── requirements.txt
│   └── Dockerfile
├── nginx
│   └── nginx.conf
├── docker-compose.yml
└── .env

七、FastAPI 应用服务示例

进入 app 目录：

cd app

创建依赖文件：

cat > requirements.txt <<'EOF'
fastapi==0.115.6
uvicorn[standard]==0.34.0
redis==5.2.1
httpx==0.28.1
pydantic==2.10.4
python-dotenv==1.0.1
EOF

创建主程序：

cat > main.py <<'EOF'
import os
import time
import hashlib
import asyncio
from typing import Optional

import httpx
import redis
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

REDIS_URL = os.getenv("REDIS_URL", "redis://redis:6379/0")
MODEL_API_URL = os.getenv("MODEL_API_URL", "https://api.openai.com/v1/chat/completions")
MODEL_API_KEY = os.getenv("MODEL_API_KEY", "")
MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o-mini")

r = redis.from_url(REDIS_URL, decode_responses=True)

app = FastAPI(title="AI High Concurrency Demo")


class ChatRequest(BaseModel):
    user_id: str
    message: str
    stream: Optional[bool] = True


def make_cache_key(message: str) -> str:
    digest = hashlib.sha256(message.encode("utf-8")).hexdigest()
    return f"ai:cache:{digest}"


def check_rate_limit(user_id: str, limit: int = 20, window: int = 60):
    key = f"rate:{user_id}:{int(time.time() // window)}"
    current = r.incr(key)
    if current == 1:
        r.expire(key, window)
    if current > limit:
        raise HTTPException(status_code=429, detail="请求过于频繁，请稍后再试")


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


@app.post("/chat")
async def chat(req: ChatRequest):
    check_rate_limit(req.user_id)

    cache_key = make_cache_key(req.message)
    cached = r.get(cache_key)
    if cached:
        return {"source": "cache", "answer": cached}

    if req.stream:
        return StreamingResponse(
            stream_model_response(req.message, cache_key),
            media_type="text/event-stream"
        )

    answer = await call_model(req.message)
    r.setex(cache_key, 3600, answer)
    return {"source": "model", "answer": answer}


async def call_model(message: str) -> str:
    headers = {
        "Authorization": f"Bearer {MODEL_API_KEY}",
        "Content-Type": "application/json"
    }

    payload = {
        "model": MODEL_NAME,
        "messages": [
            {"role": "system", "content": "你是一个专业、简洁、可靠的AI助手。"},
            {"role": "user", "content": message}
        ],
        "temperature": 0.7,
        "max_tokens": 800,
        "stream": False
    }

    timeout = httpx.Timeout(60.0, connect=10.0)

    async with httpx.AsyncClient(timeout=timeout) as client:
        resp = await client.post(MODEL_API_URL, headers=headers, json=payload)
        resp.raise_for_status()
        data = resp.json()
        return data["choices"][0]["message"]["content"]


async def stream_model_response(message: str, cache_key: str):
    headers = {
        "Authorization": f"Bearer {MODEL_API_KEY}",
        "Content-Type": "application/json"
    }

    payload = {
        "model": MODEL_NAME,
        "messages": [
            {"role": "system", "content": "你是一个专业、简洁、可靠的AI助手。"},
            {"role": "user", "content": message}
        ],
        "temperature": 0.7,
        "max_tokens": 800,
        "stream": True
    }

    full_text = []

    timeout = httpx.Timeout(120.0, connect=10.0)

    async with httpx.AsyncClient(timeout=timeout) as client:
        async with client.stream("POST", MODEL_API_URL, headers=headers, json=payload) as resp:
            resp.raise_for_status()
            async for line in resp.aiter_lines():
                if not line:
                    continue

                if line.startswith("data: "):
                    data = line.replace("data: ", "")

                    if data.strip() == "[DONE]":
                        break

                    full_text.append(data)
                    yield f"data: {data}\n\n"

    r.setex(cache_key, 3600, "".join(full_text))
    yield "data: [DONE]\n\n"
EOF

创建 Dockerfile：

cat > Dockerfile <<'EOF'
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

COPY main.py .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]
EOF

回到根目录：

cd ..

这个 FastAPI 示例实现了几个关键能力：

• 健康检查；
• 用户级 Redis 限流；
• 相同问题缓存；
• 支持普通响应；
• 支持 SSE 流式响应；
• 通过环境变量配置模型接口。

八、Celery 异步任务队列

对于耗时任务，例如批量生成、文档解析、知识库入库，不建议直接走同步接口。可以使用 Celery 加 Redis 做异步任务。

进入 worker 目录：

cd worker

创建依赖：

cat > requirements.txt <<'EOF'
celery==5.4.0
redis==5.2.1
httpx==0.28.1
python-dotenv==1.0.1
EOF

创建任务文件：

cat > tasks.py <<'EOF'
import os
import httpx
from celery import Celery

REDIS_URL = os.getenv("REDIS_URL", "redis://redis:6379/1")
MODEL_API_URL = os.getenv("MODEL_API_URL", "https://api.openai.com/v1/chat/completions")
MODEL_API_KEY = os.getenv("MODEL_API_KEY", "")
MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o-mini")

celery_app = Celery(
    "ai_worker",
    broker=REDIS_URL,
    backend=REDIS_URL
)


@celery_app.task(
    name="tasks.long_generate",
    autoretry_for=(Exception,),
    retry_kwargs={"max_retries": 3, "countdown": 5}
)
def long_generate(prompt: str):
    headers = {
        "Authorization": f"Bearer {MODEL_API_KEY}",
        "Content-Type": "application/json"
    }

    payload = {
        "model": MODEL_NAME,
        "messages": [
            {"role": "system", "content": "你是一个专业的长文本生成助手。"},
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.7,
        "max_tokens": 2000,
        "stream": False
    }

    with httpx.Client(timeout=120.0) as client:
        resp = client.post(MODEL_API_URL, headers=headers, json=payload)
        resp.raise_for_status()
        data = resp.json()
        return data["choices"][0]["message"]["content"]
EOF

创建 Dockerfile：

cat > Dockerfile <<'EOF'
FROM python:3.11-slim

WORKDIR /worker

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

COPY tasks.py .

CMD ["celery", "-A", "tasks.celery_app", "worker", "--loglevel=info", "--concurrency=4"]
EOF

回到根目录：

cd ..

Celery 的作用是把高耗时、高成本任务从 Web 请求链路中拆出来，实现削峰填谷。

九、Nginx 高并发配置

创建 Nginx 配置：

cat > nginx/nginx.conf <<'EOF'
worker_processes auto;

events {
    worker_connections 65535;
    use epoll;
    multi_accept on;
}

http {
    include       mime.types;
    default_type  application/octet-stream;

    sendfile on;
    tcp_nopush on;
    tcp_nodelay on;

    keepalive_timeout 65;
    keepalive_requests 10000;

    client_max_body_size 20m;
    client_body_timeout 30s;
    send_timeout 120s;

    proxy_connect_timeout 10s;
    proxy_send_timeout 120s;
    proxy_read_timeout 120s;

    upstream ai_backend {
        least_conn;
        server app:8000 max_fails=3 fail_timeout=30s;
        keepalive 128;
    }

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

    server {
        listen 80;
        server_name _;

        location /health {
            proxy_pass http://ai_backend;
        }

        location /chat {
            limit_req zone=ip_limit burst=20 nodelay;

            proxy_pass http://ai_backend;
            proxy_http_version 1.1;

            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_buffering off;
            proxy_cache off;
            chunked_transfer_encoding on;

            proxy_set_header Connection "";
        }
    }
}
EOF

这份配置重点解决：

• Nginx 多进程；
• 高连接数；
• upstream 长连接；
• IP 级限流；
• 流式响应关闭 buffering；
• 延长模型调用读取超时；
• 反向代理到 FastAPI 服务。

十、Docker Compose 完整部署文件

创建 .env 文件：

cat > .env <<'EOF'
MODEL_API_URL=https://api.openai.com/v1/chat/completions
MODEL_API_KEY=你的模型API_KEY
MODEL_NAME=gpt-4o-mini
REDIS_URL=redis://redis:6379/0
EOF

创建 docker-compose.yml：

cat > docker-compose.yml <<'EOF'
services:
  redis:
    image: redis:7.4
    container_name: ai_redis
    command: redis-server --appendonly yes --maxmemory 1gb --maxmemory-policy allkeys-lru
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    restart: always

  app:
    build:
      context: ./app
    container_name: ai_app
    env_file:
      - .env
    depends_on:
      - redis
    expose:
      - "8000"
    restart: always
    deploy:
      resources:
        limits:
          cpus: "2"
          memory: 2G

  worker:
    build:
      context: ./worker
    container_name: ai_worker
    env_file:
      - .env
    depends_on:
      - redis
    restart: always

  nginx:
    image: nginx:1.27
    container_name: ai_nginx
    depends_on:
      - app
    ports:
      - "80:80"
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
    restart: always

volumes:
  redis_data:
EOF

启动服务：

docker compose up -d --build

查看容器状态：

docker compose ps

查看日志：

docker compose logs -f app
docker compose logs -f nginx
docker compose logs -f worker

测试健康检查：

curl http://127.0.0.1/health

测试普通请求：

curl -X POST http://127.0.0.1/chat \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "u001",
    "message": "请用三句话介绍什么是高并发架构",
    "stream": false
  }'

测试流式请求：

curl -N -X POST http://127.0.0.1/chat \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "u001",
    "message": "请详细介绍AI工具如何应对高并发",
    "stream": true
  }'

十一、服务水平扩容命令

当单个 app 容器无法承载更多请求时，可以水平扩容。

docker compose up -d --scale app=3

查看扩容后的服务：

docker compose ps

不过需要注意：如果使用固定 container_name，Docker Compose 在 scale 时会冲突。生产环境建议去掉 container_name 字段，然后执行：

docker compose up -d --scale app=4 --scale worker=4

如果修改了 compose 文件，重新部署：

docker compose down
docker compose up -d --build --scale app=4 --scale worker=4

查看资源使用情况：

docker stats

十二、Redis 限流策略设计

限流是 AI 工具高并发架构中非常关键的一环。推荐至少做三层限流：

1. IP 限流

在 Nginx 层做，防止单个 IP 过度请求。

limit_req_zone $binary_remote_addr zone=ip_limit:10m rate=10r/s;
limit_req zone=ip_limit burst=20 nodelay;

2. 用户限流

在应用层根据 user_id 限流，例如每分钟 20 次。

def check_rate_limit(user_id: str, limit: int = 20, window: int = 60):
    key = f"rate:{user_id}:{int(time.time() // window)}"
    current = r.incr(key)
    if current == 1:
        r.expire(key, window)
    if current > limit:
        raise HTTPException(status_code=429, detail="请求过于频繁，请稍后再试")

3. 模型限流

根据上游模型的 RPM 和 TPM 限制，建立全局令牌桶。

例如某模型每分钟最多允许 300 请求，就可以用 Redis 记录模型级请求次数：

redis-cli INCR model:qps:gpt-4o-mini
redis-cli EXPIRE model:qps:gpt-4o-mini 60

生产环境可进一步使用 Lua 脚本保证原子性。

十三、缓存策略：降低成本与延迟

AI 工具一定要重视缓存。常见缓存方式有三种：

1. 精确缓存

相同输入直接返回缓存结果：

用户问题 -> SHA256 -> Redis Key -> Answer

适合 FAQ、固定知识问答、工具类解释场景。

2. 语义缓存

将用户问题向量化后，在向量数据库中检索相似问题，如果相似度足够高，就复用之前答案。

适合客服问答、知识库问答、售前咨询等场景。

3. RAG 检索缓存

对于知识库问答，可以缓存：

• query 的向量；
• 检索出的文档片段；
• rerank 结果；
• 最终 Prompt；
• 模型回答。

这样可以显著降低向量检索和模型调用开销。

十四、队列削峰填谷方案

当请求超过模型处理能力时，不能无限制地直接转发给模型，而应该进入队列。

适合进入队列的任务包括：

• 长文生成；
• 批量翻译；
• 批量总结；
• 文档解析；
• 知识库入库；
• 报表生成；
• 图片生成；
• 音视频处理。

队列设计建议：

Celery 查看 worker 状态：

docker compose exec worker celery -A tasks.celery_app inspect active

查看注册任务：

docker compose exec worker celery -A tasks.celery_app inspect registered

十五、压测工具与完整命令

高并发方案是否有效，必须压测验证。

1. 安装 wrk

Ubuntu 安装：

sudo apt update
sudo apt install -y build-essential git
git clone https://github.com/wg/wrk.git
cd wrk
make
sudo cp wrk /usr/local/bin/
wrk -v

创建 POST 压测脚本：

cat > post.lua <<'EOF'
wrk.method = "POST"
wrk.body   = '{"user_id":"test_user","message":"请解释什么是AI高并发架构","stream":false}'
wrk.headers["Content-Type"] = "application/json"
EOF

执行压测：

wrk -t4 -c100 -d30s -s post.lua http://127.0.0.1/chat

参数含义：

• -t4：4 个线程；
• -c100：100 个并发连接；
• -d30s：持续 30 秒；
• -s post.lua：使用 Lua 脚本构造 POST 请求。

更高并发测试：

wrk -t8 -c500 -d60s -s post.lua http://127.0.0.1/chat

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":"u001","message":"请解释AI工具高并发方案","stream":false}' \
  http://127.0.0.1/chat

更高并发：

hey -n 10000 -c 500 -m POST \
  -H "Content-Type: application/json" \
  -d '{"user_id":"u002","message":"请写一段关于AI系统架构的说明","stream":false}' \
  http://127.0.0.1/chat

十六、监控与排障命令

高并发系统上线后，必须持续观察。

1. 查看容器资源

docker stats

2. 查看日志

docker compose logs -f --tail=200 nginx
docker compose logs -f --tail=200 app
docker compose logs -f --tail=200 worker

3. 查看 Redis 状态

docker compose exec redis redis-cli INFO memory
docker compose exec redis redis-cli INFO stats
docker compose exec redis redis-cli DBSIZE

查看慢查询：

docker compose exec redis redis-cli SLOWLOG GET 10

4. 查看连接情况

ss -s
ss -ant | awk '{print $1}' | sort | uniq -c

查看 80 端口连接：

ss -ant sport = :80

5. 查看系统负载

top
htop
uptime
vmstat 1
iostat -x 1

如果没有 htop 和 iostat，可安装：

sudo apt install -y htop sysstat

十七、生产环境关键优化建议

1. Web 服务不要无限增加 worker

很多人以为 worker 越多越好，其实不是。worker 太多会造成 CPU 上下文切换增加，内存占用升高。

对于 FastAPI + Uvicorn，一般可以从以下配置开始：

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

如果服务器是 8 核，可以尝试：

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 8

但最终以压测结果为准。

2. 流式请求与普通请求分离

建议将接口拆分为：

/chat
/chat-stream
/task
/task/result

普通请求走短连接，流式请求单独配置超时和连接数，避免流式请求拖垮普通 API。

3. 为不同用户设置不同限额

例如：

这不仅是技术问题，也是成本控制问题。

4. 增加熔断和降级

当上游模型异常时，系统不能一直阻塞等待。可采用：

• 超时快速失败；
• 自动切换备用模型；
• 返回缓存答案；
• 返回排队提示；
• 降低 max_tokens；
• 暂停低优先级任务。

5. 模型网关统一管理

生产环境强烈建议引入模型网关，统一处理：

• 多模型 API Key；
• 模型路由；
• 请求重试；
• 失败切换；
• Token 统计；
• 成本统计；
• 限流；
• 审计日志；
• Prompt 模板管理。

模型网关可以自研，也可以基于 LiteLLM 等工具搭建。

十八、常见问题与解决方案

问题 1：接口经常 504 Gateway Timeout

可能原因：

• Nginx proxy_read_timeout 太短；
• 模型响应慢；
• 后端 worker 被占满；
• 请求没有进入队列。

解决：

proxy_read_timeout 120s;
proxy_send_timeout 120s;

同时将长任务改为异步队列。

问题 2：出现大量 429

可能原因：

• Nginx 限流太严格；
• 应用层用户限流太严格；
• 上游模型 API 限流。

解决思路：

• 区分免费用户和付费用户；
• 增加排队机制；
• 切换多模型供应商；
• 增加缓存；
• 降低单请求 Token。

问题 3：Redis 内存上涨过快

可能原因：

• 缓存没有设置过期时间；
• 对话历史保存过长；
• 队列堆积；
• 缓存 key 设计不合理。

排查命令：

docker compose exec redis redis-cli INFO memory
docker compose exec redis redis-cli --bigkeys

解决：

redis-cli CONFIG SET maxmemory 1gb
redis-cli CONFIG SET maxmemory-policy allkeys-lru

问题 4：流式响应没有实时输出

可能原因：

• Nginx 开启了 proxy buffering；
• 客户端没有使用流式读取；
• 后端没有及时 flush。

Nginx 应配置：

proxy_buffering off;
proxy_cache off;
chunked_transfer_encoding on;

curl 测试必须加 -N：

curl -N -X POST http://127.0.0.1/chat \
  -H "Content-Type: application/json" \
  -d '{"user_id":"u001","message":"测试流式输出","stream":true}'

十九、上线前检查清单

上线 AI 高并发服务前，建议逐项检查：

• [ ] 是否配置 Nginx 限流；
• [ ] 是否配置用户级限流；
• [ ] 是否设置模型级限流；
• [ ] 是否设置请求超时；
• [ ] 是否对长任务使用队列；
• [ ] 是否设置 Redis 过期时间；
• [ ] 是否控制最大输入长度；
• [ ] 是否控制最大输出 Token；
• [ ] 是否记录 Token 消耗；
• [ ] 是否有模型失败重试；
• [ ] 是否有备用模型；
• [ ] 是否有错误日志；
• [ ] 是否有监控面板；
• [ ] 是否压测过 100、500、1000 并发；
• [ ] 是否准备降级方案。

二十、总结

AI 工具的高并发解决方案，本质上不是单点优化，而是一套完整的工程体系。

如果只增加服务器，很容易遇到模型限流、连接耗尽、成本失控等问题；如果只做缓存，又无法解决长任务和突发流量；如果只做队列，也会影响实时交互体验。

一套比较稳妥的方案应该是：

Nginx 限流
+ FastAPI 异步服务
+ Redis 缓存与用户限流
+ Celery 队列削峰
+ 模型网关统一调度
+ Docker 水平扩容
+ 压测验证
+ 监控告警

对于中小型 AI 产品，可以先用本文提供的 Docker Compose 方案快速搭建；当业务规模扩大后，再逐步升级到 Kubernetes、Prometheus、Grafana、APISIX、Kafka、Milvus、vLLM 等更完整的生产级架构。

真正稳定的 AI 高并发系统，关键不在于某一个框架，而在于：请求是否可控、任务是否可排队、模型是否可切换、成本是否可监控、故障是否可降级、服务是否可扩容。只要围绕这些原则设计，AI 工具就能够在真实业务高峰中保持稳定、可用和可持续增长。