Log in
benchmarksPublic

Viewing files at commit e60fa60d3a02Commit detailsView latest

Browse files
scripts/oakbench/tokens.py 201 lines · 9.0 KB · python
1
"""Token accounting policy: estimates, billing direction, cost weights.
2
3
Billing direction: text the model EMITS (commands it types, prose) is priced
4
as output tokens; text it INGESTS (tool results it reads back) is input; cache
5
reads are heavily discounted. The historical estimated_tokens_input/output
6
field names are CLI-perspective and kept only for backward compatibility.
7
8
char/4 is a tokenizer-neutral proxy, biased for hash-heavy output; quantify
9
the bias with scripts/token_calibration.py before cross-subject token claims.
10
"""
11
12
from __future__ import annotations
13
14
from typing import Any
15
16
EMITTED_TOKEN_WEIGHT = 5.0
17
INGESTED_TOKEN_WEIGHT = 1.0
18
CACHE_READ_TOKEN_WEIGHT = 0.1
19
COST_WEIGHTS_NOTE = (
20
    f"emitted={EMITTED_TOKEN_WEIGHT},ingested={INGESTED_TOKEN_WEIGHT},cache_read={CACHE_READ_TOKEN_WEIGHT}"
21
)
22
23
# A tool call is never just the command text. The model emits a structured
24
# tool-use block (tool name, JSON argument framing, often a description
25
# field), and the harness feeds the result back inside a tool-result block.
26
# These constants are the midpoint of Anthropic/OpenAI tool-call framing as
27
# measured by `scripts/token_calibration.py --envelope` (which compares them
28
# against representative templates and flags drift); re-run it per provider
29
# before publishing absolute costs. They are additive fields: the historical
30
# estimated_tokens_* fields exclude envelopes and stay byte-identical
31
# (row parity, ADR-0005).
32
TOOL_CALL_ENVELOPE_EMITTED_TOKENS = 58
33
TOOL_RESULT_ENVELOPE_INGESTED_TOKENS = 44
34
ENVELOPE_NOTE = (
35
    f"per_call_emitted={TOOL_CALL_ENVELOPE_EMITTED_TOKENS},"
36
    f"per_call_ingested={TOOL_RESULT_ENVELOPE_INGESTED_TOKENS}; "
37
    "midpoint of exact o200k_base template measurements 2026-06-11 "
38
    "(anthropic tool_use/result 61/45, openai function call/result 56/43; "
39
    "validate with token_calibration.py --envelope)"
40
)
41
DIRECTION_NOTE = (
42
    "input/output are CLI-perspective (deprecated naming). For model billing, the "
43
    "command is model OUTPUT (agent_emitted) and stdout/stderr is model INPUT (agent_ingested)."
44
)
45
46
47
def estimate_tokens(text: str) -> int:
48
    if not text:
49
        return 0
50
    return max(1, (len(text) + 3) // 4)
51
52
53
def cost_weighted_total(emitted_tokens: float, ingested_tokens: float, cache_read_tokens: float = 0.0) -> float:
54
    return round(
55
        emitted_tokens * EMITTED_TOKEN_WEIGHT
56
        + ingested_tokens * INGESTED_TOKEN_WEIGHT
57
        + cache_read_tokens * CACHE_READ_TOKEN_WEIGHT,
58
        1,
59
    )
60
61
62
def int_or_zero(value: Any) -> int:
63
    """Best-effort non-negative integer coercion for additive counters.
64
65
    Rows may carry ``null`` for unmeasured fields (ADR-0002). Summary rows
66
    aggregate the measured children they have, so absent/null/non-numeric
67
    additive counters contribute zero instead of crashing the lane.
68
    """
69
    if value is None or isinstance(value, bool):
70
        return 0
71
    try:
72
        parsed = int(value)
73
    except (TypeError, ValueError):
74
        return 0
75
    return parsed if parsed >= 0 else 0
76
77
78
def envelope_token_fields(tool_call_count: int, emitted_tokens: int, ingested_tokens: int) -> dict[str, Any]:
79
    """Additive envelope accounting for rows that represent tool calls.
80
81
    tool_call_count == 0 (diagnostic probes) yields zero envelope overhead but
82
    the fields still appear so dashboards can sum without key checks.
83
    """
84
    envelope_emitted = tool_call_count * TOOL_CALL_ENVELOPE_EMITTED_TOKENS
85
    envelope_ingested = tool_call_count * TOOL_RESULT_ENVELOPE_INGESTED_TOKENS
86
    return {
87
        "estimated_tokens_envelope_emitted": envelope_emitted,
88
        "estimated_tokens_envelope_ingested": envelope_ingested,
89
        "estimated_tokens_total_with_envelope": emitted_tokens + ingested_tokens + envelope_emitted + envelope_ingested,
90
        "estimated_cost_weighted_tokens_with_envelope": cost_weighted_total(
91
            emitted_tokens + envelope_emitted, ingested_tokens + envelope_ingested
92
        ),
93
    }
94
95
96
def interaction_token_fields(
97
    command_text: str,
98
    stdout_text: str,
99
    stderr_text: str,
100
    stdout_bytes: int,
101
    stderr_bytes: int,
102
    stdout_truncated: bool,
103
    stderr_truncated: bool,
104
    source: str = "command_plus_admitted_output_chars_div_4",
105
    tool_call_count: int = 1,
106
) -> dict[str, Any]:
107
    """The flat token fields plus nested token_estimate every CLI-lane row carries."""
108
    command_tokens = estimate_tokens(command_text)
109
    output_tokens = estimate_tokens(stdout_text) + estimate_tokens(stderr_text)
110
    total_tokens = command_tokens + output_tokens
111
    cost_weighted = cost_weighted_total(command_tokens, output_tokens)
112
    envelope_fields = envelope_token_fields(tool_call_count, command_tokens, output_tokens)
113
    return {
114
        **envelope_fields,
115
        "estimated_tokens_total": total_tokens,
116
        "estimated_tokens_input": command_tokens,
117
        "estimated_tokens_output": output_tokens,
118
        "estimated_tokens_agent_emitted": command_tokens,
119
        "estimated_tokens_agent_ingested": output_tokens,
120
        "estimated_cost_weighted_tokens": cost_weighted,
121
        "raw_output_bytes": stdout_bytes + stderr_bytes,
122
        "stdout_bytes": stdout_bytes,
123
        "stderr_bytes": stderr_bytes,
124
        "stdout_truncated": stdout_truncated,
125
        "stderr_truncated": stderr_truncated,
126
        "output_truncated": stdout_truncated or stderr_truncated,
127
        "token_estimate": {
128
            "source": source,
129
            "total": total_tokens,
130
            "input": command_tokens,
131
            "output": output_tokens,
132
            "envelope_note": ENVELOPE_NOTE,
133
            "direction_note": DIRECTION_NOTE,
134
            "agent_emitted": command_tokens,
135
            "agent_ingested": output_tokens,
136
            "cost_weighted_total": cost_weighted,
137
            "cost_weights": COST_WEIGHTS_NOTE,
138
            "command_chars": len(command_text),
139
            "stdout_admitted_chars": len(stdout_text),
140
            "stderr_admitted_chars": len(stderr_text),
141
            "stdout_bytes": stdout_bytes,
142
            "stderr_bytes": stderr_bytes,
143
            "stdout_truncated": stdout_truncated,
144
            "stderr_truncated": stderr_truncated,
145
        },
146
    }
147
148
149
def summed_token_fields(rows: list[dict[str, Any]], source: str, command_chars: int | None = None) -> dict[str, Any]:
150
    """Aggregate the flat token fields of sub-rows into one summary row's fields."""
151
    input_tokens = sum(int_or_zero(row.get("estimated_tokens_input")) for row in rows)
152
    output_tokens = sum(int_or_zero(row.get("estimated_tokens_output")) for row in rows)
153
    total_tokens = input_tokens + output_tokens
154
    cost_weighted = cost_weighted_total(input_tokens, output_tokens)
155
    stdout_bytes = sum(int_or_zero(row.get("stdout_bytes")) for row in rows)
156
    stderr_bytes = sum(int_or_zero(row.get("stderr_bytes")) for row in rows)
157
    envelope_emitted = sum(int_or_zero(row.get("estimated_tokens_envelope_emitted")) for row in rows)
158
    envelope_ingested = sum(int_or_zero(row.get("estimated_tokens_envelope_ingested")) for row in rows)
159
    estimate: dict[str, Any] = {
160
        "source": source,
161
        "total": total_tokens,
162
        "input": input_tokens,
163
        "output": output_tokens,
164
        "agent_emitted": input_tokens,
165
        "agent_ingested": output_tokens,
166
        "cost_weighted_total": cost_weighted,
167
        "cost_weights": COST_WEIGHTS_NOTE,
168
    }
169
    if command_chars is not None:
170
        estimate["command_chars"] = command_chars
171
        estimate["stdout_admitted_chars"] = sum(
172
            int_or_zero((row.get("token_estimate") or {}).get("stdout_admitted_chars")) for row in rows
173
        )
174
        estimate["stderr_admitted_chars"] = sum(
175
            int_or_zero((row.get("token_estimate") or {}).get("stderr_admitted_chars")) for row in rows
176
        )
177
        estimate["stdout_bytes"] = stdout_bytes
178
        estimate["stderr_bytes"] = stderr_bytes
179
        estimate["stdout_truncated"] = any(bool((row.get("token_estimate") or {}).get("stdout_truncated")) for row in rows)
180
        estimate["stderr_truncated"] = any(bool((row.get("token_estimate") or {}).get("stderr_truncated")) for row in rows)
181
    return {
182
        "estimated_tokens_total": total_tokens,
183
        "estimated_tokens_input": input_tokens,
184
        "estimated_tokens_output": output_tokens,
185
        "estimated_tokens_agent_emitted": input_tokens,
186
        "estimated_tokens_agent_ingested": output_tokens,
187
        "estimated_cost_weighted_tokens": cost_weighted,
188
        "estimated_tokens_envelope_emitted": envelope_emitted,
189
        "estimated_tokens_envelope_ingested": envelope_ingested,
190
        "estimated_tokens_total_with_envelope": total_tokens + envelope_emitted + envelope_ingested,
191
        "estimated_cost_weighted_tokens_with_envelope": cost_weighted_total(
192
            input_tokens + envelope_emitted, output_tokens + envelope_ingested
193
        ),
194
        "raw_output_bytes": stdout_bytes + stderr_bytes,
195
        "stdout_bytes": stdout_bytes,
196
        "stderr_bytes": stderr_bytes,
197
        "stdout_truncated": any(bool(row.get("stdout_truncated")) for row in rows),
198
        "stderr_truncated": any(bool(row.get("stderr_truncated")) for row in rows),
199
        "output_truncated": any(bool(row.get("output_truncated")) for row in rows),
200
        "token_estimate": estimate,
201
    }