KV Cache：多轮对话的隐性账单

def kv_cache_bytes(
    batch: int,        # 并发请求数
    seq_len: int,      # 序列长度（含历史）
    n_layers: int,     # 层数
    n_kv_heads: int,   # K/V 头数（注意：不一定等于注意力头数）
    head_dim: int,     # 每个头的维度
    bytes_per_num: int = 2,   # fp16=2, fp8=1
) -> int:
    # 每层缓存 K 和 V 两份，所以乘 2
    return 2 * batch * seq_len * n_layers * n_kv_heads * head_dim * bytes_per_num
# 一个典型的 70B 级模型（GQA，kv_heads 远少于 query_heads）
total = kv_cache_bytes(
    batch=32, seq_len=32_000, n_layers=80, n_kv_heads=8, head_dim=128
)
print(f"{total / 1024**3:.1f} GiB")   # 大约 31 GiB —— 仅缓存，不含权重
# 关键观察：这个数是随 seq_len 线性长的
for s in (8_000, 32_000, 128_000):
    n = kv_cache_bytes(32, s, 80, 8, 128)
    print(f"seq_len={s:>7}: {n/1024**3:6.1f} GiB")

import json, hashlib
from typing import Any
def serialize_stable(obj: Any) -> str:
    """稳定序列化：键排序、分隔符固定、禁用 ascii 转义以保证同一对象字节一致"""
    return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
def prefix_fingerprint(messages: list[dict], tools: list[dict]) -> str:
    """对 tools + 前 N-1 条消息取指纹，验证相邻两轮是否共享同一前缀"""
    stable_part = {
        "tools": tools,                 # 工具定义必须在 messages 之前，且顺序稳定
        "history": messages[:-1],       # 最后一条是新增内容，不参与指纹
    }
    return hashlib.sha256(serialize_stable(stable_part).encode()).hexdigest()[:16]
def append_only_messages(system: str, history: list[dict], user_input: str) -> list[dict]:
    """唯一推荐的组装方式：头部固定，只在尾部追加"""
    fixed_head = [{"role": "system", "content": system}]
    return fixed_head + history + [{"role": "user", "content": user_input}]
# --- 使用示例：连续两轮，观察指纹是否一致 ---
tools = [{"name": "get_weather"}, {"name": "get_time"}]   # 注意：顺序必须写死
h1 = [{"role": "user", "content": "北京天气"}]
fp1 = prefix_fingerprint(h1, tools)
h2 = h1 + [{"role": "assistant", "content": "正在查询"},
           {"role": "tool", "content": "晴 24℃"}]
fp2 = prefix_fingerprint(h2, tools)   # 注意这里的「-1」语义：最后一轮的新输入不入指纹
print(fp1, fp2)   # 这里的差异是因为 history 长度变了，实际检查时应比较「相同长度部分」