《AI Agent 高级编程》10.2 成本与延迟治理：路由/缓存/并发 + 评测兜底

"""成本治理最小实测：语义缓存 + 复杂度路由 + 预算降级（本地假模型）。"""
from dataclasses import dataclass
import hashlib

@dataclass
class FakeModel:
    name: str
    unit_cost: int                        # 合成成本单位，非真实美元
    quality: dict[str, str]               # prompt -> answer（决定评测对错）
    calls: int = 0
    def complete(self, prompt: str) -> str:
        self.calls += 1
        return self.quality.get(prompt, "?")

class ResponseCache:
    def __init__(self):
        self.store: dict[str, str] = {}; self.hits = 0
    def key(self, prompt: str) -> str:
        return hashlib.sha256(prompt.encode()).hexdigest()[:10]
    def get(self, prompt):
        k = self.key(prompt)
        if k in self.store:
            self.hits += 1; return self.store[k]
        return None
    def put(self, prompt, ans):
        self.store[self.key(prompt)] = ans

@dataclass
class Router:
    big: FakeModel; small: FakeModel; budget: int
    spent: int = 0; degraded: int = 0
    def pick(self, prompt: str) -> FakeModel:
        complex_ = len(prompt) > 8
        if complex_ and self.spent + self.big.unit_cost <= self.budget:
            return self.big
        if complex_:
            self.degraded += 1            # 复杂请求但预算不够 -> 强制降级
        return self.small

cache = ResponseCache()
big = FakeModel("big", unit_cost=10,
                quality={"解释 CAP 定理": "cap", "总结这段长文本并给出要点": "sum"})
small = FakeModel("small", unit_cost=2,
                  quality={"解释 CAP 定理": "cap", "总结这段长文本并给出要点": "wrong"})
router = Router(big, small, budget=12)

def handle(prompt: str) -> str:
    hit = cache.get(prompt)
    if hit is not None:
        return hit                        # 命中：不路由、不调用模型
    m = router.pick(prompt)
    ans = m.complete(prompt); cache.put(prompt, ans); router.spent += m.unit_cost
    return ans

REQS = ["解释 CAP 定理", "总结这段长文本并给出要点", "解释 CAP 定理",
        "总结这段长文本并给出要点", "短问题"]
for p in REQS:
    handle(p)

缓存命中次数: 2
big 调用: 1  small 调用: 2
累计合成成本单位: 14 / 预算 12
预算不足触发降级次数: 1

"""并发上限：Semaphore 把在飞请求数压到上限之内。"""
import asyncio

class FakeModel:
    def __init__(self):
        self.calls = 0
    async def acomplete(self, prompt: str) -> str:
        self.calls += 1
        await asyncio.sleep(0.01)          # 模拟一次模型往返
        return prompt.upper()

async def main():
    model = FakeModel(); limit = 3
    sem = asyncio.Semaphore(limit)
    in_flight = 0; peak = 0
    async def guarded(p):
        nonlocal in_flight, peak
        async with sem:                    # 超过 limit 的请求在此排队
            in_flight += 1; peak = max(peak, in_flight)
            try:
                return await model.acomplete(p)
            finally:
                in_flight -= 1
    reqs = [f"q{i}" for i in range(10)]
    await asyncio.gather(*(guarded(r) for r in reqs))
    print("完成请求数:", 10)
    print("模型调用次数:", model.calls)
    print(f"并发上限 limit={limit}，实测峰值在飞数: {peak}")

asyncio.run(main())

完成请求数: 10
模型调用次数: 10
并发上限 limit=3，实测峰值在飞数: 3

EVAL = {"解释 CAP 定理": "cap", "总结这段长文本并给出要点": "sum"}

def score(model) -> float:
    return sum(1 for q, g in EVAL.items() if model.quality.get(q) == g) / len(EVAL)

print(f"big 评测通过率: {score(big):.0%}   small 评测通过率: {score(small):.0%}")

big 评测通过率: 100%   small 评测通过率: 50%
=> 降级把合成通过率从 100% 拉到 50%，必须靠评测门禁兜底