Claude Agent SDK教程:自动压缩长任务上下文
技术背景
在构建基于工具调用的AI代理时,随着任务迭代,对话历史会迅速膨胀。每次工具调用都会将完整历史发送给模型,导致令牌消耗线性增长,成本上升,响应变慢,并可能触及上下文窗口上限。例如,处理5个工单、每个工单7步操作,会产生35次工具调用,累积大量中间结果。自动上下文压缩通过定期生成摘要并清除历史,使上下文保持有界,模拟人类工作方式:完成一项任务后简要记录,再处理下一项。该技术适用于顺序处理、多阶段工作流、批量操作等场景,但需注意信息丢失和服务端采样循环的限制。
核心原理
compaction_control参数监控每轮令牌使用量,超过阈值时自动注入摘要提示作为用户回合,模型生成包含在<summary></summary>标签中的摘要,随后清除对话历史,仅保留摘要继续任务。核心原理是:1) 令牌监控:SDK在每次API调用后统计输入输出令牌;2) 阈值触发:当累计令牌超过context_token_threshold(默认100k)时触发压缩;3) 摘要生成:使用主模型或指定模型,根据summary_prompt生成结构化摘要,保留关键信息如已完成工单、进度状态、后续步骤;4) 历史替换:将整个消息列表替换为仅含摘要的用户消息,实现上下文重置。数据流为:用户消息→工具调用→结果累积→触发压缩→摘要→继续任务。关键设计决策包括阈值选择(低阈值适合独立任务,高阈值保留更多细节)、摘要模型选择(可用更便宜模型降低成本)、自定义提示词以保留特定信息。
架构分析
系统架构基于Claude Agent SDK的tool_runner,它封装了消息循环和工具调用。核心组件包括:1) 工具定义:使用@beta_tool装饰器将Python函数暴露为Claude工具,自动提取参数和文档字符串作为元数据;2) 消息历史:BetaMessageParam列表存储对话,SDK在每次迭代后更新;3) 压缩控制器:compaction_control参数配置启用、阈值、摘要模型和提示词;4) 工具运行器:迭代执行工具调用,并在检测到超阈值时自动执行压缩流程。设计思想是分离任务执行与上下文管理,使开发者专注于业务逻辑,SDK处理上下文优化。对于无工具场景,需手动实现压缩循环,通过跟踪令牌、调用摘要API、替换历史列表。
完整实战教程
什么是上下文压缩?
在构建使用工具调用的代理工作流时,随着代理在复杂任务上迭代,对话可能会变得非常庞大。compaction_control 参数通过以下方式提供自动上下文管理:
- 监控对话中每轮的令牌使用量
- 当超过阈值时,注入一个摘要提示作为用户回合
- 让模型生成包含在
<summary></summary>标签中的摘要。这些标签不会被解析,但用于引导模型。 - 清除对话历史,仅保留摘要继续
- 使用压缩后的上下文继续任务
通过本教程,你将能够:
- 理解如何在迭代工作流中有效管理上下文限制
- 编写利用自动上下文压缩的代理
- 设计在多次迭代中保持专注的工作流
先决条件
在遵循本指南之前,请确保你具备:
所需知识
- 对代理模式和工具调用的基本理解
所需工具
- Python 3.11 或更高版本
- Anthropic API 密钥
- Anthropic SDK >= 0.74.1
正在使用 Opus 4.6? 我们建议使用服务端压缩,它可以自动处理上下文管理,无需任何 SDK 级别的配置。
本手册涵盖基于 SDK 的压缩,如果您使用的是较旧的模型,或者希望使用不同的(更便宜的)模型进行摘要,这将非常有用。
环境搭建
首先,安装所需的依赖:
# %pip install -qU anthropic python-dotenv
注意:确保你的 .env 文件包含:
ANTHROPIC_API_KEY=your_key_here
加载你的环境变量并配置客户端。我们还加载了一个辅助工具来可视化 Claude 消息响应。
from dotenv import load_dotenv
load_dotenv()
MODEL = "claude-sonnet-4-6"
准备工作
在 utils/customer_service_tools.py 中,我们定义了几个用于处理客户支持工单的函数:
get_next_ticket()- 从队列中检索下一个未处理的工单classify_ticket(ticket_id, category)- 将问题分类为账单、技术、账户、产品或运输search_knowledge_base(query)- 查找相关的帮助文章和解决方案set_priority(ticket_id, priority)- 分配优先级(低、中、高、紧急)route_to_ticket(ticket_id, team)- 将工单路由到相应的支持团队draft_response(ticket_id, response_text)- 创建面向客户的回复mark_complete(ticket_id)- 完成已处理的工单
对于客户服务代理,这些工具支持系统地处理工单。每个工单都需要分类、研究、优先级排序、路由和回复草拟。当按顺序处理20-30个工单时,对话历史会填满每个分类、每个知识库搜索和每个草拟回复的工具结果,导致令牌线性增长。
工具上使用了 beta_tool 装饰器,使它们对 Claude 代理可用。该装饰器提取函数参数和文档字符串,并将其作为工具元数据提供给 Claude。
import anthropic
from anthropic import beta_tool
@beta_tool
def get_next_ticket() -> dict:
"""Retrieve the next unprocessed support ticket from the queue."""
...
import anthropic
from utils.customer_service_tools import (
classify_ticket,
draft_response,
get_next_ticket,
initialize_ticket_queue,
mark_complete,
route_to_team,
search_knowledge_base,
set_priority,
)
client = anthropic.Anthropic()
tools = [
get_next_ticket,
classify_ticket,
search_knowledge_base,
set_priority,
route_to_team,
draft_response,
mark_complete,
]
基线:不进行压缩的运行
让我们从一个现实的客户服务场景开始:处理一个支持工单队列。
工作流程如下:
对于每个工单:
- 使用
get_next_ticket()获取工单 - 分类问题类别(账单、技术、账户、产品、物流)
- 搜索知识库以获取相关信息
- 设置适当的优先级(低、中、高、紧急)
- 路由到正确的团队
- 起草客户回复
- 将工单标记为完成
- 移动到下一个工单
挑战:队列中有5个工单,每个需要7次工具调用,Claude将进行35次或更多的工具调用。每一步的结果,包括分类、知识库搜索和起草的回复,都会累积在对话历史中。如果不进行压缩,所有这些数据都会在每个工单的处理过程中保留在内存中,到第5个工单时,上下文将包含前4个工单的所有完整细节。
让我们先不进行压缩地运行这个工作流程,观察会发生什么:
from anthropic.types.beta import BetaMessageParam
num_tickets = 5
initialize_ticket_queue(num_tickets)
messages: list[BetaMessageParam] = [
{
"role": "user",
"content": f"""You are an AI customer service agent. Your task is to process support tickets from a queue.
For EACH ticket, you must complete ALL these steps:
1. **Fetch ticket**: Call get_next_ticket() to retrieve the next unprocessed ticket
2. **Classify**: Call classify_ticket() to categorize the issue (billing/technical/account/product/shipping)
3. **Research**: Call search_knowledge_base() to find relevant information for this ticket type
4. **Prioritize**: Call set_priority() to assign priority (low/medium/high/urgent) based on severity
5. **Route**: Call route_to_team() to assign to the appropriate team
6. **Draft**: Call draft_response() to create a helpful customer response using KB information
7. **Complete**: Call mark_complete() to finalize this ticket
8. **Continue**: Immediately fetch the next ticket and repeat
IMPORTANT RULES:
- Process tickets ONE AT A TIME in sequence
- Complete ALL 7 steps for each ticket before moving to the next
- Keep fetching and processing tickets until you get an error that the queue is empty
- There are {num_tickets} tickets total - process all of them
- Be thorough but efficient
Begin by fetching the first ticket.""",
}
]
total_input = 0
total_output = 0
turn_count = 0
runner = client.beta.messages.tool_runner(
model=MODEL,
max_tokens=4096,
tools=tools,
messages=messages,
)
for message in runner:
messages_list = list(runner._params["messages"])
turn_count += 1
total_input += message.usage.input_tokens
total_output += message.usage.output_tokens
print(
f"Turn {turn_count: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:8,}"
)
print(f"\n{'=' * 60}")
print("BASELINE RESULTS (NO COMPACTION)")
print(f"{'=' * 60}")
print(f"Total turns: {turn_count}")
print(f"Input tokens: {total_input:,}")
print(f"Output tokens: {total_output:,}")
print(f"Total tokens: {total_input + total_output:,}")
print(f"{'=' * 60}")
现在我们有了基线,可以更清楚地了解在没有压缩的情况下上下文是如何增长的。如你所见,每一轮都会导致 token 数量线性增长,因为每一轮都会向输入中添加更多的 token。
这会导致高 token 消耗,并可能很快达到上下文限制。到第 27 轮时,仅处理 5 个工单就累计消耗了 150,000 个输入 token。
让我们回顾一下 Claude 在没有压缩的情况下处理完所有 5 个工单后的最终响应:
print(message.content[-1].text)
理解问题
在上述基线工作流中,Claude 必须:
- 顺序处理 5 个支持工单
- 每个工单完成 7 个步骤(获取、分类、研究、优先级排序、路由、草拟、完成)
- 进行 35 次工具调用,结果累积在对话历史中
- 在内存中存储 每个分类、每次知识库搜索、每个草拟的回复
为什么会这样:
- 线性 token 增长 - 每次使用工具时,整个对话历史(包括之前所有工具结果)都会发送给 Claude
- 上下文污染 - 在处理工单 B 时,工单 A 的分类和草拟回复仍保留在上下文中
- 成本叠加 - 当你处理到第 5 个工单时,每次 API 调用都会发送前 4 个工单的数据
- 响应变慢 - 处理大量上下文需要更长时间
- 达到限制的风险 - 最终会达到 200k token 的上下文窗口
我们真正需要的:完成工单 A 后,我们只需要一个 简要摘要(工单已解决、类别、优先级)——而不是完整的分类结果、知识库搜索和完整的草拟回复。详细的工作流应该被丢弃,只保留完成摘要。
让我们看看自动上下文压缩如何解决这个问题。
启用自动上下文压缩
让我们运行完全相同的客户服务工作流,但启用自动上下文压缩。我们只需在工具运行器中添加 compaction_control 参数。
compaction_control 参数有一个必填字段和几个可选字段:
enabled(必填):布尔值,用于打开/关闭压缩context_token_threshold(可选):触发压缩的令牌数(默认:100,000)model(可选):用于摘要的模型(默认为主模型)summary_prompt(可选):用于生成摘要的自定义提示
对于此客户服务工作流,我们将使用 5,000 令牌阈值。这意味着在处理多个工单后,压缩将自动触发。这允许 Claude:
- 保留完成摘要(已解决的工单、类别、结果)
- 丢弃详细的工具结果(完整的知识库文章、完整的分类、草拟的回复文本)
- 在处理下一批工单时重新开始
这模拟了真实支持代理的工作方式:解决工单,简要记录,然后处理下一个案例。
# 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}")
启用自动上下文压缩后,我们可以看到每轮对话的令牌使用量不会线性增长,而是在每次压缩事件后减少。在处理工单期间发生了两次压缩事件,后续轮次显示总令牌使用量有所下降。
与基线版本相比,我们仅使用了 79,000 个令牌。我们还打印了每次压缩事件后生成的摘要消息,展示了 Claude 如何有效地将之前的工单细节浓缩为摘要。
现在,让我们看看启用压缩后处理完所有 5 个工单后的最终响应。
print(message.content[-1].text)
对比结果
启用压缩后,我们可以清楚地看到两次运行在令牌节省方面的明显差异,同时保持了工作流和最终摘要的质量。
以下是启用自动上下文压缩后的变化:
- 处理若干工单后上下文重置 - 当处理5-7个工单产生超过5k令牌的工具结果时,SDK会自动:
- 注入摘要提示
- 让Claude生成包含在
<summary></summary>标签中的完成摘要 - 清除对话历史,丢弃详细的分类、知识库搜索和回复
- 仅保留完成摘要继续处理
- 输入令牌保持有界 - 随着处理更多工单,输入令牌不会累积到100k以上,而是在每次压缩后重置。处理第5个工单时,我们不会携带第1-4个工单的完整工具结果。
- 任务成功完成 - 工作流顺利处理所有工单,不会触及上下文限制
- 质量得到保留 - 摘要保留了关键信息:
- 已处理的工单及其ID
- 分配的分类和优先级
- 路由到的团队
- 整体进度状态
所有工单仍然被正确分类、确定优先级、路由和回复。
- 自然的工作流 - 这反映了真实支持代理的工作方式:解决一个工单,在系统中简要记录,关闭它,然后处理下一个。你不会在处理新工单时保留每篇知识库文章和完整的回复草稿。
让我们可视化令牌节省情况:
# Compare baseline vs compaction
print("=" * 70)
print("TOKEN USAGE COMPARISON")
print("=" * 70)
print(f"{'Metric':<30} {'Baseline':<20} {'With Compaction':<20}")
print("-" * 70)
print(f"{'Input tokens:':<30} {total_input:>19,} {total_input_compact:>19,}")
print(f"{'Output tokens:':<30} {total_output:>19,} {total_output_compact:>19,}")
print(
f"{'Total tokens:':<30} {total_input + total_output:>19,} {total_input_compact + total_output_compact:>19,}"
)
print(f"{'Compactions:':<30} {'N/A':>19} {compaction_count:>19}")
print("=" * 70)
# Calculate savings
token_savings = (total_input + total_output) - (total_input_compact + total_output_compact)
savings_percent = (
(token_savings / (total_input + total_output)) * 100 if (total_input + total_output) > 0 else 0
)
print(f"\n💰 Token Savings: {token_savings:,} tokens ({savings_percent:.1f}% reduction)")
压缩在底层是如何工作的
当 tool_runner 检测到 token 使用量超过阈值时,它会自动:
- 暂停工作流,在下次 API 调用之前
- 注入一个摘要请求,作为用户消息,要求 Claude 总结进度
- 生成摘要——Claude 生成一个包裹在
<summary></summary>标签中的摘要,包含:
- 已完成的工单:已解决工单的简要记录(ID、类别、优先级、结果)
- 进度状态:已处理的工单数量,剩余数量
- 关键模式:跨工单的任何显著趋势
- 后续步骤:接下来要做什么(继续处理剩余工单)
- 清除历史记录——整个对话历史(包括所有工具结果)被替换为仅包含摘要
- 恢复处理——Claude 使用压缩后的上下文继续工作,处理下一批工单
自定义压缩配置
您可以根据具体用例自定义压缩的工作方式。以下是关键配置选项:
调整阈值
context_token_threshold 决定何时触发压缩:
compaction_control={
"enabled": True,
"context_token_threshold": 5000, # Compact after processing 5-7 tickets
}
阈值不应设置得太低,否则摘要本身可能触发压缩。我们为了演示目的设置了5,000个令牌的阈值,但在实践中,请尝试不同的设置,以找到最适合您工作流程的配置。
以下是一些通用指南:
- 低阈值(5k-20k):
- 用于具有明确边界的迭代任务处理
- 更频繁的压缩,最小化上下文积累
- 最适合顺序实体处理
- 中等阈值(50k-100k):
- 多阶段工作流,具有较少但较大的自然检查点
- 在上下文保留和管理之间取得平衡
- 适用于具有昂贵工具调用的工作流
- 高阈值(100k-150k):
- 需要大量历史上下文的任务
- 较少的压缩保留了更多原始细节
- 每次调用成本更高,但压缩次数更少
- 默认(100k):对于一般长时间运行的任务,这是一个很好的平衡。
对于工单处理:5k阈值效果很好,因为每个工单的工作流会生成大量的工具结果,但工单之间是独立的。在解决工单A后,处理工单B时不需要其详细的KB搜索。
使用不同的模型进行摘要
您还可以使用更快/更便宜的模型来生成摘要:
compaction_control={
"enabled": True,
"model": "claude-haiku-4-5", # Use Haiku for cost-effective summaries
}
自定义摘要提示
您可以提供自定义提示来指导摘要的生成方式。这对于需要保留特定类型信息的客户服务工作流尤其有用。
例如,我们可以根据需求定义一个自定义提示:
- 所有已完成工单的摘要
- 分配的类别和优先级
- 路由到的团队
- 进度状态(已完成工单、剩余工单)
- 工作流中的后续步骤
compaction_control={
"enabled": True,
"summary_prompt": """You are processing customer support tickets from a queue.
Create a focused summary that preserves:
1. **COMPLETED TICKETS**: For each ticket you've fully processed:
- Ticket ID and customer name
- Issue category and priority assigned
- Team routed to
- Brief outcome
2. **PROGRESS STATUS**:
- How many tickets you've completed
- Approximately how many remain in the queue
3. **NEXT STEPS**: Continue processing the next ticket
Format with clear sections and wrap in <summary></summary> tags."""
}
无需工具的压缩:简单聊天循环
虽然上述示例侧重于工具密集型的代理工作流,但上下文压缩对于简单的对话应用同样有价值,在这些应用中,用户驱动对话。
注意: 上面演示的 compaction_control 参数适用于带有工具的 tool_runner 代理工作流。对于没有工具的简单聊天应用,你将使用相同的原理手动实现压缩。
考虑一个聊天应用,用户与 Claude 进行长时间对话——讨论复杂主题、迭代想法或解决问题。随着对话的增长,你面临相同的上下文累积挑战。
区别:不是工具使用触发 token 增长,而是来回对话本身。每次交流都会向历史记录添加消息:
- 用户提问
- Claude 提供详细回答
- 用户要求澄清或详细说明
- Claude 用更多上下文回应
- 这重复数十次或数百次
如果没有压缩,到第 50 轮时,你每次 API 调用都会发送整个对话历史(所有 50 次交流)。
解决方案:使用相同的模式在聊天循环中手动实现压缩:
- 每轮后跟踪 token 使用情况
- 当超过阈值时,请求摘要
- 用摘要替换对话历史
- 使用压缩的上下文继续对话
让我们看看如何实现:
#!/usr/bin/env python3
"""
Simple Compaction Example - User-Driven Chat Loop
This shows the basic pattern for a chat application with compaction.
No tools required - just a simple loop where the user drives continuation.
"""
# Configuration
COMPACTION_THRESHOLD = 3000 # Compact when tokens exceed this (low for demo purposes)
# Structured summarization prompt for compaction
SUMMARY_PROMPT = """You have been working on the task described above but have not yet completed it. Write a continuation summary that will allow you (or another instance of yourself) to resume work efficiently in a future context window where the conversation history will be replaced with this summary. Your summary should be structured, concise, and actionable. Include:
1. **Task Overview**
- The user's core request and success criteria
- Any clarifications or constraints they specified
2. **Current State**
- What has been completed so far
- Files created, modified, or analyzed (with paths if relevant)
- Key outputs or artifacts produced
3. **Important Discoveries**
- Technical constraints or requirements uncovered
- Decisions made and their rationale
- Errors encountered and how they were resolved
- What approaches were tried that didn't work (and why)
4. **Next Steps**
- Specific actions needed to complete the task
- Any blockers or open questions to resolve
- Priority order if multiple steps remain
5. **Context to Preserve**
- User preferences or style requirements
- Domain-specific details that aren't obvious
- Any promises made to the user
Be concise but complete—err on the side of including information that would prevent duplicate work or repeated mistakes.
Write in a way that enables immediate resumption of the task.
Wrap your summary in <summary></summary> tags."""
# Message history
messages = []
print("Chat with Claude (type 'quit' to exit, or just hit Enter to continue)")
print("This is a demonstration - try having a conversation and watch compaction trigger")
print("=" * 60)
# Simulate a conversation for demo purposes
demo_messages = [
"Help me understand how Python decorators work",
"Can you show me an example with a timing decorator?",
"How would I make a decorator that takes arguments?",
]
for user_input in demo_messages:
print(f"\nYou: {user_input}")
# Add user message
messages.append({"role": "user", "content": user_input})
# Get Claude's response
response = client.messages.create(
model=MODEL,
max_tokens=2048,
messages=messages,
)
messages.append(
{
"role": "assistant",
"content": response.content,
}
)
print("\nClaude: ", end="")
for block in response.content:
if block.type == "text":
print(f"{block.text[:300]} ...")
# Check if we should compact
usage = response.usage
# Calculate total tokens (includes cache tokens)
total_input_tokens = (
usage.input_tokens
+ (usage.cache_creation_input_tokens or 0)
+ (usage.cache_read_input_tokens or 0)
)
total_tokens = total_input_tokens + usage.output_tokens
cache_info = ""
if usage.cache_creation_input_tokens or usage.cache_read_input_tokens:
cache_info = f" (cache: {usage.cache_creation_input_tokens or 0} write + {usage.cache_read_input_tokens or 0} read)"
print(
f"\n[Tokens: {total_input_tokens} in{cache_info} + {usage.output_tokens} out = {total_tokens} total]"
)
if total_tokens > COMPACTION_THRESHOLD:
print(f"\n{'=' * 60}")
print(f"🔄 Compacting conversation... {len(messages)} messages → ", end="", flush=True)
# Get summary using structured prompt
summary_response = client.messages.create(
model=MODEL,
max_tokens=4096,
messages=messages + [{"role": "user", "content": SUMMARY_PROMPT}],
)
summary_text = "".join(
block.text for block in summary_response.content if block.type == "text"
)
# Replace history with summary
messages = [{"role": "user", "content": summary_text}]
print("1 message")
print(f"{'=' * 60}\n")
print(f"Final conversation messages: {messages[-1].get('content')}")
print("\nDemo complete! In a real application, this loop would continue with user input.")
理解聊天循环模式
上面的示例演示了在对话上下文中手动压缩。其工作原理如下:
关键组件:
- 令牌跟踪:每次响应后,计算总令牌数(输入 + 输出 + 缓存令牌)
- 阈值检查:当总数超过阈值时,触发压缩
- 摘要请求:向 Claude 发送相同的结构化 SUMMARY_PROMPT
- 历史替换:用摘要替换整个消息历史
- 继续:下一条用户消息基于摘要,而不是完整历史
何时使用此模式:
- 扩展头脑风暴会话:用户与 Claude 多轮探索想法
- 学习对话:跨越数十次交流的教程或解释
- 迭代优化:用户对草稿、设计或解决方案提供反馈
- 聊天应用:任何多轮对话界面
与工具运行器的关键区别:
| 方面 | 工具运行器(自动) | 聊天循环(手动) |
|--------|------------------------|-------------------|
| 触发 | 达到阈值时自动 | 你实现阈值检查 |
| 摘要 | SDK 处理摘要请求 | 你显式调用 API |
| 历史管理 | SDK 替换消息 | 你手动替换列表 |
| 用例 | 带工具的代理工作流 | 用户驱动的对话 |
生产注意事项:
- 调整阈值:对于实际应用使用更大的阈值
- 自定义摘要提示:根据你的对话类型(头脑风暴 vs. 技术支持 vs. 辅导)定制
- 显示用户指示:显示类似“正在总结对话...”的消息,让用户理解暂停
- 保留关键上下文:确保摘要提示捕获用户关心的领域特定信息
此模式让你完全控制压缩发生的时间和方式,非常适合 SDK 的自动工具运行器压缩不可用的对话应用。
局限性与注意事项
虽然自动上下文压缩功能强大,但有一些重要的局限性需要了解:
服务端采样循环
当前限制:压缩功能在服务端采样循环(如服务端网络搜索工具)中无法达到最佳效果。
原因:缓存令牌在采样循环中累积,可能基于缓存内容而非实际对话历史触发过早的压缩。
此功能最适合以下场景:
- ✅ 客户端工具(如本教程中的客户服务 API)
- ✅ 常规工具使用的标准代理工作流
- ✅ 文件操作、数据库查询、API 调用
- ❌ 服务端扩展思考
- ❌ 服务端网络搜索工具
信息丢失
权衡:摘要本质上会丢失一些信息。虽然 Claude 擅长识别关键点,但某些细节会被压缩或省略。
在工单处理中:
- ✅ 保留:工单 ID、类别、优先级、团队、结果、进度状态
- ❌ 丢失:完整的知识库文章文本、完整的草拟回复文本、详细的分类推理
这通常是可以接受的,你不需要永久保留每篇知识库文章和完整的回复文本,只需要完成记录。
缓解措施:
- 使用自定义摘要提示词来保留关键信息
- 对于需要大量历史上下文的任务,设置更高的阈值
- 将任务结构化(每个阶段基于摘要而非原始细节)
何时不应使用压缩
避免对以下情况使用压缩:
- 短任务:如果任务在 50k-100k 令牌内完成,压缩会增加不必要的开销
- 需要完整审计跟踪的任务:某些任务需要访问所有先前的细节
- 服务端采样工作流:如上所述,等待此限制得到解决
- 高度迭代的细化:每一步都严重依赖前一步的确切细节的任务
何时应使用压缩
压缩适用于以下场景:
- 顺序处理:如我们的工单工作流——逐个处理多个项目
- 多阶段工作流:每个阶段可以在进入下一阶段前总结进度
- 迭代数据处理:分块或逐个实体处理大型数据集
- 扩展分析会话:跨多个实体分析数据
- 批量操作:处理数百个相互独立的项目
工单处理是完美的用例,因为:
- 每个工单工作流在很大程度上是独立的
- 你需要完成摘要,而不是完整的工具结果
- 存在自然的压缩点(处理完几个工单后)
- 工作流是迭代和顺序的
总结
自动上下文压缩是一项强大的功能,使长时间运行的智能体工作流能够超越典型的上下文限制。在本教程中,我们通过一个客户服务工单处理工作流探讨了压缩功能。
后续步骤
尝试在您自己的工作流中实现压缩:
- 识别自然的压缩点(处理每个项目后、完成每个阶段后等)
- 如果您有清晰的逐项边界,从激进的阈值(5k-10k)开始
- 使用自定义摘要提示以保留关键信息
- 监控压缩触发时机并验证质量是否保持
- 根据您的具体需求调整阈值
有关有效上下文管理的更多信息,请参阅面向AI代理的有效上下文工程。
常见问题
常见错误:1) 阈值设置过低导致频繁压缩,摘要本身触发压缩,建议阈值至少大于摘要生成所需令牌;2) 未正确加载环境变量导致API密钥错误,检查.env文件;3) 工具装饰器使用错误,确保使用@beta_tool且函数有文档字符串;4) 在服务端采样循环中使用压缩导致过早触发,应避免或等待SDK更新;5) 压缩后信息丢失影响任务,可通过自定义summary_prompt保留关键字段。排查方法:打印每轮消息数量和令牌使用,观察压缩触发时机;检查摘要内容是否完整;对比压缩前后最终输出质量。生产优化:使用更便宜的摘要模型(如Haiku)降低成本;根据任务边界调整阈值;监控压缩频率和令牌节省率;对于需要审计的任务,考虑外部日志记录。
企业应用扩展
可扩展方向:1) 将压缩与外部存储结合,将完整历史保存到数据库,摘要作为索引,支持按需回溯;2) 实现动态阈值,根据任务复杂度或剩余预算调整;3) 多级压缩,先压缩工具结果,再压缩对话轮次;4) 集成缓存机制,减少重复工具调用;5) 设计自适应摘要提示,根据任务类型切换模板;6) 结合服务端压缩(如Opus 4.6)与SDK压缩,实现混合策略;7) 开发可视化监控工具,实时展示上下文增长和压缩事件。
总结
本教程面向有编程基础的开发者,深入讲解如何使用Claude Agent SDK的compaction_control参数实现自动上下文压缩,以解决长任务中上下文线性增长、成本高昂和达到上下文限制的问题。通过客户服务工单处理示例,对比基线运行与启用压缩后的效果,展示令牌节省和上下文管理。同时涵盖压缩原理、自定义配置、无工具聊天循环的手动实现、局限性及生产优化建议,帮助开发者构建高效、可扩展的AI代理工作流。
- --
内容来源与验证
- 内容等级:PREMIUM_A
- 质量等级:A
- 难度:中级
- 验证等级:V1
- 原文作者:PedramNavid
- 原文日期:2025-11-24
- 来源许可:MIT
- 来源版本:
0521299a9cdbab567fc862580a7d3e3ed914171b - 查看原文
站内还有更多完整实战教程
加入会员,解锁全部教程、完整代码与持续更新内容。
加入会员