Codex 如何使用和优化 Prompt Caching：深度技术解析

┌─────────────────────────────────────────────────────────┐
│                    请求 Token 序列                        │
├─────────────────────────┬───────────────────────────────┤
│    缓存前缀 (Cache Hit)  │     新增后缀 (需计算)          │
│  系统指令 + 工具定义 +    │   用户输入 + 新消息 +          │
│  环境上下文 + 历史消息    │   动态内容                     │
└─────────────────────────┴───────────────────────────────┘
         ↓                              ↓
   复用 KV 张量                   执行注意力计算
   (零计算开销)                   (正常推理)

┌──────────────────────────────────────────────────────────┐
│                    Codex Agent Loop                       │
│                                                          │
│  ┌─────────┐    ┌─────────┐    ┌─────────┐              │
│  │ 思考/规划 │───▶│ 工具调用 │───▶│ 观察结果 │              │
│  └─────────┘    └─────────┘    └─────────┘              │
│       ▲                                    │             │
│       └────────────────────────────────────┘             │
│                    迭代直到完成                            │
└──────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                    Codex 请求结构                              │
├─────────────────────────────────────────────────────────────┤
│  ① 系统指令 (System Instructions)           ← 稳定，可缓存   │
│  ② 工具定义 (Tool Definitions)              ← 稳定，可缓存   │
│  ③ 沙箱配置 (Sandbox Configuration)         ← 稳定，可缓存   │
│  ④ 环境上下文 (Environment Context)          ← 稳定，可缓存   │
│  ⑤ 历史消息 (Previous Messages)             ← 追加，可缓存   │
│  ⑥ 新用户输入 (New User Input)              ← 动态，不缓存   │
└─────────────────────────────────────────────────────────────┘

# 完整工具集保持在缓存前缀中（静态）：
tools = [get_weather_def, get_location_def, calendar_def, ...]

# 每次调用通过 allowed_tools 限制可用工具（在请求元数据中，不在前缀中）：
allowed_tools = {"mode": "auto", "tools": ["get_weather", "get_location"]}

# Codex 的 Compaction 调用
response = client.responses.create(
    model="gpt-5.2-codex",
    input=conversation,
    store=False,
    context_management=[{
        "type": "compaction",
        "compact_threshold": 100000
    }],
)

┌─────────────────────────────────────────────────────────┐
│              Compaction 对缓存的影响                       │
│                                                         │
│  未压缩时：                                              │
│  [系统指令][工具][历史消息1][历史消息2][历史消息3][新输入]  │
│  ↑ 缓存命中 ↑                                            │
│                                                         │
│  压缩后：                                                │
│  [系统指令][工具][压缩摘要][历史消息3][新输入]              │
│  ↑ 缓存失效 ↑  (前缀结构改变)                             │
└─────────────────────────────────────────────────────────┘

# 稳定的请求结构
messages = [
    {"role": "system", "content": SYSTEM_INSTRUCTIONS},  # 固定
    {"role": "system", "content": TOOL_DEFINITIONS},      # 固定
    {"role": "system", "content": ENVIRONMENT_CONTEXT},   # 固定
    # ... 历史消息追加 ...
    {"role": "user", "content": user_input}               # 动态
]

# 在前缀中插入动态内容
messages = [
    {"role": "system", "content": f"当前时间: {datetime.now()}"},  # 破坏缓存！
    {"role": "system", "content": SYSTEM_INSTRUCTIONS},
    # ...
]

{
  "model": "gpt-5.1",
  "input": "Write me a haiku...",
  "prompt_cache_retention": "24h"
}

{
  "event": "session.update",
  "session": {
    "truncation": {
      "type": "retention_ratio",
      "retention_ratio": 0.7
    }
  }
}

from openai import OpenAI

client = OpenAI()

response = client.responses.create(
    model="gpt-5.2",
    input="Explain quantum computing...",
)

# 检查缓存命中
cached_tokens = response.usage.prompt_tokens_details.cached_tokens
total_tokens = response.usage.prompt_tokens
cache_hit_rate = cached_tokens / total_tokens * 100

print(f"缓存命中: {cached_tokens}/{total_tokens} tokens ({cache_hit_rate:.1f}%)")

{
  "usage": {
    "prompt_tokens": 2006,
    "completion_tokens": 300,
    "total_tokens": 2306,
    "prompt_tokens_details": {
      "cached_tokens": 1920
    }
  }
}

from openai import OpenAI

client = OpenAI()

# 稳定的系统指令（缓存友好）
SYSTEM_INSTRUCTIONS = """You are a coding assistant. You can:
- Read and write files
- Run terminal commands
- Search the web

Always follow these steps:
1. Understand the task
2. Explore the codebase
3. Plan your changes
4. Implement and test
5. Commit your work"""

# 稳定的工具定义（缓存友好）
TOOLS = [
    {"type": "function", "function": {"name": "run_command", ...}},
    {"type": "function", "function": {"name": "read_file", ...}},
    {"type": "function", "function": {"name": "write_file", ...}},
]

def agent_loop(user_request: str, max_iterations: int = 10):
    """Codex 风格的 Agent Loop"""
    conversation = [
        {"role": "system", "content": SYSTEM_INSTRUCTIONS},
        {"role": "user", "content": user_request}
    ]
    
    for i in range(max_iterations):
        response = client.responses.create(
            model="gpt-5.2-codex",
            input=conversation,
            tools=TOOLS,
            # 使用 prompt_cache_key 提高路由粘性
            prompt_cache_key="my-coding-agent",
        )
        
        # 检查是否有工具调用
        if response.output[0].type == "function_call":
            # 执行工具
            result = execute_tool(response.output[0])
            # 追加结果（不修改历史，保持缓存）
            conversation.append({"role": "assistant", "output": response.output})
            conversation.append({"role": "tool", "output": result})
        else:
            # 完成
            return response.output[0].content
    
    return "达到最大迭代次数"

# 完整工具集（静态，保持在缓存前缀中）
ALL_TOOLS = [
    {"type": "function", "function": {"name": "read_file", ...}},
    {"type": "function", "function": {"name": "write_file", ...}},
    {"type": "function", "function": {"name": "run_command", ...}},
    {"type": "function", "function": {"name": "search_web", ...}},
    {"type": "function", "function": {"name": "git_commit", ...}},
]

def create_response(conversation, allowed_tool_names: list[str]):
    """使用 allowed_tools 限制可用工具，同时保持缓存"""
    return client.responses.create(
        model="gpt-5.2",
        input=conversation,
        tools=ALL_TOOLS,
        tool_choice={
            "mode": "auto",
            "tools": allowed_tool_names  # 动态限制，不影响缓存前缀
        }
    )

def smart_conversation_management(conversation: list, threshold: int = 100000):
    """智能对话管理：平衡压缩与缓存"""
    
    # 估算当前 token 数
    current_tokens = estimate_tokens(conversation)
    
    if current_tokens < threshold:
        # 未达阈值，保持原样（维护缓存）
        return conversation
    
    # 达到阈值，执行 compaction
    response = client.responses.create(
        model="gpt-5.2-codex",
        input=conversation,
        store=False,
        context_management=[{
            "type": "compaction",
            "compact_threshold": threshold
        }],
    )
    
    # 返回压缩后的对话（包含 encrypted_content）
    return response.output

# 1. 检查单个请求的缓存状态
print(f"Cached tokens: {response.usage.prompt_tokens_details.cached_tokens}")

# 2. 使用 Usage Dashboard 查看整体缓存率
# 在 OpenAI 控制台中筛选 cached/uncached tokens

# 3. 对比请求前缀
def compare_prefixes(req1_messages, req2_messages):
    """找出前缀分歧点"""
    for i, (m1, m2) in enumerate(zip(req1_messages, req2_messages)):
        if m1 != m2:
            print(f"分歧点: 消息 {i}")
            print(f"  请求1: {m1}")
            print(f"  请求2: {m2}")
            return
    print("前缀一致")

# ❌ 过于细粒度（每请求一个 key，无复用）
prompt_cache_key=f"request-{uuid4()}"

# ❌ 过于粗粒度（所有用户共享，容易溢出）
prompt_cache_key="global"

# ✅ 按用户分组
prompt_cache_key=f"user-{user_id}"

# ✅ 按会话分组
prompt_cache_key=f"conversation-{conversation_id}"

# ✅ 按用户组分组（平衡命中率与扩展性）
group_id = hash(user_id) % 10  # 10 个 bucket
prompt_cache_key=f"group-{group_id}"