土拨鼠日里的 AI 智能体：在 Amazon Bedrock 上为 OpenAI 模型使用提示缓存

KEY = "support-agent:v7:gpt-6-astra"
INSTRUCTIONS = "Your full support policies and examples..."

# Rendered context: [provider context][TOOLS][INSTRUCTIONS] | [task]
# Cache starts at the beginning; the marker below chooses its end.
resp = client.responses.create(
    model="us.openai.gpt-6-astra",
    store=False,  # Response storage; prompt caching still works.
    include=["reasoning.encrypted_content"],
    prompt_cache_key=KEY,
    prompt_cache_options={"mode": "explicit", "ttl": "30m"},
    tools=TOOLS,  # Rendered before instructions; keep order stable.
    input=[
        {
            "role": "developer",
            "content": [{
                "type": "input_text",
                "text": INSTRUCTIONS,
                # Save through this whole block, including TOOLS.
                "prompt_cache_breakpoint": {"mode": "explicit"},
            }],
        },
        # User message is processed but not cached.
        {"role": "user", "content": task},
    ],
)

def ask_with_implicit_cache(history):
    return client.responses.create(
        model="us.openai.gpt-6-astra",
        store=False,
        include=["reasoning.encrypted_content"],
        prompt_cache_key=KEY,
        prompt_cache_options={"mode": "implicit", "ttl": "30m"},
        tools=TOOLS,
        input=history,  # Everything gets cached
    )

history = [
    {"role": "developer", "content": INSTRUCTIONS},
    {"role": "user", "content": task},
]
# Automatic endpoint: the end of the user message.
resp = ask_with_implicit_cache(history)

history = [
    {
        "role": "developer",
        "content": [{
            "type": "input_text",
            "text": INSTRUCTIONS,
            "prompt_cache_breakpoint": {"mode": "explicit"},
        }],
    },
    {"role": "user", "content": task},
]
# Still implicit mode, now with an explicit startup save point in the history.
resp = ask_with_implicit_cache(history)

import json

# Map tool names to their Python implementations.
TOOL_HANDLERS = {
    "get_owner": get_owner,
    # Register your other initial tools here.
}

def execute_function(name, arguments):
    """Run the requested tool and return its result."""
    handler = TOOL_HANDLERS.get(name)
    if handler is None:
        return {"error": f"Unknown tool: {name}"}
    return handler(**arguments)

while True:
    # Preserve all output items, including encrypted reasoning.
    history.extend(resp.output)
    calls = [x for x in resp.output if x.type == "function_call"]
    if not calls:
        break
    for call in calls:
        result = execute_function(call.name, json.loads(call.arguments))
        history.append({
            "type": "function_call_output",
            "call_id": call.call_id,
            "output": json.dumps(result),
        })
    # The last result in this group supplies the automatic endpoint.
    resp = ask_with_implicit_cache(history)

import json

while True:
    resp = client.responses.create(
        model="us.openai.gpt-6-astra", store=False,
        include=["reasoning.encrypted_content"],
        tools=TOOLS, prompt_cache_key=KEY, input=history,
        prompt_cache_options={"mode": "explicit", "ttl": "30m"},
    )
    d = resp.usage.input_tokens_details
    print("read", d.cached_tokens, "write", d.cache_write_tokens)
    history.extend(resp.output)  # Includes encrypted reasoning.
    calls = [x for x in resp.output if x.type == "function_call"]
    if not calls:
        break
    batch = []
    for call in calls:
        result = execute_function(call.name, json.loads(call.arguments))
        batch.append({
            "type": "function_call_output", "call_id": call.call_id,
            "output": [{"type": "input_text", "text": json.dumps(result)}],
        })
    # Add the newest boundary; retain every earlier marker.
    batch[-1]["output"][-1]["prompt_cache_breakpoint"] = {"mode": "explicit"}
    history.extend(batch)

tool_choice={"type": "allowed_tools", "mode": "auto",
             "tools": [{"type": "function", "name": "get_owner"}]}

TOOL_HANDLERS["get_due_date"] = get_due_date
NEW_TOOL = {
    "type": "function", "name": "get_due_date",
    "description": "Return the due date for a support case.",
    "strict": True,
    "parameters": {
        "type": "object",
        "properties": {"case_id": {"type": "string"}},
        "required": ["case_id"], "additionalProperties": False,
    },
}
history.append({
    "type": "additional_tools", "role": "developer",
    "tools": [NEW_TOOL],
})
history.append({
    "role": "user", "content": "Find the due date for case C-123.",
})
# Implicit mode saves through this new user message, including NEW_TOOL.
resp = ask_with_implicit_cache(history)
d = resp.usage.input_tokens_details
print("old prefix read", d.cached_tokens, "new write", d.cache_write_tokens)

# Example output:
# old prefix read 2048 new write 256

shared = {"role": "developer", "content": [{
    "type": "input_text", "text": INSTRUCTIONS,
    "prompt_cache_breakpoint": {"mode": "explicit"},
}]}
warm = client.responses.create(
    model="us.openai.gpt-6-astra", store=False,
    tools=TOOLS, prompt_cache_key=KEY,
    prompt_cache_options={"mode": "implicit", "ttl": "30m"},
    max_output_tokens=16,
    input=[shared, {"role": "user", "content": "Reply READY."}],
)
d = warm.usage.input_tokens_details
if not (d.cache_write_tokens or d.cached_tokens):
    raise RuntimeError("Shared prefix was not cached.")
from concurrent.futures import ThreadPoolExecutor

# helper_tasks contains the task text for each helper.
helper_inputs = [
    [shared, {"role": "user", "content": task}]
    for task in helper_tasks
]
with ThreadPoolExecutor(max_workers=4) as pool:
    helpers = list(pool.map(ask_with_implicit_cache, helper_inputs))

from copy import deepcopy

# history is the completed history built by the agent loop above.
# Copy it so the helper can add instructions without changing the parent.
helper_history = deepcopy(history)
helper_history.append({
    "role": "developer",
    "content": "Summarize this conversation in one sentence. Do not call tools.",
})
summary = ask_with_implicit_cache(helper_history)
d = summary.usage.input_tokens_details
print("history read", d.cached_tokens, "write", d.cache_write_tokens)
print(summary.output_text)

# Example output (assumed values and conversation):
# history read 4096 write 32
# Case C-123 is due on October 31, 2026.

import hashlib

def cache_key(app, model, prompt_version, instructions):
    digest = hashlib.sha256(instructions.encode()).hexdigest()[:12]
    return f"{app}:{prompt_version}:{model}:{digest}"

KEY = cache_key("support-agent", "us.openai.gpt-6-astra", "v7", INSTRUCTIONS)
prefix = {"role": "developer", "content": [{
    "type": "input_text", "text": INSTRUCTIONS,
    "prompt_cache_breakpoint": {"mode": "explicit"},
}]}
for label, question in [
    ("session A", "Explain the refund policy without calling tools."),
    ("session B", "Explain the escalation policy without calling tools."),
]:
    # A fresh conversation for each session; the startup prefix matches.
    session_history = [prefix, {"role": "user", "content": question}]
    result = ask_with_implicit_cache(session_history)
    d = result.usage.input_tokens_details
    print(label, "read", d.cached_tokens, "write", d.cache_write_tokens)