成本、Token 与延迟分析

mkdir l7-cost
cd l7-cost

"""从固定调用记录计算用量、已知费用、并发关键路径和分位数。"""

from decimal import Decimal
from math import ceil

# 示例计价参数：每百万输入 1 美元，每百万输出 3 美元。
INPUT_RATE = Decimal("1")
OUTPUT_RATE = Decimal("3")
CALLS = [
    {"id": "draft", "input": 1000, "output": 200},
    {"id": "final", "input": 600, "output": None},
]
TASKS = {
    "plan": {"ms": 40, "deps": []},
    "draft": {"ms": 300, "deps": ["plan"]},
    "stock": {"ms": 200, "deps": ["draft"]},
    "policy": {"ms": 120, "deps": ["draft"]},
    "final": {"ms": 180, "deps": ["stock", "policy"]},
}
LATENCIES = [400, 500, 600, 700, 800, 900, 1000, 1200, 1500, 3000]


def usage_cost(calls):
    known = Decimal("0")
    missing = []
    totals = {"input": 0, "output": 0}
    for call in calls:
        for field in totals:
            value = call[field]
            if value is not None:
                if type(value) is not int or value < 0:
                    raise ValueError("usage must be a nonnegative integer")
                totals[field] += value
        # 缺用量的调用不按零收费；完整费用总额仍未知。
        if call["input"] is None or call["output"] is None:
            missing.append(call["id"])
        else:
            known += (Decimal(call["input"]) * INPUT_RATE
                      + Decimal(call["output"]) * OUTPUT_RATE) / Decimal(1_000_000)
    return known, missing, totals


def critical_path(tasks):
    memo = {}
    visiting = set()
    def finish(name):
        if name in visiting:
            raise ValueError("dependency cycle")
        if name in memo:
            return memo[name]
        task = tasks[name]
        if type(task["ms"]) is not int or task["ms"] < 0:
            raise ValueError("invalid duration")
        visiting.add(name)
        # 各依赖可并行执行，开始时间取最晚的依赖完成时间。
        choices = [finish(dep) for dep in task["deps"]]
        prior_ms, prior_path = max(choices, key=lambda row: row[0], default=(0, []))
        visiting.remove(name)
        memo[name] = (prior_ms + task["ms"], prior_path + [name])
        return memo[name]
    return max((finish(name) for name in tasks), key=lambda row: row[0])


def percentile(values, p):
    # nearest-rank：先排序，再取第 ceil(p*n) 个，序号从 1 开始。
    if not values or not 0 < p <= 1:
        raise ValueError("invalid percentile")
    ordered = sorted(values)
    return ordered[ceil(p * len(ordered)) - 1]


def main():
    known, missing, totals = usage_cost(CALLS)
    wall_ms, path = critical_path(TASKS)
    print(f"known_input={totals['input']} known_output={totals['output']}")
    total = f"{known:.6f}" if not missing else "unknown"
    print(f"complete_call_cost_usd={known:.6f} missing_usage={','.join(missing)} total={total}")
    print(f"sum_task_ms={sum(t['ms'] for t in TASKS.values())} critical_path_ms={wall_ms}")
    print(f"critical_path={' -> '.join(path)}")
    print(f"latency_n={len(LATENCIES)} mean_ms={sum(LATENCIES) / len(LATENCIES):.1f} "
          f"p50_ms={percentile(LATENCIES, .50)} p95_ms={percentile(LATENCIES, .95)}")


if __name__ == "__main__":
    main()

ls
python -X utf8 cost_lab.py

known_input=1600 known_output=200
complete_call_cost_usd=0.001600 missing_usage=final total=unknown
sum_task_ms=840 critical_path_ms=720
critical_path=plan -> draft -> stock -> final
latency_n=10 mean_ms=1060.0 p50_ms=800 p95_ms=3000

ls
python -X utf8 -m unittest -v test_cost_lab.py