第 4 章：上下文工程
出处
来源：context_engineering_tools.ipynb · 状态：已读
一句话
上下文是预算，不是垃圾桶。 cache 决定重复前缀的单价，compaction / memory / tool-clearing 决定长任务还能不能活过第 N 轮，extended thinking 是另一笔要记账的注意力。
核心概念
Prompt cache：自动先于精细
misc/prompt_caching.ipynb 两种开法：
| 自动 | 显式 breakpoint | |
|---|---|---|
| 用法 | 请求顶层一个 cache_control | 打在具体 content block 上 |
| 多轮 | breakpoint 自己往前挪 | 你自己挪 |
| 控制 | 一个 TTL | 最多 4 个独立点，可混 TTL |
建议从自动开始。细节（当时文档）：Sonnet 最少 1024 token 才缓存；Opus / Haiku 4.5 要 4096。默认 TTL 5 分钟，命中刷新；1 小时 TTL 写价 2x。写 1.25x，读 0.1x。 自动占用 4 个名额里的一个。
misc/speculative_prompt_caching.ipynb：用户还在打字时发 1-token 预热，把 cache write 藏进等待。大上下文（>1k）反复查询最划算。预热和正式请求必须字节级同前缀；加 timestamp 防串会话。
Batch：不在线，换半价
misc/batch_processing.ipynb：Message Batches 异步，−50% 费用。适合离线评测、批量分类、隔夜摘要。同一 batch 可混 system、多轮、带图。交互产品不走这条。
Compaction：线性历史会先杀死客服 agent
tool_use/automatic-context-compaction.ipynb：5 张工单 × 7 步 ≈ 35 次工具，第 27 轮累计约 15 万 input，只为 5 张票。工单 A 的分类和草稿在处理 B 时仍占窗口。SDK compaction_control：超阈值注入摘要 user turn，模型打 <summary>，清历史，只留摘要继续。Opus 4.6 起优先 服务端 compaction；SDK 方案留给旧模型或想用更便宜模型做摘要的时候。
misc/session_memory_compaction.ipynb 针对对话产品：传统 compaction 撞硬顶再摘要，示例里用户干等 40+ 秒。Instant compaction 在软阈值后台线程写 session-memory.md，硬顶时直接换入。后台摘要把整段对话标 cache，增量指令才按全价，更新可到 ~10x 便宜 / ~80% 降本。和 SDK 自动压历史是两条线：一个面向 agent 工具环，一个面向聊天连续性。
Memory 是工具，不是更长的 system
tool_use/memory_cookbook.ipynb 对齐 Anthropic《Effective context engineering》：跨会话要持久，长会话要腾窗口。memory tool + context editing：该记住的写成文件，该忘的清掉。context_engineering_tools.ipynb 把 memory / compact / tool-clearing 放在一张桌上比——组合，不是三选一。Book 1 05-上下文治理 是同一制度在 Claude Code 里的实现。
Extended thinking 也占窗口
extended_thinking/extended_thinking.ipynb、extended_thinking_with_tool_use.ipynb：思考块要回传才能续想，思考本身是上下文开销。和工具叠用时，adaptive thinking（结果意外才想）比每步都想便宜。第 10 章复现官方深搜分数时，thinking: adaptive、effort、budget、compaction 是一组旋钮，不是四个装饰。
对照 recipe（默认折叠；机制片段摘自官方 notebook，全文在 GitHub）：
misc/prompt_caching.ipynb — 前缀可复用才值得 cache
python
class ConversationWithExplicitCaching:
"""Multi-turn conversation that manually places cache_control on the last user message."""
def __init__(self):
self.turns = []
def add_user(self, content):
self.turns.append({"role": "user", "content": [{"type": "text", "text": content}]})
def add_assistant(self, content):
self.turns.append({"role": "assistant", "content": [{"type": "text", "text": content}]})
def get_messages(self):
"""Return messages with cache_control on the last user message."""
result = []
last_user_idx = max(i for i, t in enumerate(self.turns) if t["role"] == "user")
for i, turn in enumerate(self.turns):
if i == last_user_idx:
result.append(
{
"role": "user",
"content": [
{
"type": "text",
"text": turn["content"][0]["text"],
"cache_control": {"type": "ephemeral"},
}
],
}
)
else:
result.append(turn)
return result
conv = ConversationWithExplicitCaching()
for i, question in enumerate(questions, 1):
print(f"\n{'=' * 50}")
print(f"Turn {i}: {question}")
print("=" * 50)
conv.add_user(question)
start = time.time()
response = client.messages.create(
model=MODEL_NAME,
max_tokens=300,
system=[
{
"type": "text",
"text": system_message,
"cache_control": {"type": "ephemeral"}, # explicit breakpoint on system
},
],
messages=conv.get_messages(),
)
elapsed = time.time() - start
assistant_reply = response.content[0].text
conv.add_assistant(assistant_reply)
print(f"\nAssistant: {assistant_reply[:200]}{'...' if len(assistant_reply) > 200 else ''}")
print()
print_usage(response, elapsed)misc/speculative_prompt_caching.ipynb — 用户打字时预热缓存
python
async def standard_prompt_caching_demo():
client = AsyncAnthropic()
# Prepare the large context
initial_message = await create_initial_message()
# Simulate user typing time (in real app, this would be actual user input)
print("User is typing their question...")
await asyncio.sleep(3) # Simulate 3 seconds of typing
user_question = "What is the purpose of the BtShared structure?"
print(f"User submitted: {user_question}")
# Now send the full request (context + question)
full_message = copy.deepcopy(initial_message)
full_message["content"].append(
{"type": "text", "text": f"Answer the user's question: {user_question}"}
)
print("\nSending request to API...")
start_time = time.time()
# Measure time to first token
first_token_time = None
async with client.messages.stream(
messages=[full_message],
model=MODEL,
**DEFAULT_CLIENT_ARGS,
) as stream:
async for text in stream.text_stream:
if first_token_time is None and text.strip():
first_token_time = time.time() - start_time
print(f"\n🕐 Time to first token: {first_token_time:.2f} seconds")
break
# Get the full response
response = await stream.get_final_message()
total_time = time.time() - start_time
print(f"Total response time: {total_time:.2f} seconds")
print_query_statistics(response, "Standard Caching")
return first_token_time, total_timemisc/batch_processing.ipynb — Message Batches：不在线、换吞吐
python
import base64
def create_complex_batch():
# Get base64 encoded image
def get_base64_encoded_image(image_path):
with open(image_path, "rb") as image_file:
binary_data = image_file.read()
base_64_encoded_data = base64.b64encode(binary_data)
base64_string = base_64_encoded_data.decode("utf-8")
return base64_string
# Mix of different request types
batch_requests = [
{
"custom_id": "simple-question",
"params": {
"model": MODEL_NAME,
"max_tokens": 1024,
"messages": [{"role": "user", "content": "What is quantum computing?"}],
},
},
{
"custom_id": "image-analysis",
"params": {
"model": MODEL_NAME,
"max_tokens": 1024,
"messages": [
{
"role": "user",
"content": [
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": get_base64_encoded_image(
"../images/sunset-dawn-nature-mountain-preview.jpg"
),
},
},
{
"type": "text",
"text": "Describe this mountain landscape. What time of day does it appear to be, and what weather conditions do you observe?",
},
],
}
],
},
},
{
"custom_id": "system-prompt",
"params": {
"model": MODEL_NAME,
"max_tokens": 1024,
"system": "You are a helpful science teacher.",
"messages": [{"role": "user", "content": "Explain gravity to a 5-year-old."}],
},
},
{
"custom_id": "multi-turn",
"params": {
"model": MODEL_NAME,
"max_tokens": 1024,
"messages": [
{"role": "user", "content": "What is DNA?"},
{
"role": "assistant",
"content": "DNA is like a blueprint for living things...",
},
{"role": "user", "content": "How is DNA copied?"},
],
},
},
]
try:
response = client.beta.messages.batches.create(requests=batch_requests)
return response.id
except Exception as e:
print(f"Error creating batch: {e}")
return None
complex_batch_id = create_complex_batch()
print(f"Complex batch ID: {complex_batch_id}")tool_use/automatic-context-compaction.ipynb — 长代理自动压历史
python
# Re-initialize queue and run with compaction
initialize_ticket_queue(num_tickets)
total_input_compact = 0
total_output_compact = 0
turn_count_compact = 0
compaction_count = 0
prev_msg_count = 0
runner = client.beta.messages.tool_runner(
model=MODEL,
max_tokens=4096,
tools=tools,
messages=messages,
compaction_control={
"enabled": True,
"context_token_threshold": 5000,
},
)
for message in runner:
turn_count_compact += 1
total_input_compact += message.usage.input_tokens
total_output_compact += message.usage.output_tokens
messages_list = list(runner._params["messages"])
curr_msg_count = len(messages_list)
if curr_msg_count < prev_msg_count:
# We can identify compaction when the message count decreases
compaction_count += 1
print(f"\n{'=' * 60}")
print(f"🔄 Compaction occurred! Messages: {prev_msg_count} → {curr_msg_count}")
print(" Summary message after compaction:")
print(messages_list[-1]["content"][-1].text) # type: ignore
print(f"\n{'=' * 60}")
prev_msg_count = curr_msg_count
print(
f"Turn {turn_count_compact:2d}: Input={message.usage.input_tokens:7,} tokens | "
f"Output={message.usage.output_tokens:5,} tokens | "
f"Messages={len(messages_list):2d} | "
f"Cumulative In={total_input_compact:8,}"
)
print(f"\n{'=' * 60}")
print("OPTIMIZED RESULTS (WITH COMPACTION)")
print(f"{'=' * 60}")
print(f"Total turns: {turn_count_compact}")
print(f"Compactions: {compaction_count}")
print(f"Input tokens: {total_input_compact:,}")
print(f"Output tokens: {total_output_compact:,}")
print(f"Total tokens: {total_input_compact + total_output_compact:,}")
print(f"{'=' * 60}")misc/session_memory_compaction.ipynb — 对话产品：后台预写记忆，硬顶瞬间换入
python
import time
class TraditionalCompactingChatSession:
"""Traditional chat session with compaction after the fact."""
def __init__(self, system_message="You are a helpful assistant", context_limit: int = 10000):
self.system_message = system_message
self.context_limit = context_limit # the point at which the conversation is compacted so it does not exceed model limits.
self.messages = []
self.current_context_window_tokens = 0
self.summary = None
def chat(self, user_message: str) -> tuple[str, anthropic.types.Usage]:
# In traditional compaction, we check if we need to compact when the user sends a message. NOT IDEAL!
if self.current_context_window_tokens >= self.context_limit:
print(
f"\n🧹 Context window at {self.current_context_window_tokens} tokens. Limit exceeded, compacting session memory..."
)
self.compact() # compacts everything before the new user message
self.messages.append({"role": "user", "content": user_message})
print(f"\nUser: {user_message}")
response = client.messages.create(
model=MODEL,
max_tokens=3500,
system=self.system_message,
messages=add_cache_control(self.messages),
)
assistant_message = response.content[0].text
self.messages.append({"role": "assistant", "content": assistant_message})
print(f"\nAssistant: \n{truncate_response(assistant_message, max_lines=15)}")
# approximate current token count in the conversation before the next user message
cache_read = getattr(response.usage, "cache_read_input_tokens", 0) or 0
total_input = response.usage.input_tokens + cache_read
self.current_context_window_tokens = total_input + response.usage.output_tokens
print(
f"Input={total_input:,}, Prompt cached used= {cache_read > 0} | "
f"Output={response.usage.output_tokens:,} | "
f"Messages={len(self.messages)}"
)
return assistant_message, response.usage
def compact(self) -> None:
start_time = time.perf_counter()
response = client.messages.create(
model=MODEL,
max_tokens=5000,
system=self.system_message, # Same as main chat for cache sharing
messages=add_cache_control(self.messages)
+ [{"role": "user", "content": SESSION_MEMORY_PROMPT}],
)
elapsed = time.perf_counter() - start_time
# Generate new summary message
self.summary, removed_text = remove_thinking_blocks(
response.content[0].text
) # clean up any <think> blocks because they are not needed in the session memory
approximate_summary_tokens = response.usage.output_tokens - round(
len(removed_text) / 4
) # rough estimate of tokens removed from summary
# Replace prior messages with new summary message
self.messages = [
{
"role": "user",
"content": f"""This session is being continued from a previous conversation. Here is the session memory: {self.summary}.Continue from where we left off.""",
}
]
# Show token reduction if we just compacted
reduction = self.current_context_window_tokens - approximate_summary_tokens
pct = (reduction / self.current_context_window_tokens) * 100
print(f"\n{'-' * 60}")
print("📝 New session memory created.")
print(
f"✅ Tokens reduced: {self.current_context_window_tokens:,} → {approximate_summary_tokens:.0f} ({reduction:,} tokens saved, {pct:.0f}% reduction)"
)
print(f"⏱️ Compaction time: {elapsed:.2f}s (user waiting...)")
print(f" Cache used: {getattr(response.usage, 'cache_read_input_tokens', 0) > 0}")
print(f"{'-' * 60}")
# Update token count to reflect compacted state
self.current_context_window_tokens = approximate_summary_tokenstool_use/memory_cookbook.ipynb — memory 作为工具，而不是「全塞 system」
python
import os
from typing import cast
from anthropic import Anthropic
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Model configuration - use alias for automatic updates
MODEL = "claude-sonnet-4-6" # Can override via ANTHROPIC_MODEL env var
if os.getenv("ANTHROPIC_MODEL"):
MODEL = os.getenv("ANTHROPIC_MODEL")
if not API_KEY:
raise ValueError("ANTHROPIC_API_KEY not found. Copy .env.example to .env and add your API key.")
if not MODEL:
raise ValueError("ANTHROPIC_MODEL not found. Copy .env.example to .env and set the model.")
MODEL = cast(str, MODEL)
client = Anthropic(api_key=API_KEY)
print("✓ API key loaded")
print(f"✓ Using model: {MODEL}")tool_use/context_engineering/context_engineering_tools.ipynb — memory / compact / tool clearing 对照与组合
python
# Force-reload the corpus module in case this kernel has a stale cached version
import importlib
import research_corpus
importlib.reload(research_corpus)
from research_corpus import COMPACTION_PROBES, CORPUS
# Count tokens via the API (cached in-memory so repeat kernel runs are fast).
_token_cache: dict[str, int] = {}
def count_tokens(text: str) -> int:
if text not in _token_cache:
_token_cache[text] = client.messages.count_tokens(
model=MODEL, messages=[{"role": "user", "content": text}]
).input_tokens
return _token_cache[text]
print(f"CORPUS is a dict of {len(CORPUS)} synthetic documents held in Python memory.")
print("When the agent calls read_file, the content is served from this dict and")
print("lands directly in the agent's context window — no disk I/O involved.\n")
_total_tokens = 0
for path, content in CORPUS.items():
n_tok = count_tokens(content)
_total_tokens += n_tok
# Strip the virtual directory prefix for display clarity
display_name = path.removeprefix("/research/")
print(f" {display_name:<26} ~{n_tok:>6,} tokens")
print(f"\n Total corpus: ~{_total_tokens:,} tokens")
# Sanity check: the corpus should be large enough to push context well past
# 200K in a single read-pass. If this fires, research_corpus.py may be stale.
assert _total_tokens > 250_000, (
f"Corpus is only {_total_tokens:,} tokens; expected >250K. "
"Restart the kernel and re-run, or verify research_corpus.py is current."
)
# Side-by-side view of each primitive's solo run
solo_results = {
"baseline": baseline,
"clearing": clearing_run,
"compaction": compaction_run,
"memory (S2 with)": memory_s2,
}
plot_summary_bars(solo_results, "Each primitive alone: session outcomes")
# Completion status and events per run
print(f"{'Run':<20} {'Status':<18} {'Context events'}")
print("-" * 60)
for name, res in solo_results.items():
status = "HIT LIMIT" if res.hit_limit else "completed"
n_clear = sum(1 for e in res.events if e["kind"] == "clearing")
n_compact = sum(1 for e in res.events if e["kind"] == "compaction")
ev_parts = []
if n_clear:
ev_parts.append(f"{n_clear} clearing")
if n_compact:
ev_parts.append(f"{n_compact} compaction")
print(f" {name:<18} {status:<18} {', '.join(ev_parts) if ev_parts else 'none'}")extended_thinking/extended_thinking.ipynb — 思考块是另一笔上下文开销
python
def token_counting_example():
# Define a function to create a sample prompt
def create_sample_messages():
messages = [{
"role": "user",
"content": "Solve this puzzle: Three people check into a hotel. They pay $30 to the manager. The manager finds out that the room only costs $25 so he gives $5 to the bellboy to return to the three people. The bellboy, however, decides to keep $2 and gives $1 back to each person. Now, each person paid $10 and got back $1, so they paid $9 each, totaling $27. The bellboy kept $2, which makes $29. Where is the missing $1?"
}]
return messages
# Count tokens without thinking
base_messages = create_sample_messages()
base_token_count = count_tokens(base_messages)
print(f"Base token count (input only): {base_token_count}")
# Make a request with thinking and check actual usage
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=8000,
thinking = {
"type": "enabled",
"budget_tokens": 2000
},
messages=base_messages
)
# Calculate and print token usage stats
thinking_tokens = sum(
len(block.thinking.split()) * 1.3 # Rough estimate
for block in response.content
if block.type == "thinking"
)
final_answer_tokens = sum(
len(block.text.split()) * 1.3 # Rough estimate
for block in response.content
if block.type == "text"
)
print(f"\nEstimated thinking tokens used: ~{int(thinking_tokens)}")
print(f"Estimated final answer tokens: ~{int(final_answer_tokens)}")
print(f"Total estimated output tokens: ~{int(thinking_tokens + final_answer_tokens)}")
print(f"Input tokens + max_tokens = {base_token_count + 8000}")
print(f"Available for final answer after thinking: ~{8000 - int(thinking_tokens)}")
# Demo with escalating thinking budgets
thinking_budgets = [1024, 2000, 4000, 8000, 16000, 32000]
context_window = 200000
for budget in thinking_budgets:
print(f"\nWith thinking budget of {budget} tokens:")
print(f"Input tokens: {base_token_count}")
print(f"Max tokens needed: {base_token_count + budget + 1000}") # Add 1000 for final answer
print(f"Remaining context window: {context_window - (base_token_count + budget + 1000)}")
if base_token_count + budget + 1000 > context_window:
print("WARNING: This would exceed the context window of 200k tokens!")
# Uncomment to run the example
token_counting_example()extended_thinking/extended_thinking_with_tool_use.ipynb — 边想边调工具
python
def tool_use_with_thinking():
# Define a weather tool
tools = [
{
"name": "weather",
"description": "Get current weather information for a location.",
"input_schema": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The location to get weather for."
}
},
"required": ["location"]
}
}
]
def weather(location):
# Mock weather data
weather_data = {
"New York": {"temperature": 72, "condition": "Sunny"},
"London": {"temperature": 62, "condition": "Cloudy"},
"Tokyo": {"temperature": 80, "condition": "Partly cloudy"},
"Paris": {"temperature": 65, "condition": "Rainy"},
"Sydney": {"temperature": 85, "condition": "Clear"},
"Berlin": {"temperature": 60, "condition": "Foggy"},
}
return weather_data.get(location, {"error": f"No weather data available for {location}"})
# Initial request with tool use and thinking
response = client.messages.create(
model=MODEL_NAME,
max_tokens=MAX_TOKENS,
thinking={
"type": "enabled",
"budget_tokens": THINKING_BUDGET_TOKENS
},
tools=tools,
messages=[{
"role": "user",
"content": "What's the weather like in Paris today?"
}]
)
# Detailed diagnostic output of initial response
print("\n=== INITIAL RESPONSE ===")
print(f"Response ID: {response.id}")
print(f"Stop reason: {response.stop_reason}")
print(f"Model: {response.model}")
print(f"Content blocks: {len(response.content)} blocks")
for i, block in enumerate(response.content):
print(f"\nBlock {i+1}: Type = {block.type}")
if block.type == "thinking":
print(f"Thinking content: {block.thinking[:150]}...")
print(f"Signature available: {bool(getattr(block, 'signature', None))}")
elif block.type == "text":
print(f"Text content: {block.text}")
elif block.type == "tool_use":
print(f"Tool: {block.name}")
print(f"Tool input: {block.input}")
print(f"Tool ID: {block.id}")
print("=== END INITIAL RESPONSE ===\n")
# Extract thinking blocks to include in the conversation history
assistant_blocks = []
for block in response.content:
if block.type in ["thinking", "redacted_thinking", "tool_use"]:
assistant_blocks.append(block)
# Handle tool use if required
full_conversation = [{
"role": "user",
"content": "What's the weather like in Paris today?"
}]
if response.stop_reason == "tool_use":
# Add entire assistant response with thinking blocks and tool use
full_conversation.append({
"role": "assistant",
"content": assistant_blocks
})
# Find the tool_use block
tool_use_block = next((block for block in response.content if block.type == "tool_use"), None)
if tool_use_block:
# Execute the tool
print(f"\n=== EXECUTING TOOL ===")
print(f"Tool name: {tool_use_block.name}")
print(f"Location to check: {tool_use_block.input['location']}")
tool_result = weather(tool_use_block.input["location"])
print(f"Result: {tool_result}")
print("=== TOOL EXECUTION COMPLETE ===\n")
# Add tool result to conversation
full_conversation.append({
"role": "user",
"content": [{
"type": "tool_result",
"tool_use_id": tool_use_block.id,
"content": json.dumps(tool_result)
}]
})
# Continue the conversation with the same thinking configuration
print("\n=== SENDING FOLLOW-UP REQUEST WITH TOOL RESULT ===")
response = client.messages.create(
model=MODEL_NAME,
max_tokens=MAX_TOKENS,
thinking={
"type": "enabled",
"budget_tokens": THINKING_BUDGET_TOKENS
},
tools=tools,
messages=full_conversation
)
print(f"Follow-up response received. Stop reason: {response.stop_reason}")
print_thinking_response(response)
# Run the example
tool_use_with_thinking()关键洞察
上下文工程的决策树其实很短：
- 前缀稳、多轮重复 → cache（必要时投机预热）
- 不需要人等 → batch
- 工具结果线性堆积 → compact 或 PTC（第 3 章）或清工具
- 跨会话的偏好 / 事实 → memory 文件，不要指望窗口
- 思考深度 → 单独记账，不要和「更长 prompt」混在一起买
「把什么都贴进 system」是最贵的默认。Book 1 用 section 边界保护缓存；这里用 API 原语做同一件事。
个人思考
第 2 章 contextual retrieval 已经依赖 cache 才跑得起；第 3 章 PTC 从源头少灌结果。本章是收口：还是灌进来了，怎么活。下一章工作流会把单次调用乘成 N+1——没有预算制度，orchestrator 只是加速破产。