生产化实战：从原型到可靠产品

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

                    ┌──────────────┐
   用户  ──────────▶│  API 网关层   │  Nginx / 负载均衡（店门迎客）
                    └──────┬───────┘
                           │
                    ┌──────▼───────┐
                    │ FastAPI 服务 │  同步请求 / SSE 流式（前台点单）
                    └──────┬───────┘
                           │
              ┌────────────┴────────────┐
              │                         │
       ┌──────▼──────┐           ┌──────▼──────┐
       │  Agent 内核  │           │  任务队列    │  Celery + Redis
       │ (LLM+工具)  │           │ (异步长任务) │ （后厨慢菜走外卖）
       └──────┬──────┘           └─────────────┘
              │
   ┌──────────┼──────────┐
   │          │          │
┌──▼──┐   ┌──▼──┐   ┌──▼──┐
│ LLM │   │工具池│   │向量库│  （厨师、刀具、菜谱库）
└─────┘   └─────┘   └─────┘

1

pip install fastapi "uvicorn[standard]" celery redis pydantic

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

# app/main.py
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from app.agent import Agent
from app.guardrails import GuardrailMiddleware

app = FastAPI(title="Agent 生产服务")

# 初始化 Agent 和护栏中间件（启动时加载一次，避免每次请求重复加载）
agent = Agent.from_config("config/agent.yaml")
guard = GuardrailMiddleware()

class ChatRequest(BaseModel):
    session_id: str          # 会话 ID，用于隔离不同用户的记忆
    message: str             # 用户输入
    stream: bool = False     # 是否流式返回

@app.post("/chat")
async def chat(req: ChatRequest):
    # 1. 输入护栏：拦截提示注入、违规内容
    blocked, reason = guard.check_input(req.message)
    if blocked:
        return {"code": 403, "msg": f"输入被拦截: {reason}"}

    if req.stream:
        # 2. 流式返回：逐 token 输出，首字延迟低
        async def gen():
            async for chunk in agent.astream(req.session_id, req.message):
                # 3. 输出护栏：逐片段脱敏
                yield guard.sanitize_output(chunk)
        # media_type 设为 text/event-stream，前端用 EventSource 接收
        return StreamingResponse(gen(), media_type="text/event-stream")

    # 4. 同步返回：等待完整结果
    result = await agent.arun(req.session_id, req.message)
    return {"code": 0, "data": guard.sanitize_output(result)}

1

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# app/worker.py
from celery import Celery
from app.agent import Agent

# broker：任务排队的地方；backend：结果存取的地方，都用 Redis
celery = Celery(
    "agent_worker",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

# 复用同一套 Agent 内核
agent = Agent.from_config("config/agent.yaml")

@celery.task(bind=True, max_retries=3, default_retry_delay=10)
def run_deep_task(self, session_id: str, task: str):
    """异步执行长任务，结果写回结果后端"""
    try:
        result = agent.run(session_id, task)
        return {"status": "done", "data": result}
    except Exception as exc:
        # 失败自动重试，最多 3 次，每次间隔 10 秒
        raise self.retry(exc=exc)

1

celery -A app.worker worker -l info --concurrency=8

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

# app/guardrails.py
import re
from typing import Tuple

# 提示注入的常见话术，用正则匹配
INJECTION_PATTERNS = [
    r"忽略.{0,4}(之前|以上|所有).{0,4}(指令|提示)",
    r"ignore (previous|above|all) (instructions?|prompts?)",
    r"(reveal|show|print).{0,10}(system prompt|系统提示)",
]

# 个人敏感信息（PII）模式：手机号、身份证、邮箱
PII_PATTERN = re.compile(
    r"(1[3-9]\d{9})"           # 手机号
    r"|(\d{15}|\d{18})"        # 身份证
    r"|([A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,4})",  # 邮箱
    re.IGNORECASE
)

class GuardrailMiddleware:
    def check_input(self, text: str) -> Tuple[bool, str]:
        """返回 (是否拦截, 原因)"""
        for pat in INJECTION_PATTERNS:
            if re.search(pat, text, re.IGNORECASE):
                return True, f"疑似提示注入: 命中 {pat}"
        return False, ""

    def sanitize_output(self, text: str) -> str:
        """输出脱敏：手机号/身份证/邮箱打码"""
        def mask(m):
            s = m.group(0)
            # 保留前 3 后 3，中间用 * 填充，比如 138****5678
            return s[:3] + "*" * (len(s) - 6) + s[-3:]
        return PII_PATTERN.sub(mask, text)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# 工具权限映射表：每个工具允许哪些角色调用
TOOL_PERMISSIONS = {
    "search": ["guest", "user", "admin"],       # 搜索：所有人可用
    "place_order": ["user", "admin"],           # 下单：需登录
    "refund": ["admin"],                        # 退款：仅管理员
}

def check_tool_permission(tool_name: str, role: str) -> bool:
    """检查某角色是否有权调用某工具"""
    return role in TOOL_PERMISSIONS.get(tool_name, [])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# 基于 Redis 的滑动窗口限流
import time

def rate_limit(redis_client, key: str, limit: int, window: int = 60) -> bool:
    """每 window 秒最多 limit 次请求，返回是否放行"""
    now = time.time()
    pipe = redis_client.pipeline()              # 用管道一次发多条命令，减少网络往返
    pipe.zremrangebyscore(key, 0, now - window) # 清除窗口外的过期记录
    pipe.zadd(key, {str(now): now})             # 加入当前请求时间戳
    pipe.zcard(key)                             # 统计窗口内请求数
    pipe.expire(key, window)                    # 设置 key 过期，避免内存泄漏
    _, _, count, _ = pipe.execute()
    return count <= limit

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# app/router.py
from app.agent import Agent

class ModelRouter:
    """根据任务复杂度路由到不同模型"""
    SIMPLE_KEYWORDS = {"查询", "天气", "翻译", "汇率", "几点"}

    def select_model(self, query: str, history_len: int) -> str:
        # 短查询 + 无历史 → 小模型（便宜 10 倍）
        if len(query) < 20 and history_len == 0:
            return "gpt-5.4-mini"
        # 命中简单关键词 → 小模型
        if any(k in query for k in self.SIMPLE_KEYWORDS):
            return "gpt-5.4-mini"
        # 复杂推理 → 大模型
        return "gpt-5.4"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# 语义缓存示例
class SemanticCache:
    def __init__(self, vector_store, threshold=0.95):
        self.store = vector_store
        self.threshold = threshold

    def get(self, query_emb):
        # 找相似度高于阈值的历史结果
        hits = self.store.search(query_emb, top_k=1)
        if hits and hits[0].score >= self.threshold:
            return hits[0].answer   # 命中缓存，直接复用
        return None

    def set(self, query_emb, answer):
        self.store.add(query_emb, answer)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# app/cost.py
from dataclasses import dataclass

# 模型单价（美元 / 1K token），按实际文档更新
PRICING = {
    "gpt-5.4":      {"input": 0.0025,  "output": 0.01},
    "gpt-5.4-mini": {"input": 0.00015, "output": 0.0006},
}

@dataclass
class CostRecord:
    model: str
    input_tokens: int
    output_tokens: int

    def cost(self) -> float:
        p = PRICING[self.model]
        # 输入输出分别计价，再除以 1000（单价是每 1K token）
        return (self.input_tokens * p["input"]
                + self.output_tokens * p["output"]) / 1000

# 使用示例
rec = CostRecord("gpt-5.4", input_tokens=1200, output_tokens=300)
print(f"本次调用成本: ${rec.cost():.4f}")
# 输出: 本次调用成本: $0.0060

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# app/resilience.py
import asyncio
import random
from typing import Callable

async def call_with_retry(
    func: Callable,
    *args,
    max_retries: int = 3,
    base_delay: float = 1.0,
    timeout: float = 30.0,
    **kwargs
):
    """带指数退避重试的 LLM 调用"""
    for attempt in range(max_retries + 1):
        try:
            # asyncio.wait_for 给单次调用加超时
            return await asyncio.wait_for(func(*args, **kwargs), timeout=timeout)
        except Exception as e:
            if attempt == max_retries:
                raise  # 最后一次失败，抛出异常
            # 退避时间 = 基础延迟 × 2^重试次数 + 随机抖动
            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
            print(f"第 {attempt+1} 次失败: {e}, {delay:.1f}s 后重试")
            await asyncio.sleep(delay)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

FALLBACK_CHAIN = ["gpt-5.4", "claude-3-5-sonnet", "qwen2.5-72b"]

async def robust_llm_call(prompt: str):
    """依次尝试降级链中的模型，全失败则走规则兜底"""
    for model in FALLBACK_CHAIN:
        try:
            return await call_with_retry(call_model, model, prompt)
        except Exception:
            print(f"{model} 不可用，尝试下一个")
    # 全部失败 → 规则降级（关键词匹配 + 模板回复）
    return rule_based_fallback(prompt)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# 在 app/main.py 中添加
@app.get("/health")
async def health():
    checks = {
        "llm": await ping_llm(),        # 探测 LLM 服务连通性
        "redis": await ping_redis(),    # 探测 Redis
        "vector_db": await ping_vector_db(),
    }
    status = "ok" if all(checks.values()) else "degraded"
    return {"status": status, "checks": checks}

1
2
3
4
5
6

用户 → Nginx → FastAPI(SSE) → Agent 内核
                              ├─ 订单工具（查/退，需登录）
                              ├─ 物流工具（只读）
                              ├─ FAQ 检索（向量库）
                              └─ 模型路由（简单→mini，复杂→5.4）
                                  └─ 降级链

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

# app/agent.py
from typing import AsyncIterator

class Agent:
    def __init__(self, llm, tools, memory, router, guard):
        self.llm = llm
        self.tools = tools
        self.memory = memory
        self.router = router
        self.guard = guard

    @classmethod
    def from_config(cls, path: str) -> "Agent":
        """从 YAML 配置文件加载 Agent，省略实现细节"""
        ...

    async def arun(self, session_id: str, query: str) -> str:
        """同步调用入口，内部调用 astream 拼接完整结果"""
        result = ""
        async for chunk in self.astream(session_id, query):
            result += chunk
        return result

    async def astream(self, session_id: str, query: str) -> AsyncIterator[str]:
        # 1. 取历史 + 路由模型
        history = self.memory.get(session_id)
        model = self.router.select_model(query, len(history))
        # 2. 构造提示词（含工具说明）
        messages = self._build_messages(history, query)
        # 3. 流式调用，遇到工具调用则暂停执行后继续
        async for chunk in self.llm.astream(model, messages, tools=self.tools):
            yield chunk
        # 4. 持久化本轮对话
        self.memory.add(session_id, query, role="user")

    def run(self, session_id: str, task: str) -> str:
        """同步版本，供 Celery worker 调用"""
        ...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# docker-compose.yml
version: "3.9"
services:
  api:
    build: .
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4
    environment:
      - REDIS_URL=redis://redis:6379/0
      - OPENAI_API_KEY=${OPENAI_API_KEY}   # 从环境变量读取，避免泄露
    depends_on: [redis, qdrant]
    deploy:
      replicas: 3                          # 开 3 个实例扛流量
    ports: ["8000:8000"]

  worker:
    build: .
    command: celery -A app.worker worker -l info --concurrency=8
    environment:
      - REDIS_URL=redis://redis:6379/0
    depends_on: [redis]

  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]

  qdrant:
    image: qdrant/qdrant:latest
    ports: ["6333:6333"]

1
2
3

from prometheus_fastapi_instrumentator import Instrumentator

Instrumentator().instrument(app).expose(app, endpoint="/metrics")