会話状態をサーバーに預けるべきか — previous_interaction_id と手元での履歴刈り込みの分岐設計

import re
from dataclasses import dataclass, field
 
# 1ターンのツール出力がこの推定トークン数を超えたら、
# 連鎖に載せ続けず要約へ差し替える判断に入る
HEAVY_TURN_TOKEN_THRESHOLD = 4000
 
 
def estimate_tokens(text: str) -> int:
    """日本語混在テキストの粗いトークン見積もり。
    厳密なトークナイザではなく、閾値判定に足る近似で十分。
    英数字は約4文字/トークン、それ以外(主に日本語)は約1.6文字/トークンとして合算する。"""
    ascii_chars = len(re.findall(r"[\x00-\x7F]", text))
    other_chars = len(text) - ascii_chars
    return int(ascii_chars / 4 + other_chars / 1.6)
 
 
@dataclass
class Conversation:
    """会話ごとの状態。server_interaction_id が生きていれば連鎖を継ぐ。
    切り直した場合はここに手元要約(carryover)を蓄える。"""
    server_interaction_id: str | None = None
    carryover: list[str] = field(default_factory=list)  # 手元に残す短い要約
 
 
def summarize_heavy_output(raw: str) -> str:
    """重い出力を、次ターン以降に引きずる価値のある最小限へ畳む。
    実際にはここに軽量モデルの1回呼び出しやルールベース抽出を置く。
    要点は「何が確定したか」だけを残し、原文は捨てること。"""
    head = raw.strip().splitlines()[0][:80] if raw.strip() else ""
    return f"[前ターンのツール出力を要約] {head} … 詳細は破棄済み"
 
 
def send_turn(conv: Conversation, user_text: str, tool_output: str | None,
              call_model, tool_roundtrip_open: bool = False
              ) -> tuple[str, Conversation]:
    """1ターンを送る。call_model は実際の Interactions API 呼び出しを包む関数で、
    (prompt_parts, previous_interaction_id) を受け取り (reply_text, interaction_id) を返す。
    tool_roundtrip_open=True は「モデルが関数呼び出しを返し、その結果をまだ返していない」
    途中状態を表す。この間に連鎖を切ると呼び出しと結果が離れ、応答が噛み合わなくなる。"""
 
    heavy = tool_output is not None and \
        estimate_tokens(tool_output) > HEAVY_TURN_TOKEN_THRESHOLD
 
    # 切り直してよいのは、ツールの往復が閉じたあとだけ
    if (heavy and not tool_roundtrip_open) or conv.server_interaction_id is None:
        # 連鎖を切り直す: これまでの carryover を手元で先頭に積み、
        # 重いツール出力はモデルに渡す前に要約へ差し替える
        parts: list[str] = list(conv.carryover)
        if tool_output is not None:
            folded = summarize_heavy_output(tool_output) if heavy else tool_output
            parts.append(folded)
            if heavy:
                conv.carryover.append(folded)  # 次回以降に引きずるのは要約だけ
        parts.append(user_text)
        reply, new_id = call_model(prompt_parts=parts, previous_interaction_id=None)
    else:
        # happy path: サーバー側連鎖に素直に継ぐ。手元は空でよい
        parts = [tool_output, user_text] if tool_output else [user_text]
        reply, new_id = call_model(prompt_parts=parts,
                                   previous_interaction_id=conv.server_interaction_id)
 
    conv.server_interaction_id = new_id
    return reply, conv

import json
from collections import defaultdict
 
 
def load_ledger(path: str) -> dict[str, list[int]]:
    """自前カウンタの JSONL を会話IDごとの入力トークン列に畳む。
    1行は {"conversation_id": "...", "turn": 3, "input_tokens": 21840} 程度を想定。"""
    per_conv: dict[str, list[int]] = defaultdict(list)
    with open(path, encoding="utf-8") as f:
        for line in f:
            if not line.strip():
                continue
            rec = json.loads(line)
            per_conv[rec["conversation_id"]].append(
                (int(rec.get("turn", 0)), int(rec["input_tokens"]))
            )
    # ターン順に並べ直してトークン列だけ返す
    return {cid: [t for _, t in sorted(turns)] for cid, turns in per_conv.items()}
 
 
def drag_report(per_conv: dict[str, list[int]], min_turns: int = 4) -> list[dict]:
    """引きずりが効いている会話を、累積で食われたトークンの多い順に並べる。"""
    rows = []
    for cid, inputs in per_conv.items():
        if len(inputs) < min_turns:
            continue  # 往復が短い会話は引きずる前に終わるので対象外
        base = max(inputs[0], 1)
        # 全ターンが初回と同じ重さだった場合との差分＝累積文脈が食べた分
        carried = sum(inputs) - base * len(inputs)
        rows.append({
            "conversation_id": cid,
            "turns": len(inputs),
            "base_in": base,
            "last_in": inputs[-1],
            "drag_ratio": round(inputs[-1] / base, 1),
            "carried_tokens": max(carried, 0),
        })
    rows.sort(key=lambda r: r["carried_tokens"], reverse=True)
    return rows
 
 
if __name__ == "__main__":
    report = drag_report(load_ledger("usage_ledger.jsonl"))
    for row in report[:10]:
        print("{conversation_id}\tturns={turns}\tx{drag_ratio}\t"
              "carried={carried_tokens}".format(**row))