AI 变现实战：从成本模型到计费系统，构建可盈利 AI 产品的完整代码指南

调用 GPT API -> 包装一个界面 -> 收月费

用户痛点 -> 可衡量结果 -> 愿意付费 -> 单位经济为正 -> 可规模化

毛利 = 收入 - 变动成本
毛利率 = 毛利 / 收入
LTV = ARPU × 毛利率 × 生命周期
CAC = 获客成本
LTV / CAC > 3 才算健康

每个用户每月审查 50 份合同
每份合同平均 8000 tokens 输入 + 2000 tokens 输出
使用 GPT-4o 级别模型：
  输入 $2.5 / 1M tokens
  输出 $10 / 1M tokens

单份合同成本：
  输入：8000 / 1_000_000 × 2.5 = $0.02
  输出：2000 / 1_000_000 × 10 = $0.02
  单份合计：$0.04

每月每用户 Token 成本：50 × $0.04 = $2.00
加上嵌入、存储、带宽：约 $0.50
变动成本合计：约 $2.50

如果定价 $29/月：
  毛利 = $29 - $2.50 = $26.50
  毛利率 = 91.4%

┌──────────────────────────────────────────────────┐
│                    用户端                          │
│   Web / App / API / 插件 / 企业集成                │
├──────────────────────────────────────────────────┤
│                  接入与认证层                       │
│   登录 / API Key / OAuth / SSO / 租户识别          │
├──────────────────────────────────────────────────┤
│                  配额与限流层                       │
│   套餐检查 / 用量配额 / 速率限制 / 并发控制          │
├──────────────────────────────────────────────────┤
│                  AI 业务层                         │
│   Prompt / RAG / Agent / 工作流 / 缓存             │
├──────────────────────────────────────────────────┤
│                  模型路由层                         │
│   模型选择 / 降级 / 重试 / 成本优化                 │
├──────────────────────────────────────────────────┤
│                  计量与计费层                       │
│   Token 计量 / 成本计算 / 用量记录 / 账单           │
├──────────────────────────────────────────────────┤
│                  支付层                            │
│   Stripe / 支付宝 / 微信支付 / 发票                 │
├──────────────────────────────────────────────────┤
│                  数据层                            │
│   PostgreSQL / Redis / 向量库 / 数仓                │
├──────────────────────────────────────────────────┤
│                  运营层                            │
│   利润看板 / 告警 / 用户分析 / 流失预警             │
└──────────────────────────────────────────────────┘

ai-monetization/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── db.py
│   ├── models.py
│   ├── auth.py
│   ├── metering.py
│   ├── pricing.py
│   ├── quota.py
│   ├── llm.py
│   └── billing.py
├── tests/
│   └── test_billing.py
├── requirements.txt
└── Dockerfile

pip install fastapi uvicorn sqlmodel tiktoken stripe pytest httpx

fastapi
uvicorn[standard]
sqlmodel
tiktoken
stripe
pytest
httpx

from datetime import datetime
from typing import Optional
from sqlmodel import SQLModel, Field


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str = Field(index=True, unique=True)
    api_key: str = Field(index=True, unique=True)
    plan: str = "free"                  # free / pro / business
    stripe_customer_id: Optional[str] = None
    created_at: datetime = Field(default_factory=datetime.utcnow)


class UsageRecord(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int = Field(index=True)
    model: str
    prompt_tokens: int
    completion_tokens: int
    cost_usd: float
    revenue_usd: float
    request_id: str = Field(index=True)
    created_at: datetime = Field(default_factory=datetime.utcnow)


class Quota(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int = Field(index=True)
    period: str                         # 例如 2026-09
    used_tokens: int = 0
    used_requests: int = 0
    updated_at: datetime = Field(default_factory=datetime.utcnow)

from sqlmodel import SQLModel, create_engine, Session

DATABASE_URL = "sqlite:///./ai_monetization.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},
)


def init_db():
    SQLModel.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session

import secrets
from fastapi import Header, HTTPException, Depends
from sqlmodel import Session, select

from app.db import get_session
from app.models import User


def generate_api_key() -> str:
    return "sk_" + secrets.token_urlsafe(32)


def get_current_user(
    x_api_key: str = Header(..., alias="X-API-Key"),
    session: Session = Depends(get_session),
) -> User:
    user = session.exec(
        select(User).where(User.api_key == x_api_key)
    ).first()
    if not user:
        raise HTTPException(status_code=401, detail="Invalid API Key")
    return user

from dataclasses import dataclass


@dataclass
class ModelPrice:
    input_per_1m: float
    output_per_1m: float


# 示例价格，请以官方最新价格为准
MODEL_PRICES = {
    "gpt-4o": ModelPrice(input_per_1m=2.5, output_per_1m=10.0),
    "gpt-4o-mini": ModelPrice(input_per_1m=0.15, output_per_1m=0.6),
    "gpt-5": ModelPrice(input_per_1m=5.0, output_per_1m=20.0),
}


@dataclass
class Plan:
    name: str
    monthly_price_usd: float
    included_tokens: int
    overage_per_1k_tokens: float
    max_requests_per_min: int


PLANS = {
    "free": Plan(
        name="free",
        monthly_price_usd=0.0,
        included_tokens=10_000,
        overage_per_1k_tokens=0.0,
        max_requests_per_min=5,
    ),
    "pro": Plan(
        name="pro",
        monthly_price_usd=29.0,
        included_tokens=1_000_000,
        overage_per_1k_tokens=0.05,
        max_requests_per_min=60,
    ),
    "business": Plan(
        name="business",
        monthly_price_usd=199.0,
        included_tokens=10_000_000,
        overage_per_1k_tokens=0.03,
        max_requests_per_min=300,
    ),
}


def calc_llm_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:
    price = MODEL_PRICES.get(model)
    if not price:
        raise ValueError(f"Unknown model: {model}")

    cost = (
        prompt_tokens / 1_000_000 * price.input_per_1m
        + completion_tokens / 1_000_000 * price.output_per_1m
    )
    return round(cost, 6)

import tiktoken
from datetime import datetime
from sqlmodel import Session, select

from app.models import UsageRecord, Quota
from app.pricing import calc_llm_cost


def count_tokens(text: str, model: str = "gpt-4o") -> int:
    try:
        enc = tiktoken.encoding_for_model(model)
    except KeyError:
        enc = tiktoken.get_encoding("cl100k_base")
    return len(enc.encode(text))


def current_period() -> str:
    return datetime.utcnow().strftime("%Y-%m")


def get_or_create_quota(session: Session, user_id: int) -> Quota:
    period = current_period()
    quota = session.exec(
        select(Quota).where(
            Quota.user_id == user_id,
            Quota.period == period,
        )
    ).first()

    if not quota:
        quota = Quota(user_id=user_id, period=period)
        session.add(quota)
        session.commit()
        session.refresh(quota)

    return quota


def record_usage(
    session: Session,
    user_id: int,
    model: str,
    prompt_tokens: int,
    completion_tokens: int,
    request_id: str,
    revenue_usd: float = 0.0,
) -> UsageRecord:
    cost = calc_llm_cost(model, prompt_tokens, completion_tokens)

    record = UsageRecord(
        user_id=user_id,
        model=model,
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
        cost_usd=cost,
        revenue_usd=revenue_usd,
        request_id=request_id,
    )
    session.add(record)

    quota = get_or_create_quota(session, user_id)
    quota.used_tokens += prompt_tokens + completion_tokens
    quota.used_requests += 1
    quota.updated_at = datetime.utcnow()

    session.add(quota)
    session.commit()
    session.refresh(record)
    return record

from datetime import datetime, timedelta
from collections import defaultdict
from fastapi import HTTPException

from app.pricing import PLANS

# 示例：内存限流。生产环境请使用 Redis。
_rate_bucket = defaultdict(list)


def check_rate_limit(user_id: int, plan_name: str):
    plan = PLANS.get(plan_name, PLANS["free"])
    now = datetime.utcnow()
    window_start = now - timedelta(minutes=1)

    bucket = _rate_bucket[user_id]
    bucket[:] = [t for t in bucket if t > window_start]

    if len(bucket) >= plan.max_requests_per_min:
        raise HTTPException(
            status_code=429,
            detail=f"Rate limit exceeded: {plan.max_requests_per_min}/min",
        )

    bucket.append(now)


def check_token_quota(used_tokens: int, plan_name: str):
    plan = PLANS.get(plan_name, PLANS["free"])
    if used_tokens >= plan.included_tokens:
        # free 套餐直接拒绝，付费套餐允许超额
        if plan.overage_per_1k_tokens <= 0:
            raise HTTPException(
                status_code=402,
                detail="Token quota exceeded. Please upgrade your plan.",
            )

from dataclasses import dataclass


@dataclass
class LLMResponse:
    text: str
    model: str
    prompt_tokens: int
    completion_tokens: int


def choose_model(task_type: str, plan: str) -> str:
    """
    模型路由：根据任务复杂度和用户套餐选择模型。
    这是控制成本的关键。
    """
    if plan == "free":
        return "gpt-4o-mini"

    if task_type == "simple":
        return "gpt-4o-mini"
    if task_type == "standard":
        return "gpt-4o"
    return "gpt-5"


def call_llm(prompt: str, model: str) -> LLMResponse:
    """
    示例实现。生产环境替换为真实 SDK 调用。
    """
    # 这里用简单估算模拟
    prompt_tokens = max(1, len(prompt) // 4)
    text = f"[mock:{model}] 针对以下内容的回答：{prompt[:50]}"
    completion_tokens = max(1, len(text) // 4)

    return LLMResponse(
        text=text,
        model=model,
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
    )

import hashlib
from typing import Optional

_cache: dict[str, LLMResponse] = {}


def cache_key(prompt: str, model: str) -> str:
    raw = f"{model}::{prompt}".encode("utf-8")
    return hashlib.sha256(raw).hexdigest()


def get_cached(prompt: str, model: str) -> Optional[LLMResponse]:
    return _cache.get(cache_key(prompt, model))


def set_cached(prompt: str, model: str, resp: LLMResponse):
    _cache[cache_key(prompt, model)] = resp

from sqlmodel import Session, select
from sqlmodel import func

from app.models import UsageRecord
from app.pricing import PLANS


def calc_revenue_for_request(plan_name: str) -> float:
    """
    按请求分摊订阅收入（简化版）。
    生产环境应按月汇总，按用量分摊。
    """
    plan = PLANS.get(plan_name, PLANS["free"])
    if plan.monthly_price_usd == 0:
        return 0.0
    # 简化：假设每月 1000 次请求
    return round(plan.monthly_price_usd / 1000, 6)


def calc_overage_revenue(
    session: Session,
    user_id: int,
    plan_name: str,
) -> float:
    plan = PLANS.get(plan_name, PLANS["free"])
    if plan.overage_per_1k_tokens <= 0:
        return 0.0

    total_tokens = session.exec(
        select(func.sum(UsageRecord.prompt_tokens + UsageRecord.completion_tokens))
        .where(UsageRecord.user_id == user_id)
    ).one() or 0

    overage_tokens = max(0, total_tokens - plan.included_tokens)
    return round(overage_tokens / 1000 * plan.overage_per_1k_tokens, 4)


def user_profit_report(session: Session, user_id: int) -> dict:
    records = session.exec(
        select(UsageRecord).where(UsageRecord.user_id == user_id)
    ).all()

    total_cost = sum(r.cost_usd for r in records)
    total_revenue = sum(r.revenue_usd for r in records)
    total_tokens = sum(r.prompt_tokens + r.completion_tokens for r in records)

    gross_profit = total_revenue - total_cost
    margin = gross_profit / total_revenue if total_revenue > 0 else 0.0

    return {
        "user_id": user_id,
        "requests": len(records),
        "total_tokens": total_tokens,
        "total_cost_usd": round(total_cost, 4),
        "total_revenue_usd": round(total_revenue, 4),
        "gross_profit_usd": round(gross_profit, 4),
        "gross_margin": round(margin, 4),
    }

import uuid
from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Session

from app.db import init_db, get_session
from app.auth import get_current_user, generate_api_key
from app.models import User
from app.metering import (
    count_tokens,
    get_or_create_quota,
    record_usage,
)
from app.quota import check_rate_limit, check_token_quota
from app.llm import choose_model, call_llm, get_cached, set_cached
from app.billing import calc_revenue_for_request, user_profit_report
from app.pricing import PLANS

from pydantic import BaseModel

app = FastAPI(title="AI Monetization SaaS")


class GenerateRequest(BaseModel):
    prompt: str
    task_type: str = "standard"   # simple / standard / complex


class GenerateResponse(BaseModel):
    request_id: str
    text: str
    model: str
    prompt_tokens: int
    completion_tokens: int
    cost_usd: float
    cached: bool


@app.on_event("startup")
def on_startup():
    init_db()


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


@app.post("/signup")
def signup(email: str, session: Session = Depends(get_session)):
    user = User(email=email, api_key=generate_api_key())
    session.add(user)
    session.commit()
    session.refresh(user)
    return {"user_id": user.id, "api_key": user.api_key}


@app.post("/generate", response_model=GenerateResponse)
def generate(
    payload: GenerateRequest,
    user: User = Depends(get_current_user),
    session: Session = Depends(get_session),
):
    plan = PLANS.get(user.plan, PLANS["free"])

    # 1. 限流
    check_rate_limit(user.id, user.plan)

    # 2. 配额检查
    quota = get_or_create_quota(session, user.id)
    check_token_quota(quota.used_tokens, user.plan)

    # 3. 模型路由
    model = choose_model(payload.task_type, user.plan)

    # 4. 缓存检查
    cached = get_cached(payload.prompt, model)
    if cached:
        revenue = calc_revenue_for_request(user.plan)
        record_usage(
            session,
            user.id,
            cached.model,
            cached.prompt_tokens,
            cached.completion_tokens,
            request_id=str(uuid.uuid4()),
            revenue_usd=revenue,
        )
        return GenerateResponse(
            request_id=str(uuid.uuid4()),
            text=cached.text,
            model=cached.model,
            prompt_tokens=cached.prompt_tokens,
            completion_tokens=cached.completion_tokens,
            cost_usd=0.0,
            cached=True,
        )

    # 5. 调用模型
    resp = call_llm(payload.prompt, model)
    set_cached(payload.prompt, model, resp)

    # 6. 计量与计费
    revenue = calc_revenue_for_request(user.plan)
    record = record_usage(
        session,
        user.id,
        resp.model,
        resp.prompt_tokens,
        resp.completion_tokens,
        request_id=str(uuid.uuid4()),
        revenue_usd=revenue,
    )

    return GenerateResponse(
        request_id=record.request_id,
        text=resp.text,
        model=resp.model,
        prompt_tokens=resp.prompt_tokens,
        completion_tokens=resp.completion_tokens,
        cost_usd=record.cost_usd,
        cached=False,
    )


@app.get("/me/profit")
def my_profit(
    user: User = Depends(get_current_user),
    session: Session = Depends(get_session),
):
    return user_profit_report(session, user.id)


@app.get("/me/quota")
def my_quota(
    user: User = Depends(get_current_user),
    session: Session = Depends(get_session),
):
    quota = get_or_create_quota(session, user.id)
    plan = PLANS.get(user.plan, PLANS["free"])
    return {
        "plan": user.plan,
        "period": quota.period,
        "used_tokens": quota.used_tokens,
        "included_tokens": plan.included_tokens,
        "used_requests": quota.used_requests,
        "monthly_price_usd": plan.monthly_price_usd,
    }

import os
import pytest
from fastapi.testclient import TestClient

os.environ["DATABASE_URL"] = "sqlite:///./test_ai_monetization.db"

from app.main import app  # noqa: E402
from app.pricing import calc_llm_cost, PLANS  # noqa: E402

client = TestClient(app)


def test_cost_calculation():
    cost = calc_llm_cost("gpt-4o-mini", 1_000_000, 1_000_000)
    assert cost == pytest.approx(0.75, rel=1e-3)


def test_signup_and_generate():
    r = client.post("/signup", params={"email": "test@example.com"})
    assert r.status_code == 200
    api_key = r.json()["api_key"]

    headers = {"X-API-Key": api_key}
    r = client.post(
        "/generate",
        json={"prompt": "Hello AI", "task_type": "simple"},
        headers=headers,
    )
    assert r.status_code == 200
    data = r.json()
    assert data["model"] == "gpt-4o-mini"
    assert data["prompt_tokens"] > 0


def test_quota_and_profit():
    r = client.post("/signup", params={"email": "profit@example.com"})
    api_key = r.json()["api_key"]
    headers = {"X-API-Key": api_key}

    for _ in range(3):
        client.post(
            "/generate",
            json={"prompt": "Test", "task_type": "simple"},
            headers=headers,
        )

    r = client.get("/me/quota", headers=headers)
    assert r.status_code == 200
    assert r.json()["used_requests"] >= 3

    r = client.get("/me/profit", headers=headers)
    assert r.status_code == 200
    assert "gross_margin" in r.json()


def test_rate_limit_free_plan():
    r = client.post("/signup", params={"email": "ratelimit@example.com"})
    api_key = r.json()["api_key"]
    headers = {"X-API-Key": api_key}

    statuses = []
    for _ in range(10):
        r = client.post(
            "/generate",
            json={"prompt": "x", "task_type": "simple"},
            headers=headers,
        )
        statuses.append(r.status_code)

    assert 429 in statuses

pytest -v

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app ./app

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

docker build -t ai-monetization .
docker run -p 8000:8000 ai-monetization

简单任务 -> mini 模型
标准任务 -> 中端模型
复杂任务 -> 高端模型

def cost_alert(session, user_id: int, threshold_usd: float = 5.0):
    from sqlmodel import select, func
    from app.models import UsageRecord

    total = session.exec(
        select(func.sum(UsageRecord.cost_usd))
        .where(UsageRecord.user_id == user_id)
    ).one() or 0.0

    if total > threshold_usd:
        return {
            "alert": True,
            "user_id": user_id,
            "cost_usd": round(total, 4),
            "message": "用户成本超过阈值，建议调整定价或限制用量",
        }
    return {"alert": False, "cost_usd": round(total, 4)}

import os
import stripe

stripe.api_key = os.getenv("STRIPE_API_KEY", "")

PRICE_IDS = {
    "pro": os.getenv("STRIPE_PRICE_PRO", "price_pro"),
    "business": os.getenv("STRIPE_PRICE_BUSINESS", "price_business"),
}


def create_checkout_session(
    customer_email: str,
    plan: str,
    success_url: str,
    cancel_url: str,
):
    if plan not in PRICE_IDS:
        raise ValueError(f"Unknown plan: {plan}")

    session = stripe.checkout.Session.create(
        mode="subscription",
        customer_email=customer_email,
        line_items=[{"price": PRICE_IDS[plan], "quantity": 1}],
        success_url=success_url,
        cancel_url=cancel_url,
        metadata={"plan": plan},
    )
    return session.url


def create_usage_record(subscription_item_id: str, quantity: int):
    """
    按量计费：向 Stripe 上报用量。
    """
    stripe.SubscriptionItem.create_usage_record(
        subscription_item_id,
        quantity=quantity,
        action="increment",
    )

from fastapi import Request

@app.post("/stripe/webhook")
async def stripe_webhook(request: Request):
    payload = await request.body()
    sig_header = request.headers.get("stripe-signature")
    webhook_secret = os.getenv("STRIPE_WEBHOOK_SECRET", "")

    try:
        event = stripe.Webhook.construct_event(
            payload, sig_header, webhook_secret
        )
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid webhook")

    if event["type"] == "checkout.session.completed":
        session = event["data"]["object"]
        # 更新用户套餐
        ...

    if event["type"] == "invoice.payment_failed":
        # 处理支付失败
        ...

    return {"received": True}

def detect_abuse(user_id: int, requests_last_minute: int, plan_limit: int):
    if requests_last_minute > plan_limit * 3:
        return {
            "abuse": True,
            "reason": "请求频率异常",
            "action": "temporary_block",
        }
    return {"abuse": False}

def business_dashboard(session):
    from sqlmodel import select, func
    from app.models import UsageRecord, User

    total_users = session.exec(select(func.count(User.id))).one()
    total_cost = session.exec(
        select(func.sum(UsageRecord.cost_usd))
    ).one() or 0.0
    total_revenue = session.exec(
        select(func.sum(UsageRecord.revenue_usd))
    ).one() or 0.0
    total_tokens = session.exec(
        select(func.sum(
            UsageRecord.prompt_tokens + UsageRecord.completion_tokens
        ))
    ).one() or 0

    gross_profit = total_revenue - total_cost
    margin = gross_profit / total_revenue if total_revenue > 0 else 0.0

    return {
        "total_users": total_users,
        "total_tokens": total_tokens,
        "total_cost_usd": round(total_cost, 2),
        "total_revenue_usd": round(total_revenue, 2),
        "gross_profit_usd": round(gross_profit, 2),
        "gross_margin": round(margin, 4),
    }

ai-monetization/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── db.py
│   ├── models.py
│   ├── auth.py
│   ├── metering.py
│   ├── pricing.py
│   ├── quota.py
│   ├── llm.py
│   ├── billing.py
│   └── billing_stripe.py
├── tests/
│   └── test_billing.py
├── requirements.txt
└── Dockerfile

pip install -r requirements.txt
pytest -v
uvicorn app.main:app --reload

http://127.0.0.1:8000/docs

用户价值 -> 单位经济 -> 计量计费 -> 支付 -> 风控 -> 利润监控 -> 增长