AI 工作流专栏 25：成本治理与质量评估

单次调用成本 = (输入 token 数 / 1,000,000 × 输入单价) + (输出 token 数 / 1,000,000 × 输出单价)
月成本 = 单次调用成本 × 日均调用次数 × 30

# cost_calculator.py 通用 LLM 成本计算器
MODEL_PRICING = {
    "gpt-4o":        {"input": 2.50, "output": 10.00},
    "gpt-4o-mini":   {"input": 0.15, "output": 0.60},
    "deepseek-chat": {"input": 0.27, "output": 0.42},
    "qwen3-local":   {"input": 0.00, "output": 0.00},   # 仅 GPU 折旧
}

def calc_single_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    p = MODEL_PRICING[model]
    return (input_tokens / 1_000_000 * p["input"]) + (output_tokens / 1_000_000 * p["output"])

def calc_monthly_cost(model: str, daily_calls: int, avg_input: int, avg_output: int, days: int = 30) -> float:
    return calc_single_cost(model, avg_input, avg_output) * daily_calls * days

# model_router.py 模型分级路由
import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def classify_complexity(question: str) -> str:
    """用便宜的小模型判断复杂度，返回 simple/medium/complex"""
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "判断用户问题的复杂度，只输出一个词：simple/medium/complex。"
                                          "simple=查信息/闲聊；medium=需要多步推理；complex=数学/编程/复杂分析。"},
            {"role": "user", "content": question},
        ],
        temperature=0,
    )
    return resp.choices[0].message.content.strip().lower()

ROUTE_TABLE = {
    "simple":  "deepseek-chat",
    "medium":  "deepseek-chat",
    "complex": "gpt-4o",
}

def smart_chat(question: str) -> str:
    level = classify_complexity(question)
    model = ROUTE_TABLE.get(level, "deepseek-chat")
    print(f"[路由] 复杂度={level} -> 选用 {model}")
    # 实际项目中通过统一的模型适配层调用对应模型
    return f"（用 {model} 回答了：{question}）"

# semantic_cache.py 语义缓存
import os, json
import redis
from openai import OpenAI

r = redis.Redis(host="localhost", port=6379, decode_responses=True)
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

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

def find_cached_answer(question_emb: list, threshold: float = 0.92):
    """在缓存里找语义相似的问题，生产环境建议用向量库做近邻检索"""
    for key in r.keys("cache:*"):
        cached_emb = json.loads(r.hget(key, "embedding"))
        sim = cosine_similarity(question_emb, cached_emb)  # 余弦相似度
        if sim >= threshold:
            return r.hget(key, "answer"), sim
    return None, 0

def chat_with_cache(question: str) -> str:
    q_emb = get_embedding(question)
    cached, sim = find_cached_answer(q_emb)
    if cached:
        print(f"[缓存命中] 相似度={sim:.3f}，省一次 LLM 调用")
        return cached
    answer = call_llm(question)
    r.hset(f"cache:{hash(question)}", mapping={"embedding": json.dumps(q_emb), "answer": answer})
    r.expire(f"cache:{hash(question)}", 86400)
    return answer

# cost_monitor.py 成本监控装饰器
import time
from datetime import datetime

COST_LOG = []

def track_cost(model: str, user_id: str, endpoint: str):
    """包装 LLM 调用，自动记录每次成本，生产环境可接入 Prometheus 等监控系统"""
    def decorator(fn):
        def wrapper(*args, **kwargs):
            start = time.time()
            resp = fn(*args, **kwargs)
            usage = resp.usage
            cost = calc_single_cost(model, usage.prompt_tokens, usage.completion_tokens)
            COST_LOG.append({
                "ts": datetime.now().isoformat(),
                "model": model, "user_id": user_id, "endpoint": endpoint,
                "cost": cost, "latency": time.time() - start,
            })
            return resp
        return wrapper
    return decorator

# golden_set_eval.py 跑一遍测试集，算通过率
import os, json
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

GOLDEN_SET = [
    {"id": "q1", "question": "你们的营业时间是什么？",
     "expected_keywords": ["9:00", "18:00", "周一至周五"], "category": "simple"},
    {"id": "q2", "question": "我上周买的耳机有杂音，能退货吗？",
     "expected_keywords": ["7天", "退货", "质量问题"], "category": "policy"},
]

def get_ai_answer(question: str) -> str:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "你是客服助手，基于公司知识回答。"},
            {"role": "user", "content": question},
        ],
    )
    return resp.choices[0].message.content

def evaluate_single(answer: str, expected_keywords: list) -> bool:
    return all(kw in answer for kw in expected_keywords)

def run_golden_eval() -> dict:
    results, passed = [], 0
    for item in GOLDEN_SET:
        answer = get_ai_answer(item["question"])
        ok = evaluate_single(answer, item["expected_keywords"])
        results.append({**item, "answer": answer, "passed": ok})
        passed += ok
        print(f"[{item['id']}] {'通过' if ok else '失败'} | {item['question']}")
    return {"pass_rate": passed / len(GOLDEN_SET), "total": len(GOLDEN_SET), "details": results}

if __name__ == "__main__":
    report = run_golden_eval()
    print(f"通过率: {report['pass_rate']:.1%}")
    with open("eval_report.json", "w") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)

# llm_as_judge.py 用强模型当裁判打分
import os, json
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

JUDGE_PROMPT = """你是一个严格的评分员。请根据用户问题和参考答案，给候选回答打分（1-5 分）。
评分维度：准确性（2分）、完整性（2分）、简洁性（1分）。
只输出 JSON：{"score": 数字, "reason": "一句话理由"}
"""

def judge_answer(question: str, candidate: str, reference: str) -> dict:
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": JUDGE_PROMPT},
            {"role": "user", "content": f"问题：{question}\n参考答案：{reference}\n候选回答：{candidate}"},
        ],
        response_format={"type": "json_object"},
        temperature=0,
    )
    return json.loads(resp.choices[0].message.content)

# ragas_eval.py 用 RAGAS 评估 RAG 系统（pip install ragas）
import os
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall

os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY")

eval_data = {
    "question": ["你们的退货政策是什么？"],
    "answer": ["商品售出 7 天内，非人为损坏可无理由退货。"],
    "contexts": [["退货政策：商品售出 7 天内可退货，需保留小票。"]],
    "ground_truth": ["7 天内非人为损坏可退货。"],
}

dataset = Dataset.from_dict(eval_data)
result = evaluate(dataset, metrics=[faithfulness, answer_relevancy, context_precision, context_recall])
print(result)

# regression_test.py 对比新旧两次评估报告
def compare_reports(old_report: dict, new_report: dict, threshold: float = 0.05) -> dict:
    old_pass = old_report["pass_rate"]
    new_pass = new_report["pass_rate"]
    delta = new_pass - old_pass
    status = "通过，允许上线" if delta >= -threshold else "出现回归，禁止上线"
    return {"old": old_pass, "new": new_pass, "delta": delta, "status": status}

# drift_monitor.py 每天定时跑测试集，追踪通过率变化
import os, json

DRIFT_THRESHOLD = 0.05
HISTORY_FILE = "eval_history.json"

def check_drift(today_pass_rate: float) -> dict:
    history = json.load(open(HISTORY_FILE)) if os.path.exists(HISTORY_FILE) else []
    if len(history) < 7:
        return {"status": "数据不足，继续积累"}
    recent = [h["pass_rate"] for h in history[-7:]]
    baseline = sum(recent) / len(recent)
    delta = today_pass_rate - baseline
    if delta < -DRIFT_THRESHOLD:
        return {"status": "警告：检测到质量漂移", "delta": delta,
                "排查方向": ["模型是否被厂商升级", "知识库是否陈旧", "用户提问分布是否变化"]}
    return {"status": "稳定", "baseline_7d": baseline, "delta": delta}