容量、并发、缓存与 FinOps

New-Item -ItemType Directory -Force l8-capacity
cd l8-capacity

mkdir -p l8-capacity
cd l8-capacity

"""用固定业务输入计算并发与费用；没有计时或真实模型调用。"""
from decimal import Decimal
from math import ceil

HIT_RATE = Decimal("0.50")

def plan(rate, latency, hit_rate, slot_limit):
    if rate < 0 or latency < 0 or not Decimal("0") <= hit_rate <= Decimal("1"):
        raise ValueError("速率、耗时非负，命中率在0到1之间")
    if slot_limit <= 0:
        raise ValueError("并发上限必须为正")
    # 仅未命中流量消耗模型并发，平均在途数为速率乘耗时。
    model_rate = rate * (1 - hit_rate)
    in_flight = model_rate * latency
    slots = ceil(in_flight)
    return model_rate, in_flight, slots, slots <= slot_limit

def cost(requests, hit_rate, call_price):
    # 单价是教学假设，每次未命中固定一次调用。
    return Decimal(requests) * (1 - hit_rate) * call_price

def cache_key(tenant, document_version, model, prompt_version, question):
    # 租户与版本参与键，授权仍须在查缓存之前完成。
    return tenant, document_version, model, prompt_version, question

def main():
    rate, in_flight, slots, fits = plan(Decimal("10"), Decimal("2"), HIT_RATE, 12)
    print(f"模型请求: {rate:.2f}次/秒")
    print(f"平均在途: {in_flight:.2f} / 向上取整{slots}个并发")
    print(f"12个槽位: {'容纳' if fits else '不足'}")
    print(f"1000次请求假设费用: {cost(1000, HIT_RATE, Decimal('0.01')):.2f}元")
    print("租户隔离:",
          cache_key("A", "v1", "m1", "p1", "报销") !=
          cache_key("B", "v1", "m1", "p1", "报销"))

if __name__ == "__main__":
    main()

"""覆盖全命中、无命中、容量边界、版本隔离与费用计算。"""
from decimal import Decimal as D
import unittest
from capacity_plan import plan, cost, cache_key

class CapacityTests(unittest.TestCase):
    def test_baseline(self):
        self.assertEqual(plan(D(10), D(2), D("0.5"), 12), (D(5), D(10), 10, True))

    def test_zero_and_full_hit(self):
        self.assertEqual(plan(D(10), D(2), D(0), 12)[2:], (20, False))
        self.assertEqual(plan(D(10), D(2), D(1), 12)[2:], (0, True))

    def test_capacity_boundary(self):
        self.assertTrue(plan(D(6), D(2), D(0), 12)[3])
        self.assertFalse(plan(D("6.1"), D(2), D(0), 12)[3])

    def test_invalid_inputs(self):
        for hit in (D("-0.1"), D("1.1")):
            with self.assertRaises(ValueError):
                plan(D(10), D(2), hit, 12)
        with self.assertRaises(ValueError):
            plan(D(10), D(2), D(0), 0)

    def test_cost_and_version_key(self):
        self.assertEqual(cost(1000, D("0.5"), D("0.01")), D("5"))
        self.assertNotEqual(cache_key("A", "v1", "m1", "p1", "q"),
                            cache_key("A", "v2", "m1", "p1", "q"))

if __name__ == "__main__":
    unittest.main()

ls
python -X utf8 capacity_plan.py
python -X utf8 -m unittest -v test_capacity_plan.py

模型请求: 5.00次/秒
平均在途: 10.00 / 向上取整10个并发
12个槽位: 容纳
1000次请求假设费用: 5.00元
租户隔离: True

python -X utf8 capacity_plan.py