AI变现实战：从“卖Token”到“卖结果”的最小成本交付框架

import time, hashlib
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from openai import OpenAI

app = FastAPI()
client = OpenAI(api_key="sk-xxx", base_url="https://api.deepseek.com/v1")
PRICE_PER_1K_TOKENS = 0.002  # 对客户的定价

class Query(BaseModel):
    api_key: str
    prompt: str

@app.post("/v1/generate")
def generate(q: Query):
    if not valid_key(q.api_key):          # 简单的Key校验
        raise HTTPException(401, "Invalid key")
    resp = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": q.prompt}],
        max_tokens=512)
    tokens_used = resp.usage.total_tokens
    cost = tokens_used / 1000 * PRICE_PER_1K_TOKENS
    bill(q.api_key, cost)                  # 写入计费记录
    return {"result": resp.choices[0].message.content,
            "tokens": tokens_used, "cost": round(cost, 4)}

import chromadb, openai, hashlib

openai.api_key = "sk-xxx"
client = chromadb.PersistentClient(path="./chroma_db")
col = client.get_or_create_collection("kb")

def index(text: str, source: str):
    doc_id = hashlib.md5(source.encode()).hexdigest()
    old = col.get(where={"source": source})
    if old["ids"]:
        col.delete(ids=old["ids"])
    chunks = [text[i:i+500] for i in range(0, len(text), 450)]
    col.add(
        ids=[f"{doc_id}_{i}" for i in range(len(chunks))],
        documents=chunks,
        embeddings=[openai.embeddings.create(
            model="text-embedding-3-small",
            input=c).data[0].embedding for c in chunks],
        metadatas=[{"source": source}] * len(chunks))

def ask(question: str) -> dict:
    q_emb = openai.embeddings.create(
        model="text-embedding-3-small", input=question).data[0].embedding
    hits = col.query(query_embeddings=[q_emb], n_results=3)
    context = "\n".join(hits["documents"][0])
    resp = openai.chat.completions.create(
        model="gpt-4o-mini", temperature=0.1,
        messages=[
            {"role": "system", "content":
             "仅根据以下资料回答。不确定则回复'请联系人工'。\n" + context},
            {"role": "user", "content": question}])
    return {"answer": resp.choices[0].message.content,
            "sources": [m["source"] for m in hits["metadatas"][0]]}

import json
from openai import OpenAI

client = OpenAI(api_key="sk-xxx", base_url="https://api.deepseek.com/v1")

def audit_expense(expense: dict) -> dict:
    """输入报销单字段，返回审核结果与置信度"""
    prompt = f"""审核以下报销单，输出JSON：
    - 金额：{expense['amount']}元
    - 类别：{expense['category']}
    - 发票号：{expense['invoice_id']}
    - 申请人：{expense['submitter']}
规则：金额>5000需标记待复核；
类别为"餐饮"且金额>2000需附事由；
发票号重复则直接拒绝。
输出格式：{{"decision": "通过/拒绝/待复核",
"reasons": [...], "confidence": 0.0-1.0}}"""
    resp = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1,
        response_format={"type": "json_object"})
    return json.loads(resp.choices[0].message.content)