Kimi 端点 256k 间歇墙根因与 PreCompact 拦截实战

Claude_Code Kimi

Claude Code 直连 Kimi 官方端点(api.kimi.com/coding,模型 kimi-k3[1m] 标称 1M 上下文窗口),上下文用到 ~26% 就自动压缩,后来干脆直接报 400。本文是从 transcript 取证 → CLI 二进制逆向 → 主动探测 → PreCompact hook 拦截的完整复盘。

TL;DR

  • 根因:Kimi 端点 1M/256k 池混部、按请求间歇路由。命中 256k 池时,>262144 tokens 的请求被 400 拒绝,CLI 误判”窗口已满”触发响应式压缩——实际 1M 窗口远没用满。
  • 锤死证据:错误原文 Your request exceeded model token limit: 262144 (requested: 262561);且同会话 80 秒后 456k 请求正常通过。
  • 解法:PreCompact hook 拦截 auto 压缩(二进制实证可行),手动 /compact 保留为逃生门。
  • 注意:Kimi 新错误文案绕开了 CLI 的 PTL 分类器,不再触发压缩、直接报错——两种文案两种表现,下文详述。

一、现象

时间表现
08-17 ~ 08-18上下文用到 230k314k(相对 1M 显示 23%~32%)就自动压缩,无确认、无预警
08-19 12:36 探测272k / 615k / 988k / 1,095k 全部 HTTP 200,以为”已恢复”
08-19 17:07会话中突然 API Error: 400 ... exceeded model token limit: 262144,不压缩、直接报错,看起来就是”上下文又重置了”

用户的体感是”上下文提前被干掉”,但两种表现的机制完全不同,需要分开定性。

二、环境

  • CLI:Claude Code 2.1.235(claude.exe,330MB node SEA 单文件)
  • 端点:https://api.kimi.com/coding(官方直连,不经任何中转 relay,项目级 settings.local.json 注入 ANTHROPIC_BASE_URL / ANTHROPIC_AUTH_TOKEN)
  • 模型:kimi-k3[1m]([1m] 后缀 = 1M 窗口变体)
  • 全局配置走的是另一个端点(z.ai/glm),与本问题无关——先确认链路再排查,避免 v1 排查时把锅甩给不存在的 relay。

三、排查过程

3.1 transcript 取证:先拿事件清单,不猜

Claude Code 的会话 transcript(~/.claude/projects/<project-slug>/*.jsonl)是黑匣子级的取证源。每次完成的压缩都会落一条带 compactMetadata 的记录:

python3 - "$f" <<'EOF'
import json,sys
for line in open(sys.argv[1], errors='replace'):
    if '"trigger":"auto"' not in line: continue
    obj = json.loads(line)
    cm = obj.get('compactMetadata') or {}
    print(obj.get('timestamp'), cm.get('trigger'), cm.get('preTokens'))
EOF

全项目扫描结果:

时间(北京)triggerpreTokens相对 1M
08-17 15:13auto314,234~32%
08-17 21:42auto230,459~23%
08-18 15:11auto229,352~23%
08-18 15:38auto263,605~26%

对照组:手动 /compact × 5(08-0408-17 凌晨)@ 565k697k 全被 API 正常接受——证明窗口本身远超 314k,压缩点异常。

注意:该清单只统计完成的压缩;被 CTRL+C 中断的压缩不落 compactMetadata,查不到。

3.2 CLI 压缩机制:二进制逆向(strings 取证)

对 claude.exe 做 strings/grep 逆向(grep -aoE '.{120}关键词.{150}' claude.exe),确认:

  1. 唯一生效的自动压缩 = 响应式:仅当 API 返回 400/413 且 body 含以下任一关键词时触发:
    • prompt is too long
    • input is too long for requested model
    • context window
    • input length and max_tokens exceed context limit
  2. 分类器(内部符号 mQo)命中后,compactMetadata 记 trigger:"auto";429/403/5xx 不触发
  3. 阈值式/预计算压缩被特性门 tengu_sepia_moth=false 锁死——CLAUDE_CODE_AUTO_COMPACT_WINDOW 等环境变量过不了门,CLI 侧没有任何可调旋钮。
  4. 压缩是完成时一次性提交(摘要 + compactMetadata 作为边界整体写入)——中途 CTRL+C 安全,上下文原样保留。

结论:自动压缩不是 CLI 的阈值策略,是对 API 报错的被动响应。问题在端点。

3.3 主动探测:count_tokens 配料法

响应式排查太被动,改主动探测——构造精确尺寸的请求直接打端点:

  1. 先用 POST /coding/v1/messages/count_tokens 量出填充文本的精确 token 数(不消耗生成);
  2. 裁剪到目标尺寸后 POST /coding/v1/messages,max_tokens=1,只看 HTTP 码和 body。

08-19 12:36 探测:272k / 615k / 988k / 1,095k 全部 200 → 当时端点无 ≤1.09M 的墙,误判为”已自行恢复”。完整脚本见附录。

3.4 转折:08-19 17:07 新错误文案锤死窗口值

当天傍晚会话 transcript 实录:

17:07:05  ctx=261,974  正常工作
17:07:19  API Error: 400 Invalid request: Your request exceeded model token limit: 262144 (requested: 262561)
17:07:36  API Error: 400 ... (requested: 262579)
17:08:55  ctx=456,852  HTTP 200 恢复,继续跑到 482k

三个决定性信息:

  1. 窗口值锤死 = 262144(256k)。之前 230k 的压缩点只是反推(229k314k 全 > 224k ≈ 256k - 32k 输出预留,与 256k 池吻合),这次错误原文直接报出数字。
  2. 间歇性实锤:80 秒前 262.5k 被拒,80 秒后 456k 通过——不是端点全天降级,是按请求路由,命中 256k 池则拒
  3. 行为分叉(更坏):新文案 exceeded model token limit: N 不含 3.2 节分类器认的任何关键词 → 不判 PTL、不触发压缩,400 直接抛给用户,会话卡住。08-17/18 的旧文案还能触发压缩”自救”,新文案连自救都没有。

当晚 20:45 复探测:263.5k(超墙 1.4k)× 5 全 200 → 路由又回到 1M 池。间歇性意味着任何时刻都可能再撞

四、根因

Kimi 官方端点上游 1M 池与 256k 池混部,按请求负载均衡路由:

  • 命中 1M 池:正常,kimi-k3[1m] 跑满 1M 窗口;
  • 命中 256k 池:>262144 tokens 的请求被 400 拒绝。

CLI 侧无辜:旧文案时它按设计响应式压缩(牺牲上下文换可用);新文案时分类器不认识,直接报错。两种情况的共同点是用户上下文”提前消失”的体感

非配额问题(08-13 的 403 Quota 是另一件事)、非 CLI 配置问题、非中转问题(无中转)。

五、解决方案:PreCompact hook 拦截自动压缩

CLI 没有防压缩旋钮,但 PreCompact hook 可以拦截——这是给”无确认直接压”的原生行为加闸门。

5.1 可行性:二进制契约实证

Reactive compact blocked by PreCompact hook: ${y.blockedBy}     ← 响应式压缩可拦
Compaction blocked by PreCompact hook
hook_event_name:"PreCompact", trigger: Or(["manual","auto"])    ← stdin 契约
wU({..., matchQuery: t.trigger, ...})                            ← settings matcher 按 trigger 匹配

拦截信号沿用 hook 通用约定:exit code 2 + stderr = 拦截,stderr 为理由

5.2 hook 脚本

.claude/hooks/block_auto_compact.py(完整可跑):

#!/usr/bin/env python3
"""PreCompact hook:拦截 auto(响应式)自动压缩。
trigger=auto → 拦截(exit 2),上下文原样保留,重试赌路由回 1M 池;
trigger=manual → 放行(手动 /compact 永远是逃生门)。
日志: /tmp/precompact_hook.log"""
import json, sys, datetime
 
LOG = "/tmp/precompact_hook.log"
 
def last_ctx_tokens(transcript_path):
    """读 transcript 最后一条 assistant usage,估算触发时上下文 tokens。"""
    if not transcript_path:
        return None
    try:
        last = None
        with open(transcript_path, errors="replace") as f:
            for line in f:
                if '"usage"' not in line:
                    continue
                try:
                    obj = json.loads(line)
                except Exception:
                    continue
                if obj.get("type") != "assistant":
                    continue
                u = (obj.get("message") or {}).get("usage") or {}
                if u.get("input_tokens") is None:
                    continue
                last = (u.get("input_tokens", 0)
                        + u.get("cache_read_input_tokens", 0)
                        + u.get("cache_creation_input_tokens", 0))
        return last
    except Exception:
        return None
 
def main():
    try:
        data = json.load(sys.stdin)
    except Exception:
        return 0                      # 解析失败默认放行,绝不卡死 CLI
    if data.get("trigger") != "auto":
        return 0                      # 手动 /compact 永远放行
    ctx = last_ctx_tokens(data.get("transcript_path"))
    ctx_str = f"~{ctx // 1000}k tokens({ctx / 10000:.0f}% of 1M)" if ctx else "未知"
    msg = (f"[auto-compact 已拦截] 触发时上下文 {ctx_str}。"
           "Kimi 端点 1M/256k 池间歇路由误触概率高,"
           "建议:直接重发消息重试;连续 3+ 次仍 400 说明墙粘死,再手动 /compact。")
    try:
        with open(LOG, "a") as f:
            f.write(f"{datetime.datetime.now().isoformat()} BLOCKED auto-compact "
                    f"ctx={ctx} session={data.get('session_id')}\n")
    except Exception:
        pass
    sys.stderr.write(msg + "\n")
    return 2
 
if __name__ == "__main__":
    sys.exit(main())

设计要点:

  • 只拦 auto:matcher 和脚本双重判断,手动 /compact 不受任何影响;
  • 失败放行:stdin 解析异常返回 0,hook 永远不能成为新的故障点;
  • 带读数的提示:从 transcript 读触发时的真实 token 数,用户一看就知道是不是误触(26% 被压 = 明显误触);
  • 留痕:拦截写日志,长期可统计墙的出现频率。

5.3 注册到 settings.json

"hooks": {
  "PreCompact": [
    {
      "matcher": "auto",
      "hooks": [
        { "type": "command",
          "command": "python3 \"/path/to/.claude/hooks/block_auto_compact.py\"" }
      ]
    }
  ]
}

改完后开一次 /hooks 菜单或重启会话生效(配置监听只对会话启动时已存在 settings 文件的目录生效)。

5.4 验证

管道三连测(合成 stdin 直灌脚本):

# auto → 期望 exit 2 + 拦截消息
echo '{"trigger":"auto","transcript_path":"<真实transcript>","session_id":"t"}' | python3 block_auto_compact.py
# → [auto-compact 已拦截] 触发时上下文 ~198k tokens(20% of 1M)。...  exit=2
# manual → exit 0;垃圾输入 → exit 0

jq 校验注册:

jq -e '.hooks.PreCompact[] | select(.matcher=="auto") | .hooks[].command' .claude/settings.json

PreCompact 无法在会话中主动触发,无法做端到端实证,等下次真撞墙时自然验证——这也是 hook 类配置的共同局限。

5.5 拦截的边界

hook 只够得着旧文案路径(报错含 PTL 关键词 → 触发压缩 → hook 拦截)。新文案路径(exceeded model token limit)压根不触发压缩,没有压缩可拦——那种情况直接重试即可,本来就是无损的。

六、应急 SOP(撞墙时)

  1. 看报错文案:
    • exceeded model token limit: 262144 → 没触发压缩,直接重发消息,等路由回 1M 池(实测间歇期 ~80 秒自愈);
    • 开始转圈 “Compacting…” → 已被 hook 拦截则按提示重试;没装 hook 可 CTRL+C 中断(安全,压缩完成才提交,中断即原样保留)。
  2. 不确定墙还在不在,跑探测脚本(见附录)10 秒定生死:263k 全 200 → 放心重试;有 400 → 墙粘着,手动 /compact 压到 256k 以下继续干活。
  3. 连续 3+ 次 400(间隔几分钟仍失败)→ 粘性降级(08-17/18 曾粘 ~24 小时),别等,手动 /compact

七、方法论沉淀

  1. transcript 是第一取证源:compactMetadata.preTokens 直接给出每次压缩的触发规模,比任何体感/截图都硬。
  2. strings 逆向定机制:330MB 的 claude.exe 用 grep -aoE '.{N}关键词.{M}' 抽上下文,契约(stdin 字段/matcher/exit code 语义)全在二进制里,比翻文档准。
  3. count_tokens 配料探测:想精确复现”尺寸相关”的服务端行为,先用 count_tokens 量尺寸再打 max_tokens=1 的实请求,成本最低、结论最硬。
  4. 同一故障可以有两种表象:服务端错误文案变更(旧→新)直接改变了 CLI 的自救行为——排查”行为变了”时,先查上游返回的原文,别只盯客户端版本。
  5. 响应式压缩无确认环节是原生设计——“直接压缩没提示”不是 bug,要闸门就自己加 PreCompact hook。

附录:探测脚本

#!/usr/bin/env python3
"""Kimi 端点 256k 墙探测:count_tokens 配料 + max_tokens=1 实测。
token 从环境变量 ANTHROPIC_AUTH_TOKEN 读,不要写死。"""
import json, os, time, urllib.request, urllib.error
 
BASE  = os.environ.get("ANTHROPIC_BASE_URL", "https://api.kimi.com/coding")
TOKEN = os.environ["ANTHROPIC_AUTH_TOKEN"]
MODEL = "kimi-k3[1m]"
HDRS  = {"Authorization": f"Bearer {TOKEN}",
         "anthropic-version": "2023-06-01",
         "content-type": "application/json"}
 
def post(path, body, timeout=180):
    req = urllib.request.Request(BASE + path, data=json.dumps(body).encode(),
                                 headers=HDRS, method="POST")
    t0 = time.time()
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return r.status, json.loads(r.read().decode()), time.time() - t0
    except urllib.error.HTTPError as e:
        raw = e.read().decode(errors="replace")[:400]
        try:    payload = json.loads(raw)
        except Exception: payload = {"raw": raw}
        return e.code, payload, time.time() - t0
 
# 配料:实测 46 字符 ≈ 10 tokens;先做一大块再按 count_tokens 精确裁剪
filler = "the quick brown fox jumps over the lazy dog. " * 26400
s, b, _ = post("/v1/messages/count_tokens",
               {"model": MODEL, "messages": [{"role": "user", "content": filler}]})
n = b["input_tokens"]
trim = int(len(filler) * 263_500 / n)          # 目标:略超 262144,复现撞墙尺寸
msgs = [{"role": "user", "content": filler[:trim] + "\nReply with: ok"}]
print("calibrated:", post("/v1/messages/count_tokens",
                          {"model": MODEL, "messages": msgs})[1])
 
for i in range(5):                              # 连发 5 次,采样池路由命中率
    s, b, dt = post("/v1/messages", {"model": MODEL, "max_tokens": 1, "messages": msgs})
    if s == 200:
        print(f"[{i+1}/5] HTTP 200 ({dt:.1f}s) in={b.get('usage',{}).get('input_tokens')}")
    else:
        print(f"[{i+1}/5] HTTP {s} ({dt:.1f}s) -> {json.dumps(b, ensure_ascii=False)[:300]}")
    time.sleep(2)

成本提示:每发一次 263k 探测约耗 263k input tokens,5 连发 ~1.3M;墙是间歇的,单次通过不完全代表墙不存在,5 连发是性价比权衡。