RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Improve prefix cache eval harness

This commit is contained in:
adavyas 2026-03-13 22:33:23 -07:00
parent 1b99469335
commit e86cbac75b
1 changed files with 177 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

327
scripts/compare_prefix_cache.py
View File

@ -2,23 +2,14 @@
"""
Compare prompt-prefix cache behavior between two Honcho worktrees.
This is intentionally a narrow probe. It does not start Honcho servers, touch the
database, or exercise Hermes/session state. Instead, it imports each worktree's
`src.utils.clients.honcho_llm_call_inner` and runs a few controlled message
patterns against live providers.
This is an eval-only probe. It does not start Honcho servers, touch the database,
or exercise Hermes/session state. Instead, it imports each worktree's
`src.utils.clients.honcho_llm_call_inner` and runs controlled message patterns
against live providers.
The important scenario is `change_history`: the first system block stays stable
while the second rolling system block changes. The candidate branch should retain
more cache reuse there because it preserves multiple cacheable system blocks
instead of flattening them into one blob.
Example:
uv run python scripts/compare_prefix_cache.py \
--baseline-worktree /path/to/honcho-main \
--candidate-worktree /path/to/honcho-branch \
--provider anthropic-haiku=anthropic:claude-haiku-4-5 \
--provider openrouter-haiku=custom:anthropic/claude-haiku-4.5 \
--provider openai-mini=openai:gpt-4.1-mini
The key scenario is `change_history`: the stable base prefix stays the same while
the rolling context block changes. The candidate branch should preserve more
cache reuse there because it keeps multiple cacheable system blocks separate.
"""
from __future__ import annotations
@ -43,41 +34,37 @@ from src.utils.clients import honcho_llm_call_inner
async def run() -> None:
results = []
for scenario in payload["scenarios"]:
calls = []
for call in scenario["calls"]:
start = time.perf_counter()
response = await honcho_llm_call_inner(
provider=payload["provider"],
model=payload["model"],
prompt="",
max_tokens=payload["max_tokens"],
temperature=0,
messages=call["messages"],
)
elapsed_ms = (time.perf_counter() - start) * 1000
calls.append(
{
"label": call["label"],
"duration_ms": elapsed_ms,
"input_tokens": response.input_tokens,
"output_tokens": response.output_tokens,
"cache_creation_input_tokens": response.cache_creation_input_tokens,
"cache_read_input_tokens": response.cache_read_input_tokens,
"finish_reasons": response.finish_reasons,
"content_preview": (response.content or "")[:120],
}
)
results.append({"name": scenario["name"], "calls": calls})
calls = []
for call in payload["calls"]:
start = time.perf_counter()
response = await honcho_llm_call_inner(
provider=payload["provider"],
model=payload["model"],
prompt="",
max_tokens=payload["max_tokens"],
temperature=0,
messages=call["messages"],
)
elapsed_ms = (time.perf_counter() - start) * 1000
calls.append(
{
"label": call["label"],
"duration_ms": elapsed_ms,
"input_tokens": response.input_tokens,
"output_tokens": response.output_tokens,
"cache_creation_input_tokens": response.cache_creation_input_tokens,
"cache_read_input_tokens": response.cache_read_input_tokens,
"finish_reasons": response.finish_reasons,
"content_preview": (response.content or "")[:120],
}
)
print(json.dumps({"scenarios": results}))
print(json.dumps({"name": payload["name"], "calls": calls}))
asyncio.run(run())
"""
BASE_PREFIX = "\n".join(
[
"You are Honcho's memory-backed reasoning layer.",
@ -198,10 +185,16 @@ def parse_provider_spec(raw: str) -> ProviderSpec:
return ProviderSpec(label=label, provider=provider, model=model)
def build_messages(base_prefix: str, rolling_history: str, user_query: str) -> list[dict[str, str]]:
def add_namespace(namespace: str, content: str) -> str:
return f"<cache_namespace>{namespace}</cache_namespace>\n{content}"
def build_messages(
namespace: str, base_prefix: str, rolling_history: str, user_query: str
) -> list[dict[str, str]]:
return [
{"role": "system", "content": base_prefix},
{"role": "system", "content": rolling_history},
{"role": "system", "content": add_namespace(namespace, base_prefix)},
{"role": "system", "content": add_namespace(namespace, rolling_history)},
{"role": "user", "content": user_query},
]
@ -210,87 +203,75 @@ def build_scenarios(selected: set[str] | None) -> list[dict[str, Any]]:
scenario_defs = [
{
"name": "repeat_exact",
"calls": [
{
"label": "cold",
"messages": build_messages(
BASE_PREFIX,
ROLLING_HISTORY_A,
"What is the user's preferred morning drink schedule?",
),
},
{
"label": "warm_same",
"messages": build_messages(
BASE_PREFIX,
ROLLING_HISTORY_A,
"What is the user's preferred morning drink schedule?",
),
},
],
"prime": {
"base_prefix": BASE_PREFIX,
"rolling_history": ROLLING_HISTORY_A,
"user_query": "What is the user's preferred morning drink schedule?",
},
"transition": {
"base_prefix": BASE_PREFIX,
"rolling_history": ROLLING_HISTORY_A,
"user_query": "What is the user's preferred morning drink schedule?",
},
"steady": {
"base_prefix": BASE_PREFIX,
"rolling_history": ROLLING_HISTORY_A,
"user_query": "What is the user's preferred morning drink schedule?",
},
},
{
"name": "change_user",
"calls": [
{
"label": "cold",
"messages": build_messages(
BASE_PREFIX,
ROLLING_HISTORY_A,
"What is the user's preferred morning drink schedule?",
),
},
{
"label": "warm_user_changed",
"messages": build_messages(
BASE_PREFIX,
ROLLING_HISTORY_A,
"What exact launch date should be remembered for the user?",
),
},
],
"prime": {
"base_prefix": BASE_PREFIX,
"rolling_history": ROLLING_HISTORY_A,
"user_query": "What is the user's preferred morning drink schedule?",
},
"transition": {
"base_prefix": BASE_PREFIX,
"rolling_history": ROLLING_HISTORY_A,
"user_query": "What exact launch date should be remembered for the user?",
},
"steady": {
"base_prefix": BASE_PREFIX,
"rolling_history": ROLLING_HISTORY_A,
"user_query": "What exact launch date should be remembered for the user?",
},
},
{
"name": "change_history",
"calls": [
{
"label": "cold",
"messages": build_messages(
BASE_PREFIX,
ROLLING_HISTORY_A,
"Summarize the user's communication preference in one sentence.",
),
},
{
"label": "warm_history_changed",
"messages": build_messages(
BASE_PREFIX,
ROLLING_HISTORY_B,
"Summarize the user's communication preference in one sentence.",
),
},
],
"prime": {
"base_prefix": BASE_PREFIX,
"rolling_history": ROLLING_HISTORY_A,
"user_query": "Summarize the user's communication preference in one sentence.",
},
"transition": {
"base_prefix": BASE_PREFIX,
"rolling_history": ROLLING_HISTORY_B,
"user_query": "Summarize the user's communication preference in one sentence.",
},
"steady": {
"base_prefix": BASE_PREFIX,
"rolling_history": ROLLING_HISTORY_B,
"user_query": "Summarize the user's communication preference in one sentence.",
},
},
{
"name": "change_base",
"calls": [
{
"label": "cold",
"messages": build_messages(
BASE_PREFIX,
ROLLING_HISTORY_A,
"What city is the user considering for a move?",
),
},
{
"label": "warm_base_changed",
"messages": build_messages(
BASE_PREFIX_VARIANT,
ROLLING_HISTORY_A,
"What city is the user considering for a move?",
),
},
],
"prime": {
"base_prefix": BASE_PREFIX,
"rolling_history": ROLLING_HISTORY_A,
"user_query": "What city is the user considering for a move?",
},
"transition": {
"base_prefix": BASE_PREFIX_VARIANT,
"rolling_history": ROLLING_HISTORY_A,
"user_query": "What city is the user considering for a move?",
},
"steady": {
"base_prefix": BASE_PREFIX_VARIANT,
"rolling_history": ROLLING_HISTORY_A,
"user_query": "What city is the user considering for a move?",
},
},
]
if not selected:
@ -298,18 +279,40 @@ def build_scenarios(selected: set[str] | None) -> list[dict[str, Any]]:
return [scenario for scenario in scenario_defs if scenario["name"] in selected]
def run_probe(
variant: VariantSpec,
provider: ProviderSpec,
scenarios: list[dict[str, Any]],
max_tokens: int,
def build_scenario_payload(
variant: VariantSpec, provider: ProviderSpec, scenario: dict[str, Any], max_tokens: int
) -> dict[str, Any]:
payload = {
namespace = f"{variant.label}:{provider.label}:{scenario['name']}"
calls = []
for label in ("prime", "transition", "steady"):
call = scenario[label]
calls.append(
{
"label": label,
"messages": build_messages(
namespace,
call["base_prefix"],
call["rolling_history"],
call["user_query"],
),
}
)
return {
"name": scenario["name"],
"provider": provider.provider,
"model": provider.model,
"max_tokens": max_tokens,
"scenarios": scenarios,
"calls": calls,
}
def run_scenario_probe(
variant: VariantSpec,
provider: ProviderSpec,
scenario: dict[str, Any],
max_tokens: int,
) -> dict[str, Any]:
payload = build_scenario_payload(variant, provider, scenario, max_tokens)
process = subprocess.run(
[sys.executable, "-c", CHILD_CODE, json.dumps(payload)],
cwd=variant.worktree,
@ -319,19 +322,40 @@ def run_probe(
)
if process.returncode != 0:
raise RuntimeError(
f"{variant.label} probe failed for {provider.label}.\n"
f"{variant.label} probe failed for {provider.label} ({scenario['name']}).\n"
f"stdout:\n{process.stdout}\n"
f"stderr:\n{process.stderr}"
)
return json.loads(process.stdout)
def run_probe(
variant: VariantSpec,
provider: ProviderSpec,
scenarios: list[dict[str, Any]],
max_tokens: int,
) -> dict[str, Any]:
return {
"scenarios": [
run_scenario_probe(variant, provider, scenario, max_tokens)
for scenario in scenarios
]
}
def format_metric(value: Any) -> str:
if isinstance(value, float):
return f"{value:.2f}"
return str(value)
def cache_ratio_pct(call: dict[str, Any]) -> float:
input_tokens = call["input_tokens"] or 0
if input_tokens <= 0:
return 0.0
return (call["cache_read_input_tokens"] / input_tokens) * 100
def print_variant_result(variant: VariantSpec, result: dict[str, Any]) -> None:
print(f" {variant.label}")
for scenario in result["scenarios"]:
@ -339,10 +363,11 @@ def print_variant_result(variant: VariantSpec, result: dict[str, Any]) -> None:
for call in scenario["calls"]:
print(
" "
f"{call['label']:<18} "
f"{call['label']:<12} "
f"read={format_metric(call['cache_read_input_tokens']):>8} "
f"create={format_metric(call['cache_creation_input_tokens']):>8} "
f"input={format_metric(call['input_tokens']):>8} "
f"cached_pct={cache_ratio_pct(call):>7.1f} "
f"ms={format_metric(call['duration_ms']):>8}"
)
@ -355,27 +380,29 @@ def print_delta_summary(
candidate_by_name = {scenario["name"]: scenario for scenario in candidate["scenarios"]}
print(" delta summary (candidate - baseline)")
for name in baseline_by_name:
base_calls = baseline_by_name[name]["calls"]
cand_calls = candidate_by_name[name]["calls"]
if len(base_calls) < 2 or len(cand_calls) < 2:
continue
base_warm = base_calls[1]
cand_warm = cand_calls[1]
read_delta = (
cand_warm["cache_read_input_tokens"] - base_warm["cache_read_input_tokens"]
)
create_delta = (
cand_warm["cache_creation_input_tokens"]
- base_warm["cache_creation_input_tokens"]
)
latency_delta = cand_warm["duration_ms"] - base_warm["duration_ms"]
print(
" "
f"{name:<16} "
f"read_delta={read_delta:+8.2f} "
f"create_delta={create_delta:+8.2f} "
f"warm_latency_delta_ms={latency_delta:+8.2f}"
)
base_calls = {call["label"]: call for call in baseline_by_name[name]["calls"]}
cand_calls = {call["label"]: call for call in candidate_by_name[name]["calls"]}
for label in ("transition", "steady"):
base_call = base_calls[label]
cand_call = cand_calls[label]
read_delta = (
cand_call["cache_read_input_tokens"]
- base_call["cache_read_input_tokens"]
)
create_delta = (
cand_call["cache_creation_input_tokens"]
- base_call["cache_creation_input_tokens"]
)
ratio_delta = cache_ratio_pct(cand_call) - cache_ratio_pct(base_call)
latency_delta = cand_call["duration_ms"] - base_call["duration_ms"]
print(
" "
f"{name}:{label:<11} "
f"read_delta={read_delta:+8.2f} "
f"create_delta={create_delta:+8.2f} "
f"cached_pct_delta={ratio_delta:+7.1f} "
f"latency_delta_ms={latency_delta:+8.2f}"
)
def main() -> None: