04 · 调用层

# llm.py —— 调用层的地基：统一返回结构 + 假客户端
import time
import random
from dataclasses import dataclass


@dataclass
class LLMResult:            # ⭐ 业务代码只认这个形状，永远不认某个 SDK 的返回对象
    text: str
    model: str
    in_tokens: int
    out_tokens: int
    latency_ms: int
    cached: bool = False


class FakeLLM:
    """不联网、不花钱、结果可预测。测试、CI、本地开发共用这一个。"""

    def __init__(self, reply="（假模型）收到。", latency=0.05, seed=0):
        self.reply, self.latency = reply, latency
        self.calls = []                  # ⭐ 记下每次调用，测试里可以直接断言
        self._rng = random.Random(seed)

    def complete(self, prompt, model="fake-1", temperature=0.0, timeout=30):
        t0 = time.perf_counter()
        self.calls.append({"prompt": prompt, "temperature": temperature})
        time.sleep(self.latency)
        # ⭐ temperature > 0 就每次不一样 —— 让"不确定性"在测试里也是真的
        text = self.reply if temperature == 0 else self.reply + str(self._rng.randint(100, 999))
        return LLMResult(text, model, len(prompt) // 3 + 1, len(text) // 3 + 1,
                         int((time.perf_counter() - t0) * 1000))


if __name__ == "__main__":
    llm = FakeLLM()
    a, b = (llm.complete("同一个问题", temperature=0.7).text for _ in range(2))
    print(a, b, "｜两次相同吗：", a == b)      # ⭐ False —— 断言 == 的测试写不出来

# 🗓️ 未实跑 —— 需要 httpx 和真实网络。要看的是【形状】：两个数分开设。
import httpx

# ⚠️ read 这个数按【最长的一次合法静默】定，不是拍脑袋：流式 + 工具调用时它可能是几十秒
timeout = httpx.Timeout(connect=5.0, read=60.0, write=10.0, pool=5.0)   # ⭐ 分开设
client = httpx.Client(timeout=timeout, limits=httpx.Limits(max_connections=100))

# retry.py —— 哪些错该重试、退避怎么算（可直接跑，看得见退避序列）
import random
import time


class RateLimited(Exception):    # 上游 429
    pass


class UpstreamError(Exception):  # 上游 5xx：500 / 502 / 503 / 504，通常是瞬时的
    pass


class BadRequest(Exception):     # 400：参数错、超长、被内容策略拒绝
    pass


# ⭐ 白名单，不是黑名单。⚠️ 上面那张表里"该重试"的四类，这里必须一个不少地对上：
#    429 → RateLimited｜5xx → UpstreamError｜超时 → TimeoutError｜网络错误 → ConnectionError
RETRIABLE = (RateLimited, UpstreamError, TimeoutError, ConnectionError)


def call_with_retry(fn, tries=4, base=0.5, cap=8.0, sleep=time.sleep, rng=None):
    rng = rng or random.Random(0)
    for i in range(tries):
        try:
            return fn()
        except RETRIABLE as e:
            if i == tries - 1:
                raise
            wait = rng.uniform(0, min(cap, base * 2 ** i))     # ⭐ 全抖动，别让大家一起回来
            print("  第 %d 次失败(%s)，等 %.2f 秒" % (i + 1, type(e).__name__, wait))
            sleep(wait)
        # ⭐ 白名单之外的异常（如 BadRequest）根本不进 except，直接往上抛


if __name__ == "__main__":
    n = [0]

    def flaky():                      # 第一次被限流，第二次上游 5xx —— 两类都在白名单里
        n[0] += 1
        if n[0] == 1:
            raise RateLimited("429")
        if n[0] == 2:
            raise UpstreamError("503 上游故障")
        return "第 %d 次成功" % n[0]

    def always_400():
        raise BadRequest("400 prompt 超长")

    print("可重试：", call_with_retry(flaky, sleep=lambda s: None))   # 演示时不真睡
    try:
        call_with_retry(always_400, sleep=lambda s: None)
    except BadRequest as e:
        print("不可重试，一次就抛出来：", e)

# cost.py —— 每次调用都记 token 和钱。⚠️ 单价是【占位数字】，用前去官方价格页抄一遍
PRICES = {"small": {"in": 2.0, "out": 8.0},          # 🗓️ 元 / 百万 token
          "big": {"in": 20.0, "out": 60.0}}


def cost_of(model, in_tokens, out_tokens):
    p = PRICES.get(model)
    if p is None:
        return None        # ⭐ 不认识的模型返回 None，别默默算成 0 —— 新模型的账会凭空消失
    return (in_tokens * p["in"] + out_tokens * p["out"]) / 1_000_000


def record(trace_id, user_id, feature, model, in_tok, out_tok, ms, cached=False, retries=0):
    """最小字段集。少任何一个，事后都有一类问题查不了：
    user_id 少了答不出"谁把钱花光的"，⭐ feature 少了答不出"哪个功能在烧钱"，
    model 少了答不出"换模型省了多少"，
    cached 少了算不出缓存省了多少，retries 少了看不见重试烧掉的那笔账。"""
    return {"trace_id": trace_id, "user_id": user_id, "feature": feature, "model": model,
            "in": in_tok, "out": out_tok, "ms": ms, "cached": cached, "retries": retries,
            "cost": cost_of(model, in_tok, out_tok)}


if __name__ == "__main__":
    for m in ("small", "big", "忘了配单价的新模型"):
        print(m, "→", record("t1", "u_9", "摘要", m, 1200, 800, 940)["cost"])

# cache.py —— 什么能缓存、key 里必须放什么（可直接跑）
import hashlib
import json


def cacheable(temperature, stream=False, tools=None):
    """⭐ 判据只有一条：同样的输入，是不是【应该】得到同样的输出。"""
    return temperature == 0 and not stream and not tools


def cache_key(*, model, prompt, system="", temperature=0.0, doc_ids=()):
    """⚠️ 凡是会影响输出的东西都必须进 key —— 漏一个就是把别人的答案发给这个人。"""
    blob = json.dumps({"model": model, "system": system, "prompt": prompt,
                       "temperature": temperature,
                       "docs": sorted(doc_ids)},        # ⭐ 检索到的文档也影响输出
                      sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:32]


if __name__ == "__main__":
    print("能缓存吗：t=0", cacheable(0.0), "｜t=0.7", cacheable(0.7),
          "｜带工具", cacheable(0.0, tools=["搜索"]))
    k = [cache_key(model="s", prompt="这份合同有什么风险", doc_ids=[d]) for d in "AB"]
    print("同问题、不同文档，key 相同吗：", k[0] == k[1])   # ⭐ 必须 False