AI 落地实战：从 POC 到生产的最后一公里

用户 → API 网关(鉴权/限流) → 编排服务 → 缓存 → 向量检索 → 模型生成 → 输出校验 → 审计
                                      ↓
                                   人工兜底队列

import os, json, hashlib, time
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from openai import OpenAI
from qdrant_client import QdrantClient
import redis

app = FastAPI()
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
qdrant = QdrantClient(url=os.environ["QDRANT_URL"])
r = redis.from_url(os.environ["REDIS_URL"], decode_responses=True)

class Ask(BaseModel):
    question: str
    tenant_id: str

def embed(text: str) -> list[float]:
    return client.embeddings.create(
        model="text-embedding-3-small", input=text
    ).data[0].embedding

def retrieve(question: str, tenant_id: str, top_k: int = 5):
    hits = qdrant.search(
        collection_name=f"docs_{tenant_id}",
        query_vector=embed(question),
        limit=top_k,
        score_threshold=0.35,  # 低于阈值视为无相关知识
    )
    return [h.payload["text"] for h in hits]

@app.post("/ask")
def ask(body: Ask):
    # 1. 缓存：相同问题直接返回
    cache_key = f"qa:{body.tenant_id}:{hashlib.md5(body.question.encode()).hexdigest()}"
    if cached := r.get(cache_key):
        return {"answer": cached, "source": "cache"}

    # 2. 限流：每租户每分钟 30 次
    rate_key = f"rate:{body.tenant_id}:{int(time.time() // 60)}"
    if r.incr(rate_key) > 30:
        raise HTTPException(429, "请求过于频繁")
    r.expire(rate_key, 60)

    # 3. 检索：无相关知识直接拒答，避免幻觉
    contexts = retrieve(body.question, body.tenant_id)
    if not contexts:
        return {"answer": "抱歉，知识库中暂无相关内容，已转人工。",
                "source": "no_context", "need_human": True}

    # 4. 生成：强制基于上下文，低温度
    context = "\n---\n".join(contexts)
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content":
                "仅根据给定上下文回答，禁止编造。"
                "若上下文不足，回复'无法确定'。"},
            {"role": "user", "content": f"上下文：\n{context}\n\n问题：{body.question}"},
        ],
        temperature=0.1,
        max_tokens=500,
    )
    answer = resp.choices[0].message.content

    # 5. 输出校验：拒答词检测，触发人工
    if "无法确定" in answer or len(answer) < 5:
        return {"answer": answer, "source": "low_confidence", "need_human": True}

    # 6. 审计与缓存
    r.setex(cache_key, 600, answer)
    r.hincrby(f"usage:{body.tenant_id}", "tokens", resp.usage.total_tokens)
    r.hincrby(f"usage:{body.tenant_id}", "requests", 1)
    return {"answer": answer, "source": "llm"}

def evaluate(testset: list[dict]) -> dict:
    correct, refused, wrong = 0, 0, 0
    for item in testset:
        result = ask(Ask(question=item["q"], tenant_id="test"))
        ans = result["answer"]
        if result.get("need_human"):
            refused += 1
        elif item["keyword"] in ans:
            correct += 1
        else:
            wrong += 1
    total = len(testset)
    return {
        "准确率": round(correct / total, 3),
        "拒答率": round(refused / total, 3),
        "错误率": round(wrong / total, 3),
    }

def create_ticket(tenant_id: str, question: str, answer: str):
    ticket = {
        "tenant": tenant_id, "question": question,
        "ai_answer": answer, "status": "pending",
        "created_at": time.time(),
    }
    r.lpush(f"tickets:{tenant_id}", json.dumps(ticket, ensure_ascii=False))
    return {"ticket_created": True}