Spaces:
Sleeping
Sleeping
File size: 3,776 Bytes
116524e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | #!/usr/bin/env python3
"""Stress test: RR analyzing 30 real benchmark traces at once.
Usage:
uv run python examples/ace/rr_stress_test.py
"""
import json
import logging
import os
import sys
import time
from pathlib import Path
from dotenv import load_dotenv
_root = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(_root))
load_dotenv(_root / ".env")
from ace.steps.rr_step import RRConfig, RRStep
from ace.core.context import ACEStepContext, SkillbookView
from ace.core.skillbook import Skillbook
MODEL = os.getenv("ACE_MODEL", "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0")
logging.basicConfig(level=logging.INFO, format=" %(name)s | %(message)s")
for name in ("LiteLLM", "litellm", "httpx", "httpcore"):
logging.getLogger(name).setLevel(logging.WARNING)
def load_traces(n: int = 30) -> list[dict]:
"""Load n traces from benchmark results, deduplicating by task."""
traces_path = _root / "ace-eval" / "results" / "benchmarks" / "bench_20260314_154608" / "benchmark" / "traces.json"
data = json.loads(traces_path.read_text())
batch = []
for key, val in data.items():
for trial in val.get("trials", []):
if trial.get("trace"):
t = trial["trace"]
batch.append({
"task_id": f"task_{key}_r{trial.get('reward', '?')}",
"question": t.get("question", ""),
"feedback": t.get("feedback", ""),
"trace": t,
})
if len(batch) >= n:
return batch
return batch
def main():
traces = load_traces(30)
print(f"Model: {MODEL}")
print(f"Loaded {len(traces)} traces")
print(f"Total trace chars: {sum(len(t['trace'].get('reasoning', '')) for t in traces):,}")
print()
rr = RRStep(
MODEL,
config=RRConfig(
max_requests=80,
max_depth=1, # allow recursion
max_tokens=1_500_000,
),
)
ctx = ACEStepContext(
trace={
"question": "Analyze these agent traces from a customer service benchmark. Identify common failure patterns, categorize them, and extract actionable learnings.",
"batch_items": traces,
"item_ids": [t["task_id"] for t in traces],
},
skillbook=SkillbookView(Skillbook()),
)
print("Running RR...")
t0 = time.time()
result_ctx = rr(ctx)
elapsed = time.time() - t0
print(f"\n{'=' * 60}")
print(f" Completed in {elapsed:.1f}s")
print(f" Reflections: {len(result_ctx.reflections)}")
print(f"{'=' * 60}\n")
for i, ref in enumerate(result_ctx.reflections):
print(f"--- Reflection {i} ---")
print(f" Reasoning: {ref.reasoning[:200]}...")
print(f" Key insight: {ref.key_insight[:200] if ref.key_insight else '(none)'}")
if ref.error_identification:
print(f" Error: {ref.error_identification[:200]}")
if ref.root_cause_analysis:
print(f" Root cause: {ref.root_cause_analysis[:200]}")
if ref.correct_approach:
print(f" Correct approach: {ref.correct_approach[:200]}")
raw = ref.raw or {}
if "rr_trace" in raw:
rt = raw["rr_trace"]
print(f" RR trace: depth={rt.get('depth')}, iters={rt.get('total_iterations')}, "
f"compactions={rt.get('compactions')}, timed_out={rt.get('timed_out')}")
if "usage" in raw:
u = raw["usage"]
print(f" Usage: {u.get('total_tokens'):,} tokens, {u.get('requests')} requests")
print()
if __name__ == "__main__":
main()
|