Paradigm Comparison¶

class Paradigm(Enum):
    PURE_RLM = "pure_rlm"
    CODEACT = "codeact"
    TRADITIONAL = "traditional"

@dataclass
class ParadigmResult:
    paradigm: Paradigm
    success: bool
    answer: str

    # Token metrics
    context_tokens: int = 0
    total_tokens: int = 0
    prompt_tokens: int = 0
    completion_tokens: int = 0

    # Cost metrics
    estimated_cost: float = 0.0

    # Time metrics
    duration_seconds: float = 0.0
    iterations: int = 0

    # Quality metrics (if ground truth available)
    accuracy: float | None = None
    f1_score: float | None = None

    # LLM call breakdown
    root_llm_calls: int = 0
    sub_llm_calls: int = 0

    # Event trace
    events: list[dict[str, Any]] = field(default_factory=list)

    # Error info
    error: str | None = None

@dataclass
class ComparisonResult:
    comparison_id: str
    task: str
    context_length: int

    # Results by paradigm
    results: dict[Paradigm, ParadigmResult] = field(default_factory=dict)

    # Timing
    started_at: str = field(...)
    finished_at: str = ""
    total_duration_seconds: float = 0.0

    # Ground truth (if available)
    ground_truth: str | None = None

winner = comparison.get_winner("total_tokens")     # Lower is better
winner = comparison.get_winner("estimated_cost")    # Lower is better
winner = comparison.get_winner("duration_seconds")  # Lower is better
winner = comparison.get_winner("accuracy")          # Higher is better

summary = comparison.get_summary()
# {
#     "comparison_id": "abc12345",
#     "task": "Analyze sentiment...",
#     "context_length": 45230,
#     "paradigms_tested": ["pure_rlm", "codeact", "traditional"],
#     "total_duration_seconds": 45.2,
#     "total_tokens_by_paradigm": {"pure_rlm": 5200, "codeact": 12400, "traditional": 8100},
#     "total_tokens_winner": "pure_rlm",
#     "estimated_cost_by_paradigm": {...},
#     "estimated_cost_winner": "pure_rlm",
#     ...
# }

print(comparison.format_table())

class ParadigmComparator:
    def __init__(
        self,
        runner: Any,           # RLMRunner instance
        event_bus: RLMEventBus | None = None,
    ):

def compare(
    self,
    task: str,
    context: str,
    paradigms: list[Paradigm] | None = None,
    ground_truth: str | None = None,
    max_steps: int = 5,
    exec_timeout: int = 60,
) -> ComparisonResult:

answer_tokens = set(answer.lower().split())
truth_tokens = set(ground_truth.lower().split())
accuracy = len(answer_tokens & truth_tokens) / len(answer_tokens | truth_tokens)

from rlm_code.rlm.comparison import create_comparison_report

report = create_comparison_report(comparison)
print(report)

from rlm_code.rlm.comparison import ParadigmComparator, Paradigm
from rlm_code.rlm.runner import RLMRunner

# Set up runner
runner = RLMRunner(
    llm_connector=my_connector,
    execution_engine=my_engine,
    workdir=my_project_dir,
)

# Create comparator
comparator = ParadigmComparator(runner=runner)

# Load a large context
with open("large_document.txt") as f:
    context = f.read()

# Run comparison
result = comparator.compare(
    task="Summarize the key findings in this document",
    context=context,
    paradigms=[Paradigm.PURE_RLM, Paradigm.CODEACT],
    ground_truth="The document discusses three main findings: ...",
    max_steps=5,
    exec_timeout=120,
)

# Display results
print(result.format_table())

# Get winner
token_winner = result.get_winner("total_tokens")
cost_winner = result.get_winner("estimated_cost")
print(f"Token winner: {token_winner.value}")
print(f"Cost winner: {cost_winner.value}")

# Detailed analysis
from rlm_code.rlm.comparison import create_comparison_report
print(create_comparison_report(result))

# Access individual paradigm results
pure_rlm = result.results[Paradigm.PURE_RLM]
codeact = result.results[Paradigm.CODEACT]
print(f"Pure RLM: {pure_rlm.total_tokens} tokens, ${pure_rlm.estimated_cost:.4f}")
print(f"CodeAct:  {codeact.total_tokens} tokens, ${codeact.estimated_cost:.4f}")
print(f"Token savings: {1 - pure_rlm.total_tokens / codeact.total_tokens:.1%}")

from rlm_code.rlm.events import RLMEventBus, RLMEventType

bus = RLMEventBus()

def on_paradigm_end(event):
    data = event.event_data.metadata
    print(f"  {data['paradigm']}: {data['total_tokens']} tokens, "
          f"{'SUCCESS' if data['success'] else 'FAILED'}")

bus.subscribe_to_type(RLMEventType.COMPARISON_PARADIGM_END, on_paradigm_end)

comparator = ParadigmComparator(runner=runner, event_bus=bus)
result = comparator.compare(task=task, context=context)