Claude 跑 8 小时长任务，我中途崩溃了 3 次才学会的 4 件事

实测环境：Claude Sonnet 4.5 (claude-sonnet-4-5-20250929) · 怪招本 v3 改版 · 8 小时 · 38k token · 3 次崩溃 · 4 次跑通

8 小时任务为什么这么难

Claude 单次对话最长 200K token——听起来够用。但长任务的真正瓶颈不是 token 限额，是时间：

• Sonnet 4.5 单次推理 30 秒-2 分钟（看工具调用次数）
• 8 小时任务 = 50-100 次连续工具调用 + 推理
• 中间任何一步崩了：context 丢、token 烧、时间作废

我跑怪招本 v3 改版（5 type × 5 栏目 = 25 篇骨架）就是这个场景。一次跑完应该 6-7 小时，但实际跑了 8 小时，崩了 3 次。

3 次崩溃的真实账单（按 Sonnet 4.5 报价）：

崩溃	触发	已烧 token	已花时间	损失
第 1 次	单次 API timeout（>10 min）	18k	2.5h	$1.20
第 2 次	context window overflow	26k	5.0h	$1.85
第 3 次	tool result 超大（>100K）	32k	7.5h	$2.45

总损失：3 次崩溃累计烧了 $5.50，但产出 = 0。

第 4 次跑通，用了 4 个方法。下面是它们。

---

方法 1：Task Breakdown + 独立 checkpoint

核心思路：把 8 小时任务拆成 N 个 30 分钟子任务，每个子任务结束写 checkpoint 到磁盘。

# task.yaml - 主任务定义
main:
  goal: 怪招本 v3 改版（25 篇骨架）
  subtasks:
    - id: type-1
      goal: Commander 类型志
      duration: 30min
      checkpoint: type-1.md
    - id: type-2
      goal: Conversationalist 类型志
      duration: 30min
      checkpoint: type-2.md
    - ...

子任务之间是无状态的——下一个子任务从磁盘读上次的 checkpoint，不依赖 in-context memory。

# agent.py - checkpoint 模式
import json, os

CHECKPOINT_DIR = "/checkpoints"

def run_subtask(subtask):
    # 1. 读上次的 checkpoint（如果有）
    state_path = f"{CHECKPOINT_DIR}/{subtask['id']}.json"
    if os.path.exists(state_path):
        state = json.load(open(state_path))
        print(f"Resume from {state['completed']}")
    else:
        state = {"completed": [], "results": {}}

    # 2. 跑子任务
    for step in subtask['steps']:
        if step['id'] in state['completed']:
            continue  # 跳过已完成的
        try:
            result = call_claude(step['prompt'])
            state['results'][step['id']] = result
            state['completed'].append(step['id'])
        except ClaudeAPIError as e:
            # 3. 失败时保存 checkpoint，下次重试从这里开始
            json.dump(state, open(state_path, 'w'))
            raise e

    # 4. 完成时保存完整 checkpoint
    json.dump(state, open(state_path, 'w'))
    return state['results']

好处：

• 第 1 次崩溃在 type-1 中间 → 重启只跑 type-1 剩余部分
• 不用从头跑
• 时间损失从 8 小时 → 30 分钟

---

方法 2：单次 API timeout 防御

第 1 次崩溃的根因：Sonnet 4.5 单次推理超过 10 分钟触发 Anthropic 服务端 timeout。

Anthropic 默认 timeout 是 10 分钟（HTTP 连接）。但 Sonnet 4.5 在大量 tool nesting 时，单次推理经常跑到 8-12 分钟。

# 防御 1：客户端 timeout 设短一点
import anthropic
client = anthropic.Anthropic(
    timeout=300.0  # 5 分钟，触发超时比 10 分钟损失小
)

# 防御 2：拆任务，让单次推理 < 5 分钟
def call_claude_safe(prompt, max_steps=5):
    """限制单次推理的 tool 步骤数"""
    response = client.messages.create(
        model="claude-sonnet-4-5-20250929",
        max_tokens=8192,
        tools=TOOLS,
        messages=[{"role": "user", "content": prompt}]
    )

    steps = 0
    while response.stop_reason == "tool_use" and steps < max_steps:
        steps += 1
        # ... 处理 tool call ...

关键 trick：max_steps=5 限制单次推理的 tool 嵌套次数。超了就 break，让外层代码决定怎么处理（重试 / 拆分 / 跳过）。

---

方法 3：Context Window Overflow 防御

第 2 次崩溃的根因：对话历史累积超过 200K token。

怪招本 v3 任务里，每次 tool result 都把上一轮的整段内容塞回 context。跑 5 小时后 context 突破 150K token，最后一次 API 调用失败。

# 防御：定期压缩 / 截断对话历史
def compress_history(messages, target_tokens=100_000):
    """每 50 轮压缩一次历史"""
    if len(messages) < 50:
        return messages

    # 保留前 5 轮 + 后 40 轮，中间 5 轮做 summary
    head = messages[:5]
    middle = messages[5:-40]
    tail = messages[-40:]

    # 用 Claude 压缩中间部分
    summary_prompt = f"""总结以下对话的关键决策和产出，保留所有事实和数据：
{middle}"""

    summary = client.messages.create(
        model="claude-sonnet-4-5-20250929",
        max_tokens=4096,
        messages=[{"role": "user", "content": summary_prompt}]
    ).content[0].text

    return head + [{"role": "user", "content": f"[前面对话摘要]\n{summary}"}] + tail

实测数据：怪招本 v3 改版用这个方法后，context 从”5 小时崩溃” → “8 小时平跑”。

token 使用对比：

时刻	不用压缩	用压缩
1h	32K	32K
4h	98K	64K
6h	156K	82K
8h	崩溃（>200K）	96K

---

方法 4：Tool Result 大小限制

第 3 次崩溃的根因：单次 tool result 太大（>100K token）。

我的 vector_search 工具一次返回 1000 条结果，每条 100 字符 = 100K token，直接撑爆 context。

# 防御：限制 tool result 大小
MAX_TOOL_RESULT_TOKENS = 20_000  # 20K 是安全线

def truncate_tool_result(result: str, max_tokens: int = MAX_TOOL_RESULT_TOKENS) -> str:
    # 粗略估算：4 字符 = 1 token
    max_chars = max_tokens * 4

    if len(result) <= max_chars:
        return result

    # 截断 + 加摘要说明
    truncated = result[:max_chars]
    return f"{truncated}\n\n[... 已截断，原结果 {len(result)} 字符，超出 {max_chars} 字符限制。如需更多数据请缩小查询范围 ...]"

配套修改：让 tool 返回结构化数据而不是大段文本。

# 不要这样
def vector_search(query):
    return "\n".join([f"{r['title']}: {r['content']}" for r in results])

# 要这样
def vector_search(query, top_k=10):
    return [
        {"id": r["id"], "title": r["title"], "score": r["score"]}
        for r in results[:top_k]
    ]  # Claude 需要详情再调 get_detail(id)

---

最终战报（第 4 次跑通）

任务:        怪招本 v3 改版（25 篇骨架 + 5 张 SVG + 26 页结构）
时长:        7 小时 23 分
token:       38,142 (input 24K / output 14K)
账单:        $1.85
崩溃次数:    0
方法:        4 个（breakdown + timeout + compress + truncate）

对比之前 3 次崩溃：累计 $5.50 烧成 0，第 4 次 $1.85 跑通。

---

给也想跑长任务的 4 条建议

1. 必须拆任务——8 小时任务 = 16 个 30 分钟子任务。每个子任务结束写 checkpoint 到磁盘。崩了重启只损失 30 分钟，不是 8 小时。

2. 限制单次推理 < 5 分钟——Anthropic 默认 timeout 10 分钟，但工具嵌套多时单次推理经常超过。客户端 timeout 设 5 分钟 + max_steps=5 限制嵌套。

3. 每 50 轮压缩历史——context overflow 是最隐蔽的崩溃。compress_history() 函数每 50 轮跑一次，把中间的 5 轮摘要成 1 轮。

4. Tool result 不超过 20K token——返回结构化数据 + 详情单独调。大段文本直接撑爆 context。

---

现场：第 4 次跑通的 checkpoint 时间线

[00:00] type-1: Commander  ← checkpoint 写盘
[00:31] type-2: Conversationalist  ← checkpoint 写盘
[01:05] type-3: Supervisor  ← checkpoint 写盘
...
[07:23] ✅ 25/25 完成，全部分发到怪招本 dist/

崩溃次数：0
token: 38,142
账单: $1.85

之前 3 次都烧成 $5.50 + 0 产出。第 4 次用 4 个方法，$1.85 跑通 + 25 篇骨架。

---

附：完整的 agent.py 模板

import anthropic
import json
import os

client = anthropic.Anthropic(timeout=300.0)
CHECKPOINT_DIR = "/checkpoints"
MAX_TOOL_RESULT_TOKENS = 20_000

def load_checkpoint(task_id):
    path = f"{CHECKPOINT_DIR}/{task_id}.json"
    if os.path.exists(path):
        return json.load(open(path))
    return {"completed": [], "results": {}}

def save_checkpoint(task_id, state):
    path = f"{CHECKPOINT_DIR}/{task_id}.json"
    json.dump(state, open(path, 'w'))

def truncate_tool_result(result, max_tokens=MAX_TOOL_RESULT_TOKENS):
    max_chars = max_tokens * 4
    if len(result) <= max_chars:
        return result
    return result[:max_chars] + f"\n\n[... 已截断，原 {len(result)} 字符 ...]"

def compress_history(messages, target_size=40):
    if len(messages) < 50:
        return messages
    head = messages[:5]
    middle = messages[5:-target_size]
    tail = messages[-target_size:]
    summary = client.messages.create(
        model="claude-sonnet-4-5-20250929",
        max_tokens=4096,
        messages=[{"role": "user", "content": f"总结：{middle}"}]
    ).content[0].text
    return head + [{"role": "user", "content": f"[摘要]\n{summary}"}] + tail

def run_subtask_safe(subtask):
    state = load_checkpoint(subtask['id'])
    for step in subtask['steps']:
        if step['id'] in state['completed']:
            continue
        try:
            result = client.messages.create(
                model="claude-sonnet-4-5-20250929",
                max_tokens=8192,
                messages=compress_history([
                    {"role": "user", "content": step['prompt']}
                ])
            )
            state['results'][step['id']] = truncate_tool_result(
                result.content[0].text
            )
            state['completed'].append(step['id'])
            save_checkpoint(subtask['id'], state)  # 每步保存
        except Exception as e:
            save_checkpoint(subtask['id'], state)
            raise
    return state['results']

这套模板让 8 小时任务崩溃成本从 8 小时 → 30 分钟。

下一篇 T7 讲怎么把 prompt cache 命中率从 60% 拉到 95%——比 t2 那篇再深入一层。

— 怪招本 #011 · 2026-06-28