会話状態をサーバーに預けるべきか — 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) -> tuple[str, Conversation]:
    """1ターンを送る。call_model は実際の Interactions API 呼び出しを包む関数で、
    (prompt_parts, previous_interaction_id) を受け取り (reply_text, interaction_id) を返す。"""
 
    heavy = tool_output is not None and \
        estimate_tokens(tool_output) > HEAVY_TURN_TOKEN_THRESHOLD
 
    if heavy 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