Claude Agent SDK教程:自动压缩长任务上下文

本教程面向有编程基础的开发者,深入讲解如何使用Claude Agent SDK的compaction_control参数实现自动上下文压缩,以解决长任务中上下文线性增长、成本高昂和达到上下文限制的问题。通过客户服务工单处理示例,对比基线运行与启用压缩后的效果,展示令牌节省和上下文管理。同时涵盖压缩原理、自定义配置、无工具聊天循环的手动实现、局限性及生产优化建议,帮助开发者构建高效、可扩展的AI代理工作流。

技术背景

在构建基于工具调用的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 参数通过以下方式提供自动上下文管理:

  1. 监控对话中每轮的令牌使用量
  2. 当超过阈值时,注入一个摘要提示作为用户回合
  3. 让模型生成包含在 <summary></summary> 标签中的摘要。这些标签不会被解析,但用于引导模型。
  4. 清除对话历史,仅保留摘要继续
  5. 使用压缩后的上下文继续任务

通过本教程,你将能够:

先决条件

在遵循本指南之前,请确保你具备:

所需知识

所需工具

正在使用 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 中,我们定义了几个用于处理客户支持工单的函数:

对于客户服务代理,这些工具支持系统地处理工单。每个工单都需要分类、研究、优先级排序、路由和回复草拟。当按顺序处理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,
]

基线:不进行压缩的运行

让我们从一个现实的客户服务场景开始:处理一个支持工单队列。

工作流程如下:

对于每个工单:

  1. 使用 get_next_ticket() 获取工单
  2. 分类问题类别(账单、技术、账户、产品、物流)
  3. 搜索知识库以获取相关信息
  4. 设置适当的优先级(低、中、高、紧急)
  5. 路由到正确的团队
  6. 起草客户回复
  7. 将工单标记为完成
  8. 移动到下一个工单

挑战:队列中有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 必须:

为什么会这样:

  1. 线性 token 增长 - 每次使用工具时,整个对话历史(包括之前所有工具结果)都会发送给 Claude
  2. 上下文污染 - 在处理工单 B 时,工单 A 的分类和草拟回复仍保留在上下文中
  3. 成本叠加 - 当你处理到第 5 个工单时,每次 API 调用都会发送前 4 个工单的数据
  4. 响应变慢 - 处理大量上下文需要更长时间
  5. 达到限制的风险 - 最终会达到 200k token 的上下文窗口

我们真正需要的:完成工单 A 后,我们只需要一个 简要摘要(工单已解决、类别、优先级)——而不是完整的分类结果、知识库搜索和完整的草拟回复。详细的工作流应该被丢弃,只保留完成摘要。

让我们看看自动上下文压缩如何解决这个问题。

启用自动上下文压缩

让我们运行完全相同的客户服务工作流,但启用自动上下文压缩。我们只需在工具运行器中添加 compaction_control 参数。

compaction_control 参数有一个必填字段和几个可选字段:

对于此客户服务工作流,我们将使用 5,000 令牌阈值。这意味着在处理多个工单后,压缩将自动触发。这允许 Claude:

  1. 保留完成摘要(已解决的工单、类别、结果)
  2. 丢弃详细的工具结果(完整的知识库文章、完整的分类、草拟的回复文本)
  3. 在处理下一批工单时重新开始

这模拟了真实支持代理的工作方式:解决工单,简要记录,然后处理下一个案例。

# 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)

对比结果

启用压缩后,我们可以清楚地看到两次运行在令牌节省方面的明显差异,同时保持了工作流和最终摘要的质量。

以下是启用自动上下文压缩后的变化:

  1. 处理若干工单后上下文重置 - 当处理5-7个工单产生超过5k令牌的工具结果时,SDK会自动:
  1. 输入令牌保持有界 - 随着处理更多工单,输入令牌不会累积到100k以上,而是在每次压缩后重置。处理第5个工单时,我们不会携带第1-4个工单的完整工具结果。
  1. 任务成功完成 - 工作流顺利处理所有工单,不会触及上下文限制
  1. 质量得到保留 - 摘要保留了关键信息:

所有工单仍然被正确分类、确定优先级、路由和回复。

  1. 自然的工作流 - 这反映了真实支持代理的工作方式:解决一个工单,在系统中简要记录,关闭它,然后处理下一个。你不会在处理新工单时保留每篇知识库文章和完整的回复草稿。

让我们可视化令牌节省情况:

# 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 使用量超过阈值时,它会自动:

  1. 暂停工作流,在下次 API 调用之前
  2. 注入一个摘要请求,作为用户消息,要求 Claude 总结进度
  3. 生成摘要——Claude 生成一个包裹在 <summary></summary> 标签中的摘要,包含:
  1. 清除历史记录——整个对话历史(包括所有工具结果)被替换为仅包含摘要
  2. 恢复处理——Claude 使用压缩后的上下文继续工作,处理下一批工单

自定义压缩配置

您可以根据具体用例自定义压缩的工作方式。以下是关键配置选项:

调整阈值

context_token_threshold 决定何时触发压缩:

compaction_control={
    "enabled": True,
    "context_token_threshold": 5000,  # Compact after processing 5-7 tickets
}

阈值不应设置得太低,否则摘要本身可能触发压缩。我们为了演示目的设置了5,000个令牌的阈值,但在实践中,请尝试不同的设置,以找到最适合您工作流程的配置。

以下是一些通用指南:

对于工单处理: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 增长,而是来回对话本身。每次交流都会向历史记录添加消息:

如果没有压缩,到第 50 轮时,你每次 API 调用都会发送整个对话历史(所有 50 次交流)。

解决方案:使用相同的模式在聊天循环中手动实现压缩:

  1. 每轮后跟踪 token 使用情况
  2. 当超过阈值时,请求摘要
  3. 用摘要替换对话历史
  4. 使用压缩的上下文继续对话

让我们看看如何实现:

#!/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.")

理解聊天循环模式

上面的示例演示了在对话上下文中手动压缩。其工作原理如下:

关键组件:

  1. 令牌跟踪:每次响应后,计算总令牌数(输入 + 输出 + 缓存令牌)
  2. 阈值检查:当总数超过阈值时,触发压缩
  3. 摘要请求:向 Claude 发送相同的结构化 SUMMARY_PROMPT
  4. 历史替换:用摘要替换整个消息历史
  5. 继续:下一条用户消息基于摘要,而不是完整历史

何时使用此模式:

与工具运行器的关键区别:

| 方面 | 工具运行器(自动) | 聊天循环(手动) |

|--------|------------------------|-------------------|

| 触发 | 达到阈值时自动 | 你实现阈值检查 |

| 摘要 | SDK 处理摘要请求 | 你显式调用 API |

| 历史管理 | SDK 替换消息 | 你手动替换列表 |

| 用例 | 带工具的代理工作流 | 用户驱动的对话 |

生产注意事项:

  1. 调整阈值:对于实际应用使用更大的阈值
  2. 自定义摘要提示:根据你的对话类型(头脑风暴 vs. 技术支持 vs. 辅导)定制
  3. 显示用户指示:显示类似“正在总结对话...”的消息,让用户理解暂停
  4. 保留关键上下文:确保摘要提示捕获用户关心的领域特定信息

此模式让你完全控制压缩发生的时间和方式,非常适合 SDK 的自动工具运行器压缩不可用的对话应用。

局限性与注意事项

虽然自动上下文压缩功能强大,但有一些重要的局限性需要了解:

服务端采样循环

当前限制:压缩功能在服务端采样循环(如服务端网络搜索工具)中无法达到最佳效果。

原因:缓存令牌在采样循环中累积,可能基于缓存内容而非实际对话历史触发过早的压缩。

此功能最适合以下场景:

信息丢失

权衡:摘要本质上会丢失一些信息。虽然 Claude 擅长识别关键点,但某些细节会被压缩或省略。

在工单处理中:

这通常是可以接受的,你不需要永久保留每篇知识库文章和完整的回复文本,只需要完成记录。

缓解措施:

何时不应使用压缩

避免对以下情况使用压缩:

  1. 短任务:如果任务在 50k-100k 令牌内完成,压缩会增加不必要的开销
  2. 需要完整审计跟踪的任务:某些任务需要访问所有先前的细节
  3. 服务端采样工作流:如上所述,等待此限制得到解决
  4. 高度迭代的细化:每一步都严重依赖前一步的确切细节的任务

何时应使用压缩

压缩适用于以下场景:

  1. 顺序处理:如我们的工单工作流——逐个处理多个项目
  2. 多阶段工作流:每个阶段可以在进入下一阶段前总结进度
  3. 迭代数据处理:分块或逐个实体处理大型数据集
  4. 扩展分析会话:跨多个实体分析数据
  5. 批量操作:处理数百个相互独立的项目

工单处理是完美的用例,因为:

总结

自动上下文压缩是一项强大的功能,使长时间运行的智能体工作流能够超越典型的上下文限制。在本教程中,我们通过一个客户服务工单处理工作流探讨了压缩功能。

后续步骤

尝试在您自己的工作流中实现压缩:

  1. 识别自然的压缩点(处理每个项目后、完成每个阶段后等)
  2. 如果您有清晰的逐项边界,从激进的阈值(5k-10k)开始
  3. 使用自定义摘要提示以保留关键信息
  4. 监控压缩触发时机并验证质量是否保持
  5. 根据您的具体需求调整阈值

有关有效上下文管理的更多信息,请参阅面向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代理工作流。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员,解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部AI 智能体教程