長期メモリのトークン数を見るとき、まず分母を確かめる

AI の長期メモリを比べるとき、正答率と一緒に「何トークンで答えたか」が気になります。ただ、同じ token という単位でも、根拠文だけなのか、質問や指示も含むのかで意味が変わります。

Memoket の開発に関わる立場から、今回はオープンソースの KITE の計測部分を読み直しました。性能の紹介よりも、ログを見るときに取り違えやすいところを整理します。

対象はコミット c007108f です。以下の数値は集計方法を確認するための架空データで、ベンチマークの実測値ではありません。

まず、何を数えているか

中心になるのは ChannelLedger です。

フィールド	数えている範囲	読み取れないこと
compact_total	行を選択するときの予算計算に使うカウント	完成したプロンプト全体の長さ
real_total	初回の根拠パックで、各チャネルを別々に数えた合計	後から根拠を組み直した分も含む累積量
reader_calls[].prompt_tokens の合計	計測フックに渡した各論理呼び出しのプロンプト文字列全体をローカルに数えた値	プロバイダー側の総使用量や請求額

real_total の real という名前だけを見ると、API が返す実使用量にも見えます。実際にはローカルのカウントです。また、各チャネルの合計なので、結合後の文章を一度だけトークン化した値とも区別します。

これらを、同じ条件で測った「小・中・大」の3段階として扱うのは避けたいところです。数えている対象が違います。

欠測を0にすると、平均が小さくなる

次の3行を synthetic-demo.jsonl として保存します。数値はすべて説明用に置いたものです。

{"question":"説明用の架空ログ1","telemetry":{"compact_total":800,"real_total":900,"reader_calls":[{"site":"answer","prompt_tokens":1250},{"site":"support_check","prompt_tokens":1350}]}}
{"question":"説明用の架空ログ2","telemetry":{"compact_total":1000,"real_total":1200,"reader_calls":[{"site":"answer","prompt_tokens":1500}]}}
{"question":"説明用の架空ログ3（計測なし）"}

3行目には計測値がありません。real_total の平均は (900 + 1200) / 2 = 1050。全3行で割って700にすると、「測れていない」を「使用量0」に置き換えてしまいます。

reader の論理プロンプトは、1問目が 1250 + 1350 = 2600、2問目が1500。計測された2問の平均は2050です。初回の根拠量の平均1050とは、別の問いに答える数字です。

LoCoMo の scorer も、telemetry のある行の割合を出し、部分的な計測を全件計測と比較しないよう注意を表示しています。

手元で確認するため、Python の標準ライブラリだけを使う集計スクリプトを用意しました。各指標の計測行数を別々に表示し、欠けた値は0で埋めません。

"""KITE telemetry の範囲と欠測を確認する。API料金は計算しない。"""

import argparse
import json
from pathlib import Path


def nonnegative_int(value, where):
    if type(value) is not int or value < 0:
        raise ValueError(f"{where}: expected a nonnegative integer")
    return value


def describe(values, total):
    return {
        "records_measured": len(values),
        "records_total": total,
        "mean": sum(values) / len(values) if values else None,
    }


def summarize(path, unit):
    total = telemetry_count = 0
    compact, rendered, reader = [], [], []
    with path.open(encoding="utf-8") as source:
        for line_number, line in enumerate(source, 1):
            if not line.strip():
                continue
            row = json.loads(line)
            if not isinstance(row, dict):
                raise ValueError(f"line {line_number}: expected an object")
            total += 1
            telemetry = row.get("telemetry")
            if telemetry is None:
                continue
            if not isinstance(telemetry, dict):
                raise ValueError(f"line {line_number}: invalid telemetry")
            if not telemetry:
                continue
            telemetry_count += 1
            for key, values in (("compact_total", compact), ("real_total", rendered)):
                if key in telemetry:
                    values.append(nonnegative_int(telemetry[key], f"line {line_number}: {key}"))
            if "reader_calls" in telemetry:
                calls = telemetry["reader_calls"]
                if not isinstance(calls, list):
                    raise ValueError(f"line {line_number}: reader_calls must be a list")
                amounts = []
                for call in calls:
                    if not isinstance(call, dict) or "prompt_tokens" not in call:
                        raise ValueError(f"line {line_number}: missing prompt_tokens")
                    amounts.append(nonnegative_int(call["prompt_tokens"], f"line {line_number}: prompt_tokens"))
                reader.append(sum(amounts))
    if not total:
        raise ValueError("no records")
    return {
        "unit_declared": unit,
        "records_with_telemetry": telemetry_count,
        "records_total": total,
        "row_admission": describe(compact, total),
        "initial_memory_channels": describe(rendered, total),
        "logical_reader_prompt_input": describe(reader, total),
    }


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("jsonl", type=Path)
    parser.add_argument("--unit", required=True, help="元のrun設定で確認した計数単位")
    args = parser.parse_args()
    try:
        print(json.dumps(summarize(args.jsonl, args.unit), ensure_ascii=False, indent=2))
    except (ValueError, OSError) as error:
        parser.exit(2, f"error: {error}\n")

summarize_tokens.py として保存し、同じディレクトリで実行します。

python3 summarize_tokens.py synthetic-demo.jsonl --unit demo_count

この例では、入力3行のうち telemetry があるのは2行。各指標の計測行数も2で、平均は次のようになります。

出力キー	平均	計測行数 / 入力行数
row_admission	900	2 / 3
initial_memory_channels	1050	2 / 3
logical_reader_prompt_input	2050	2 / 3

demo_count は架空の数値につけたラベルです。実ログでは元の run manifest で tokenizer を確認して指定します。このスクリプト自体は単位の正しさや、異なる run の混在、質問の重複を検証しません。元の scorer は、telemetry 自体がある行では欠けた指標を0で補う箇所があります。この例は指標ごとに分母を持つため、その処理とも異なります。公開 scorer の完全な代替ではありません。

料金を知るには、別の記録が要る

もう一つ気をつけたいのがリトライです。対象コミットでは、ledger への記録は論理呼び出しの前にあり、内側で起きる HTTP リトライや JSON の再生成試行は、それぞれ独立した記録になりません。プロバイダー実装 もレスポンスから本文を取り出し、usage は保持していません。

そのため、reader のプロンプト合計に単価を掛けても総費用にはなりません。出力やキャッシュの扱いに加え、抽出、コンパイル、判定など、reader の外で使う分も必要です。

メモリから渡す根拠を短くする工夫と、システム全体の費用を測る作業は、それぞれ役に立ちます。ログを比較する前に、数える範囲、計数単位、計測できた行数を並べておく。それだけでも、数字から言えることがはっきりします。

---

資料の確認、文章・サンプルコードの作成、架空データのローカル集計には AI アシスタントを使いました。このサンプルの集計では推論 API にアクセスせず、KITE のベンチマークも再実行していません。