Claude API のストリーミングが「エラーも出さずに」止まるとき — 沈黙する停止を検知して途中から続ける運用メモ

def resume_stream(client, base_messages, received_text, **kw):
    """
    received_text（停止時点までの本文）を assistant プレフィルとして渡し、続きを生成する。
    """
    messages = base_messages + [
        {"role": "assistant", "content": received_text},
    ]
    # continued は received_text の「続き」だけが返る
    for text in stream_with_watchdog(client, messages, **kw):
        yield text
 
 
def robust_generate(client, base_messages, max_resumes=3, **kw):
    full = ""
    messages = base_messages
    for attempt in range(max_resumes + 1):
        try:
            chunks = (stream_with_watchdog if attempt == 0 else resume_stream)
            args = (client, base_messages, full, *([] )) if attempt else (client, base_messages,)
            gen = resume_stream(client, base_messages, full, **kw) if attempt else stream_with_watchdog(client, base_messages, **kw)
            for text in gen:
                full += text
            return full  # 正常完了
        except StreamStalled as e:
            full += e.args[0] if e.args and isinstance(e.args[0], str) else ""
            # 次のループでプレフィル再開
            continue
    return full

def stitch(prev: str, cont: str, max_overlap: int = 80) -> str:
    """prev の末尾と cont の先頭の重複を検出して除去し、滑らかに結合する"""
    window = prev[-max_overlap:]
    for size in range(min(len(window), len(cont)), 0, -1):
        if window.endswith(cont[:size]):
            return prev + cont[size:]
    return prev + cont

def deadline_for(received_any: bool, first_token_s: float, inter_token_s: float) -> float:
    """まだ1つも delta が来ていなければ初トークン予算、来ていればトークン間予算を返す"""
    return first_token_s if not received_any else inter_token_s

import logging, time
logger = logging.getLogger("claude.stream")
 
def timed_stream(client, messages, **kw):
    t0 = time.monotonic()
    first_t = None
    gaps = []
    last = t0
    for text in stream_with_watchdog(client, messages, **kw):
        now = time.monotonic()
        if first_t is None:
            first_t = now - t0
        else:
            gaps.append(now - last)
        last = now
        yield text
    total = time.monotonic() - t0
    p99_gap = sorted(gaps)[int(len(gaps) * 0.99)] if gaps else 0.0
    logger.info("first_token=%.2fs total=%.2fs max_gap=%.2fs p99_gap=%.2fs",
                first_t or -1, total, max(gaps) if gaps else 0.0, p99_gap)

location /api/chat {
    proxy_pass http://backend;
    proxy_http_version 1.1;
    proxy_set_header Connection "";
 
    proxy_buffering off;          # これが無いとデルタがまとまって遅延する
    proxy_cache off;
    chunked_transfer_encoding on;
 
    proxy_connect_timeout 10s;
    proxy_read_timeout 600s;      # ソケット層の保険（沈黙停止はアプリ側で別に守る）
    proxy_send_timeout 600s;
}

def assert_complete(final_message) -> None:
    sr = final_message.stop_reason
    if sr == "max_tokens":
        raise RuntimeError("max_tokens 到達。続きが残っている可能性")
    if sr not in ("end_turn", "stop_sequence"):
        raise RuntimeError(f"未完了の終了理由: {sr}")