摘要：本文用 Python 搭建一个生产级 AI 文档分析平台，覆盖 LLM 调用的 6 个关键层——重试策略、流式输出、成本追踪、限流削峰、响应缓存、异步编排。每一步都附有可直接部署的代码，以及踩过的坑。读完你会有一个清晰的认知：调大模型 API 和调数据库一样，需要一层正经的中间件。

---

目录

• 一、一个 429 引发的血案
• 二、LLM 调用层：生产级六件套
  • 2.1 连接池 + Retry-After 退避
  • 2.2 流式与 SSE 心跳
  • 2.3 Token 成本追踪
  • 2.4 上下文窗口管理
• 三、可靠性：限流 + 缓存 + 降级
  • 3.1 令牌桶限流器
  • 3.2 LLM 响应缓存与成本收益分析
  • 3.3 模型降级策略
• 四、微服务架构落地
  • 4.1 项目结构
  • 4.2 API Gateway（FastAPI）
  • 4.3 异步任务（Celery）
• 五、启动与验证
• 六、生产环境 Checklist
• 七、总结

---

一、一个 429 引发的血案

某客服系统上线第一天，架构是这样的：

import requests
# 每次用户发消息 → 直接调 GPT-4
resp = requests.post(
    "https://api.openai.com/v1/chat/completions",
    headers={"Authorization": f"Bearer {api_key}"},
    json={"model": "gpt-4", "messages": [...]}
)

高峰期 QPS 冲到 200，GPT-4 限流 429 瞬间打爆。 没有重试 → 消息全丢；没有缓存 → 相同问题反复调 API；没有成本追踪 → 月底账单 8 万。

这不是模型的问题，是调用方式的问题。 一张表看清差距：

维度	Demo 写法	生产级要求
重试	无	429/5xx 指数退避 + Retry-After 头
超时	无	connect=5s / read=60s
流式	无	SSE 逐 token + 心跳防超时
Token	不关心	每次调用记录 prompt+completion 消耗
限流	无	令牌桶控 QPS，Semaphore 控并发
缓存	无	SHA256(model+messages) → Redis
降级	无	主模型超限 → 自动切备选
上下文	不处理	超出窗口自动截断 + 摘要压缩

我们要搭建的是 AI 文档分析平台：上传文档 → 自动摘要、分类、关键词提取。

                          ┌──────────┐
                          │ 用户/前端  │
                          └────┬─────┘
                        ┌──────▼──────┐
                        │ API Gateway │
                        │  (FastAPI)  │
                        └──┬──────┬───┘
                    ┌──────┘      └──────────┐
            ┌───────▼──────┐        ┌───────▼──────┐
            │  LLM Client  │        │ Celery Queue │
            │ (httpx+Redis)│        │ (文档异步处理) │
            └───────┬──────┘        └──────────────┘
                    │
            ┌───────▼──────┐
            │  Redis       │
            │ (限流+缓存)   │
            └──────────────┘

---

二、LLM 调用层：生产级六件套

2.1 连接池 + Retry-After 退避

# llm_service/client.py
import time
import httpx
from dataclasses import dataclass


@dataclass
class LLMConfig:
    api_key: str
    base_url: str = "https://api.deepseek.com"
    model: str = "deepseek-chat"
    max_retries: int = 3
    temperature: float = 0.3
    max_tokens: int = 2048
    # 分阶段超时：连接 5s，读 LLM 响应 60s
    connect_timeout: float = 5.0
    read_timeout: float = 60.0


class LLMClient:
    """生产级 LLM 客户端"""

    def __init__(self, config: LLMConfig):
        self.config = config
        # ★ 全局复用一个 Client，利用连接池
        self.client = httpx.Client(
            timeout=httpx.Timeout(
                connect=config.connect_timeout,
                read=config.read_timeout,
                write=10.0, pool=5.0
            ),
            limits=httpx.Limits(
                max_keepalive_connections=10,
                max_connections=20
            )
        )

    def chat(self, messages: list[dict]) -> dict:
        """同步调用，自动重试 + Retry-After 解析"""
        last_error = None

        for attempt in range(self.config.max_retries):
            try:
                resp = self.client.post(
                    f"{self.config.base_url}/v1/chat/completions",
                    headers={
                        "Authorization": f"Bearer {self.config.api_key}",
                        "Content-Type": "application/json"
                    },
                    json={
                        "model": self.config.model,
                        "messages": messages,
                        "temperature": self.config.temperature,
                        "max_tokens": self.config.max_tokens,
                    }
                )

                if resp.status_code == 200:
                    return resp.json()

                # ── 可重试的错误码 ──
                if resp.status_code in (429, 500, 502, 503):
                    wait = self._get_retry_delay(resp, attempt)
                    print(f"[LLM] {resp.status_code}, "
                          f"重试 {attempt+1}/{self.config.max_retries}, 等{wait:.1f}s")
                    time.sleep(wait)
                    continue

                # ── 不可重试：先读 body 再抛 ──
                body = resp.text[:500]
                raise RuntimeError(
                    f"LLM API 返回 {resp.status_code}: {body}")

            except httpx.TimeoutException:
                last_error = f"超时(attempt {attempt+1})"
                time.sleep(2 ** attempt)
            except httpx.ConnectError:
                last_error = "连接失败"
                time.sleep(2 ** attempt)

        raise RuntimeError(f"LLM 调用失败（重试{self.config.max_retries}次）: {last_error}")

    def _get_retry_delay(self, resp: httpx.Response, attempt: int) -> float:
        """解析 Retry-After 头，有就用；没有就指数退避"""
        retry_after = resp.headers.get("Retry-After", "")
        if retry_after.isdigit():
            return float(retry_after)
        # 指数退避：1s, 2s, 4s + 50% 随机抖动
        return (2 ** attempt) * (0.5 + 0.5 * (time.time() % 1))

    def chat_simple(self, user_message: str, system_prompt: str = "") -> str:
        """一句话调用，返回纯文本"""
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": user_message})
        return self.chat(messages)["choices"][0]["message"]["content"]

    def close(self):
        self.client.close()

三个关键改进（对比基础版）：

改进	旧版	新版
超时控制	单一 timeout=60	分阶段：connect=5s, read=60s
429 退避	固定 2^attempt	优先读 API 的 Retry-After 头
错误信息	raise_for_status() 不读 body	先读 resp.text 再抛，便于排查

2.2 流式与 SSE 心跳

用户等 30 秒看全量结果 vs 逐字显示——这是两种不同的用户体验。但 SSE 还有一个隐藏坑：长时间无 token 输出会导致浏览器/Nginx 超时断开。

# llm_service/streaming.py
import json
import asyncio
import httpx
from typing import AsyncGenerator


class StreamingLLMClient:
    """SSE 流式 + 心跳保活"""

    HEARTBEAT_INTERVAL = 15  # 每 15s 无数据发一次心跳

    def __init__(self, config: LLMConfig):
        self.config = config
        self.client = httpx.AsyncClient(
            timeout=httpx.Timeout(
                connect=config.connect_timeout,
                read=config.read_timeout
            ),
            limits=httpx.Limits(max_keepalive_connections=20)
        )

    async def chat_stream(self, messages: list[dict]) -> AsyncGenerator[str, None]:
        """逐 token yield，含心跳"""
        last_data_time = asyncio.get_event_loop().time()

        async with self.client.stream(
            "POST", f"{self.config.base_url}/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {self.config.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": self.config.model,
                "messages": messages,
                "temperature": self.config.temperature,
                "max_tokens": self.config.max_tokens,
                "stream": True,
            }
        ) as resp:
            resp.raise_for_status()
            async for line in resp.aiter_lines():
                if not line.startswith("data: "):
                    continue
                data_str = line[6:]
                if data_str == "[DONE]":
                    break
                try:
                    chunk = json.loads(data_str)
                    delta = chunk["choices"][0].get("delta", {})
                    content = delta.get("content", "")
                    if content:
                        last_data_time = asyncio.get_event_loop().time()
                        yield content
                except (json.JSONDecodeError, KeyError, IndexError):
                    continue

                # ★ 心跳检测：长时间无 token → 发 SSE 注释保持连接
                now = asyncio.get_event_loop().time()
                if now - last_data_time > self.HEARTBEAT_INTERVAL:
                    yield ": heartbeat\n\n"       # SSE 注释，浏览器忽略
                    last_data_time = now

    async def close(self):
        await self.client.aclose()

Gateway 端接入：

# 在 gateway/main.py 中（完整代码见 §4.2）
@app.post("/api/chat/stream")
async def chat_stream(prompt: str, system: str = ""):
    messages = [{"role": "user", "content": prompt}]
    if system:
        messages.insert(0, {"role": "system", "content": system})

    async def generate():
        try:
            async for token in stream_client.chat_stream(messages):
                yield f"data: {token}\n\n"
            yield "data: [DONE]\n\n"
        except Exception as e:
            yield f"data: [ERROR] {e}\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={"X-Accel-Buffering": "no"}  # 禁用 Nginx 缓冲
    )

关键：X-Accel-Buffering: no 让 Nginx 不缓冲 SSE 流，token 实时送达前端。配合 15s 心跳，浏览器不会误判超时。

2.3 Token 成本追踪

# llm_service/tracker.py


class TokenTracker:
    """Token 成本追踪 — 每次 LLM 调用都计入"""

    PRICING = {  # 元 / 百万 token（2025 年 7 月）
        "deepseek-chat":  {"prompt": 1.0, "completion": 2.0},
        "deepseek-reasoner": {"prompt": 4.0, "completion": 16.0},
        "gpt-4o":         {"prompt": 20.0, "completion": 80.0},
        "gpt-4o-mini":    {"prompt": 1.0, "completion": 4.0},
    }

    def __init__(self, model: str = "deepseek-chat"):
        self.model = model
        self.total_prompt = 0
        self.total_completion = 0
        self.call_count = 0
        self.cache_hits = 0

    def record(self, prompt_tokens: int, completion_tokens: int):
        """记录一次调用（直接用 API 返回的 usage，最准）"""
        self.total_prompt += prompt_tokens
        self.total_completion += completion_tokens
        self.call_count += 1

    def record_cache_hit(self):
        self.cache_hits += 1

    @property
    def total_cost(self) -> float:
        p = self.PRICING.get(self.model, {"prompt": 0, "completion": 0})
        return (self.total_prompt / 1_000_000 * p["prompt"]
                + self.total_completion / 1_000_000 * p["completion"])

    @property
    def cache_hit_rate(self) -> float:
        total = self.call_count + self.cache_hits
        return self.cache_hits / total if total > 0 else 0

    def report(self) -> dict:
        return {
            "model": self.model,
            "calls": self.call_count,
            "cache_hits": self.cache_hits,
            "hit_rate": f"{self.cache_hit_rate:.1%}",
            "total_tokens": self.total_prompt + self.total_completion,
            "total_cost_yuan": round(self.total_cost, 4)
        }

2.4 上下文窗口管理

LLM 有上下文长度上限。一篇 5000 字的文档直接扔进去可能超限。策略：按 token 数截断 + 重叠窗口：

# llm_service/context.py
import tiktoken

class ContextManager:
    """上下文窗口管理器 — 防 token 超限"""

    def __init__(self, max_context_tokens: int = 6000,
                 encoding: str = "cl100k_base"):
        self.max_tokens = max_context_tokens
        self.encoder = tiktoken.get_encoding(encoding)

    def count_tokens(self, text: str) -> int:
        return len(self.encoder.encode(text))

    def fit_context(self, text: str, reserved: int = 1500) -> str:
        """
        确保 prompt + 文档 ≤ max_context_tokens。
        reserved: 留给 prompt 模板和输出的 token 数
        """
        budget = self.max_tokens - reserved
        tokens = self.encoder.encode(text)

        if len(tokens) <= budget:
            return text

        # 截断：保留开头（摘要）+ 结尾（结论），各占一半
        head_size = budget // 2
        tail_size = budget - head_size

        head = self.encoder.decode(tokens[:head_size])
        tail = self.encoder.decode(tokens[-tail_size:])
        return f"{head}\n\n... [中间 {len(tokens)-budget} tokens 已省略] ...\n\n{tail}"


# 使用示例
ctx = ContextManager(max_context_tokens=6000)
long_doc = open("report.txt").read()

# 第一轮：完整摘要（只传开头）
summary = llm.chat_simple(
    f"总结以下文档（仅开头部分）：\n{ctx.fit_context(long_doc, reserved=2000)}",
    "你是专业文档分析助手")

# 第二、三轮... 如果文档超长，继续对后续部分提问

---

三、可靠性：限流 + 缓存 + 降级

3.1 令牌桶限流器

# llm_service/limiter.py
import asyncio
import time


class TokenBucket:
    """异步令牌桶 — 精确控 QPS"""

    def __init__(self, rate: int, capacity: int = None):
        self.rate = rate              # 每秒生成令牌数
        self.capacity = capacity or rate
        self.tokens = float(self.capacity)
        self.last_refill = time.monotonic()
        self._lock = asyncio.Lock()

    async def acquire(self):
        async with self._lock:
            now = time.monotonic()
            self.tokens = min(self.capacity,
                              self.tokens + (now - self.last_refill) * self.rate)
            self.last_refill = now

            if self.tokens >= 1:
                self.tokens -= 1
                return
            wait = (1 - self.tokens) / self.rate
            self.tokens = 0.0
        await asyncio.sleep(wait)


class ConcurrentLimiter:
    def __init__(self, max_concurrent: int = 5):
        self._sem = asyncio.Semaphore(max_concurrent)

    async def __aenter__(self):
        await self._sem.acquire()
        return self

    async def __aexit__(self, *args):
        self._sem.release()

3.2 LLM 响应缓存与成本收益分析

相同问题反复调 API = 烧钱。SHA256(model+messages) 做 key，Redis 缓存：

# llm_service/cache.py
import hashlib
import json
from redis.asyncio import Redis
from typing import Optional


class LLMCache:
    """LLM 响应缓存 — SHA256 精确匹配"""

    def __init__(self, redis: Redis, default_ttl: int = 3600):
        self.redis = redis
        self.default_ttl = default_ttl
        self.prefix = "llm:cache:"

    def _hash(self, model: str, messages: list[dict]) -> str:
        payload = json.dumps({"model": model, "messages": messages},
                             ensure_ascii=False, sort_keys=True)
        return hashlib.sha256(payload.encode()).hexdigest()[:16]

    async def get(self, model: str, messages: list[dict]) -> Optional[str]:
        key = self.prefix + self._hash(model, messages)
        val = await self.redis.get(key)
        return val.decode() if val else None

    async def set(self, model: str, messages: list[dict],
                  response: str, ttl: int = None):
        key = self.prefix + self._hash(model, messages)
        await self.redis.setex(key, ttl or self.default_ttl, response)

    @property
    async def entry_count(self) -> int:
        keys = await self.redis.keys(self.prefix + "*")
        return len(keys)

成本收益分析（真实数据）：

场景：AI 客服系统，日均 10,000 次调用
假设 35% 是重复问题（"退货流程"、"如何退款" 等）
每次调用均价 ¥0.005

无缓存：
  10,000 次 × ¥0.005 = ¥50/天 = ¥18,250/年

有缓存（35% 命中率）：
  6,500 次 × ¥0.005 = ¥32.5/天 = ¥11,863/年
  年节省：¥6,387（≈ 一台 iPad Pro）

缓存成本：Redis 内存 ≈ 50MB（5000 条 × 10KB）= ¥50/月
ROI：¥6,387 / ¥600 ≈ 10.6 倍

3.3 模型降级策略

主模型限流或故障时，自动切备选模型：

# llm_service/fallback.py
from dataclasses import dataclass


@dataclass
class FallbackConfig:
    """模型降级链：按优先级尝试"""
    models: list[str]  # ["gpt-4o", "gpt-4o-mini", "deepseek-chat"]


class FallbackLLMClient:
    """带降级的 LLM 客户端 — 主模型不可用时自动切换"""

    def __init__(self, base_config: LLMConfig, fallback: FallbackConfig):
        self.base_config = base_config
        self.fallback_config = fallback
        self.clients: dict[str, LLMClient] = {}

        # 预创建所有客户端实例
        for model in [base_config.model] + fallback.models:
            cfg = LLMConfig(
                api_key=base_config.api_key,
                base_url=base_config.base_url,
                model=model,
            )
            self.clients[model] = LLMClient(cfg)

    def chat(self, messages: list[dict]) -> dict:
        """按优先级尝试，哪个模型成功用哪个"""
        order = [self.base_config.model] + self.fallback_config.models

        for i, model in enumerate(order):
            try:
                return self.clients[model].chat(messages)
            except Exception as e:
                if i < len(order) - 1:
                    print(f"[降级] {model} 失败 → 尝试 {order[i+1]}: {e}")
                else:
                    raise  # 全部失败

实用建议：降级模型要选同协议（OpenAI 兼容）的。deepseek → qwen → moonshot 都可以用同一个 base_url 格式。

---

四、微服务架构落地

4.1 项目结构

ai-platform/
├── requirements.txt
├── llm_service/
│   ├── __init__.py
│   ├── client.py         # LLMClient（§2.1）
│   ├── streaming.py      # StreamingLLMClient（§2.2）
│   ├── tracker.py        # TokenTracker（§2.3）
│   ├── context.py        # ContextManager（§2.4）
│   ├── limiter.py        # TokenBucket + ConcurrentLimiter（§3.1）
│   ├── cache.py          # LLMCache（§3.2）
│   └── fallback.py       # FallbackLLMClient（§3.3）
├── gateway/
│   └── main.py           # FastAPI 统一入口
└── tasks/
    ├── celery_app.py      # Celery 配置
    └── doc_tasks.py       # 文档异步分析

requirements.txt：

fastapi==0.115.0
uvicorn[standard]==0.30.0
httpx==0.27.0
redis==5.0.0
celery[redis]==5.4.0
tiktoken==0.7.0
python-multipart==0.0.9

4.2 API Gateway（FastAPI）

# gateway/main.py —— 统一入口
import os, time
import redis.asyncio as aioredis
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware

from llm_service.client import LLMClient, LLMConfig
from llm_service.streaming import StreamingLLMClient
from llm_service.cache import LLMCache
from llm_service.limiter import TokenBucket, ConcurrentLimiter
from llm_service.tracker import TokenTracker
from llm_service.context import ContextManager

app = FastAPI(title="AI Platform", version="2.0.0")
app.add_middleware(CORSMiddleware, allow_origins=["*"],
                   allow_methods=["*"], allow_headers=["*"])

config = LLMConfig(
    api_key=os.getenv("LLM_API_KEY", "sk-your-key"),
    base_url=os.getenv("LLM_BASE_URL", "https://api.deepseek.com"),
    model=os.getenv("LLM_MODEL", "deepseek-chat"),
)

llm = LLMClient(config)
stream_llm = StreamingLLMClient(config)
ctx_mgr = ContextManager()

redis = aioredis.from_url(
    os.getenv("REDIS_URL", "redis://localhost:6379"), decode_responses=True)
cache = LLMCache(redis, default_ttl=1800)

bucket = TokenBucket(rate=20)
concurrent = ConcurrentLimiter(8)
tracker = TokenTracker(config.model)

# ========== 中间件 ==========

@app.middleware("http")
async def log(request: Request, call_next):
    t0 = time.time()
    resp = await call_next(request)
    print(f"[Gateway] {request.method} {request.url.path} "
          f"→ {resp.status_code} ({time.time()-t0:.3f}s)")
    return resp

# ========== 路由 ==========

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

@app.post("/api/chat/sync")
async def chat_sync(prompt: str, system: str = ""):
    """同步 LLM 调用"""
    await bucket.acquire()
    async with concurrent:
        try:
            # 上下文窗口保护
            safe_prompt = ctx_mgr.fit_context(prompt)
            result = llm.chat_simple(safe_prompt, system)
            return {"success": True, "content": result}
        except Exception as e:
            raise HTTPException(502, detail=str(e))

@app.post("/api/chat/stream")
async def chat_stream(prompt: str, system: str = ""):
    """SSE 流式"""
    msgs = []
    if system:
        msgs.append({"role": "system", "content": system})
    msgs.append({"role": "user", "content": ctx_mgr.fit_context(prompt)})

    async def gen():
        try:
            async for token in stream_llm.chat_stream(msgs):
                yield f"data: {token}\n\n"
            yield "data: [DONE]\n\n"
        except Exception as e:
            yield f"data: [ERROR] {e}\n\n"

    return StreamingResponse(gen(), media_type="text/event-stream",
                             headers={"X-Accel-Buffering": "no"})

@app.post("/api/chat/cached")
async def chat_cached(prompt: str, system: str = ""):
    """带缓存调用"""
    safe_prompt = ctx_mgr.fit_context(prompt)
    msgs = []
    if system:
        msgs.append({"role": "system", "content": system})
    msgs.append({"role": "user", "content": safe_prompt})

    cached = await cache.get(config.model, msgs)
    if cached:
        tracker.record_cache_hit()
        return {"success": True, "content": cached, "cached": True}

    await bucket.acquire()
    async with concurrent:
        try:
            resp = llm.chat(msgs)
            content = resp["choices"][0]["message"]["content"]
            usage = resp.get("usage", {})
            if usage:
                tracker.record(usage["prompt_tokens"], usage["completion_tokens"])
            await cache.set(config.model, msgs, content)
            return {"success": True, "content": content, "cached": False}
        except Exception as e:
            raise HTTPException(502, detail=str(e))

@app.get("/api/stats")
async def stats():
    return {
        "tracker": tracker.report(),
        "cache_entries": await cache.entry_count
    }

注意：llm / stream_llm 全局复用，利用 httpx 连接池。不要每次请求 new LLMClient()。

4.3 异步任务（Celery）

# tasks/celery_app.py
from celery import Celery

app = Celery("ai_tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1")
app.conf.update(task_serializer="json", accept_content=["json"],
                result_expires=3600, task_track_started=True,
                worker_max_tasks_per_child=100)

# tasks/doc_tasks.py
import os
from .celery_app import app
from llm_service.client import LLMClient, LLMConfig
from llm_service.context import ContextManager

llm = LLMClient(LLMConfig(
    api_key=os.getenv("LLM_API_KEY", "sk-your-key"),
    base_url=os.getenv("LLM_BASE_URL", "https://api.deepseek.com"),
))
ctx = ContextManager()


@app.task(bind=True, max_retries=3, default_retry_delay=10)
def analyze_document(self, text: str) -> dict:
    """异步文档分析：摘要 + 分类 + 关键词"""
    result = {"task_id": self.request.id, "status": "processing"}
    doc = ctx.fit_context(text, reserved=2500)

    # 摘要
    try:
        result["summary"] = llm.chat_simple(
            f"用 100 字以内总结：\n{doc}",
            "你是专业文档分析助手")
    except Exception as e:
        result["summary"] = f"摘要失败: {e}"

    # 分类
    try:
        result["category"] = llm.chat_simple(
            f"归类（技术/商业/法律/教育/其他）：\n{doc[:1000]}",
            "只回复分类名称").strip()
    except Exception as e:
        result["category"] = f"分类失败: {e}"

    # 关键词
    try:
        kw = llm.chat_simple(
            f"提取 5 个关键词，逗号分隔：\n{doc[:1500]}",
            "只回复关键词")
        result["keywords"] = [k.strip() for k in kw.split(",") if k.strip()]
    except Exception as e:
        result["keywords"] = []

    result["status"] = "completed"
    return result

Gateway 追加：

# gateway/main.py 追加
from tasks.doc_tasks import analyze_document
from celery.result import AsyncResult

@app.post("/api/doc/analyze")
async def submit(text: str):
    task = analyze_document.delay(text)
    return {"task_id": task.id, "status": "submitted"}

@app.get("/api/doc/result/{task_id}")
async def result(task_id: str):
    t = AsyncResult(task_id, app=analyze_document.app)
    return ({"status": "completed", "result": t.result}
            if t.ready() else {"status": t.status})

---

五、启动与验证

pip install -r requirements.txt

# 终端1
redis-server

# 终端2
cd ai-platform
uvicorn gateway.main:app --host 0.0.0.0 --port 8000 --reload

# 终端3
cd ai-platform
celery -A tasks.celery_app worker --loglevel=info --concurrency=4

# 健康检查
curl http://localhost:8000/health

# 同步（自动上下文截断保护）
curl -X POST http://localhost:8000/api/chat/sync \
  -H "Content-Type: application/json" \
  -d '{"prompt":"一句话解释微服务"}'

# 流式（逐字输出）
curl -N -X POST http://localhost:8000/api/chat/stream \
  -H "Content-Type: application/json" \
  -d '{"prompt":"写一首五言诗"}'

# 缓存（首次 cached=false，再次 cached=true）
curl -X POST http://localhost:8000/api/chat/cached \
  -H "Content-Type: application/json" \
  -d '{"prompt":"什么是 Redis"}'

# 异步文档分析
curl -X POST http://localhost:8000/api/doc/analyze \
  -H "Content-Type: application/json" \
  -d '{"text":"Kubernetes is an open-source container orchestrator..."}'
# → {"task_id":"xxx","status":"submitted"}
curl http://localhost:8000/api/doc/result/xxx

# 统计面板（调用数/缓存命中率/花费）
curl http://localhost:8000/api/stats

---

六、生产环境 Checklist

#	检查项	要点
1	API Key 走环境变量	os.getenv()，严禁硬编码
2	分阶段超时	connect=5s / read=60s，不是单一 timeout
3	429 读 Retry-After	优先用 API 给的退避时间
4	并发 + QPS 限制	TokenBucket + Semaphore
5	SSE 有 X-Accel-Buffering	Nginx 不缓冲流式数据
6	SSE 有心跳	15s 无 token → 发 : heartbeat
7	Token 消耗 + 缓存命中可追踪	/api/stats 实时面板
8	长文档有窗口管理	ContextManager.fit_context()
9	长任务异步化	Celery 处理 30s+ 的分析任务
10	Worker 内存保护	worker_max_tasks_per_child=100
11	模型可降级	FallbackLLMClient 备选模型链
12	日志结构化	每次 LLM 调用：耗时 / Token / status / 缓存命中

---

七、总结

┌──────────────────────────────────────────────────────────────────┐
│                  Python AI 应用全链路速查                         │
├──────────────┬────────────────────┬──────────────────────────────┤
│  API Gateway │    LLM Service     │       Task Queue             │
│  (FastAPI)   │   (httpx+Redis)    │      (Celery)                │
├──────────────┼────────────────────┼──────────────────────────────┤
│ SSE + 心跳    │ 连接池 + Retry-After│ 异步文档分析                  │
│ 令牌桶限流     │ 上下文窗口管理       │ 摘要+分类+关键词               │
│ 缓存拦截       │ Token 实时追踪      │ 三任务独立失败隔离             │
│ 模型降级       │ 成本收益分析         │ 结果轮询                     │
│ Nginx 反压    │ SHA256 精确匹配     │ 内存保护                     │
├──────────────┴────────────────────┴──────────────────────────────┤
│ 核心原则：                                                       │
│ 1. 每次 LLM 调用都要可回答：花了多少 Token？缓存命中了没？          │
│ 2. httpx 连接池 > requests | 流式 + 心跳 > 同步 | 缓存 > 重调     │
│ 3. 4321 法则：4 层防护（限流+重试+缓存+降级），3 秒内返回，        │
│    2 套模型（主+备），1 个统计面板全掌握                           │
└──────────────────────────────────────────────────────────────────┘

---

运行环境：Python 3.11+ · Redis 7 · FastAPI 0.115+

参考资源：OpenAI API 文档 · FastAPI StreamingResponse · Celery 最佳实践 · httpx 连接池 · tiktoken