一人公司的 AI 产品：用最小架构跑通“检索—生成—计费”闭环

# main.py
import os, time
from fastapi import FastAPI, HTTPException, Header
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"], api_key=os.environ.get("QDRANT_API_KEY"))
r = redis.from_url(os.environ["REDIS_URL"], decode_responses=True)

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

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

@app.post("/ask")
def ask(body: Ask, authorization: str = Header(...)):
    # 每用户每分钟 20 次限流
    key = f"rate:{body.user_id}:{int(time.time() // 60)}"
    if r.incr(key) > 20:
        raise HTTPException(429, "Rate limit exceeded")
    r.expire(key, 60)

    # 检索私有知识库
    vector = embed(body.question)
    hits = qdrant.search(
        collection_name="docs", query_vector=vector, limit=5
    )
    context = "\n".join(h.payload["text"] for h in hits)

    # 生成回答
    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.2,
        max_tokens=600,
    )
    answer = resp.choices[0].message.content
    usage = resp.usage.total_tokens

    # 记录用量，用于计费与成本分析
    r.hincrby(f"usage:{body.user_id}", "tokens", usage)
    r.hincrby(f"usage:{body.user_id}", "requests", 1)
    return {"answer": answer, "tokens": usage}

# billing.py
import os, redis, stripe
stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
r = redis.from_url(os.environ["REDIS_URL"], decode_responses=True)

def charge_overage(user_id: str, free_tokens: int = 100_000):
    used = int(r.hget(f"usage:{user_id}", "tokens") or 0)
    if used <= free_tokens:
        return
    overage = used - free_tokens
    amount = int(overage / 1000 * 0.2)  # 每千 token 约 0.2 美分
    stripe.PaymentIntent.create(
        amount=amount, currency="usd",
        customer=r.get(f"stripe_customer:{user_id}"),
        description=f"AI 超额用量 {overage} tokens"
    )

# docker-compose.yml
services:
  api:
    build: .
    ports: ["8000:8000"]
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - QDRANT_URL=http://qdrant:6333
      - REDIS_URL=redis://redis:6379
    depends_on: [qdrant, redis]
  qdrant:
    image: qdrant/qdrant:latest
    ports: ["6333:6333"]
  redis:
    image: redis:7-alpine