This is a gist for running the long mem eval purely on a language model and not using an end to end system like Honcho
Direct LongMemEval Test Runner (No Honcho)
A script that executes longmemeval tests by sending questions directly to an LLM. This script:
- Loads longmemeval test definitions from JSON files
- Formats haystack conversations as context
- Sends the question directly to the model via API
- Judges the response using an LLM
To use
- Set up env:
uv sync
source .venv/bin/activate
- Run this file with a selected test file:
python -m tests.bench.longmem_no_honcho --test-file tests/bench/longmemeval_data/longmemeval_oracle.json
Optional arguments:
--anthropic-api-key: Anthropic API key for response judging (can be set in .env as LLM_ANTHROPIC_API_KEY or provided as an argument)
--batch-size: Number of questions to run concurrently in each batch (default: 10)
--json-output: Path to write JSON summary results for analytics (if not provided, creates timestamped file in tests/bench/eval_results)
import argparse
import asyncio
import json
import logging
import os
import time
from datetime import datetime
from pathlib import Path
from typing import Any, cast
import tiktoken
from anthropic import AsyncAnthropic
from dotenv import load_dotenv
from typing_extensions import TypedDict
from src.utils.clients import honcho_llm_call
from src.utils.types import SupportedProviders
load_dotenv()
class SessionResult(TypedDict):
"""Type definition for session creation results."""
name: str
message_count: int
class QueryResult(TypedDict):
"""Type definition for query execution results."""
question: str
expected_answer: str
actual_response: str
judgment: dict[str, Any]
token_efficiency: dict[str, Any] | None
class TestResult(TypedDict):
"""Type definition for test execution results."""
question_id: str
question_type: str
workspace_id: str
sessions_created: list[SessionResult]
query_executed: QueryResult | None
passed: bool
error: str | None
start_time: float
end_time: float
duration_seconds: float
output_lines: list[str]
class DirectLongMemEvalRunner:
"""
Executes longmemeval JSON tests by sending questions directly to an LLM.
"""
def __init__(
self,
provider: str,
model: str,
anthropic_api_key: str | None = None,
):
"""
Initialize the test runner.
Args:
provider: LLM provider to use
model: Model name to use
anthropic_api_key: Anthropic API key for judging responses
"""
self.provider: SupportedProviders = cast(SupportedProviders, provider)
self.model: str = model
self.anthropic_api_key: str | None = anthropic_api_key
# Configure logging
logging.basicConfig(
level=logging.WARNING, format="%(asctime)s - %(levelname)s - %(message)s"
)
self.logger: logging.Logger = logging.getLogger(__name__)
# Suppress HTTP request logs
logging.getLogger("httpx").setLevel(logging.ERROR)
logging.getLogger("httpcore").setLevel(logging.ERROR)
if self.anthropic_api_key:
self.anthropic_client: AsyncAnthropic = AsyncAnthropic(
api_key=self.anthropic_api_key
)
else:
api_key = os.getenv("LLM_ANTHROPIC_API_KEY")
if not api_key:
raise ValueError("LLM_ANTHROPIC_API_KEY is not set")
self.anthropic_client = AsyncAnthropic(api_key=api_key)
def _format_duration(self, total_seconds: float) -> str:
"""Format a duration in seconds into a human-readable string.
Args:
total_seconds: The duration in seconds.
Returns:
A formatted duration string.
"""
minutes = int(total_seconds // 60)
if minutes > 0:
seconds_rounded = int(round(total_seconds - minutes * 60))
if seconds_rounded == 60:
minutes += 1
seconds_rounded = 0
return f"{minutes}m{seconds_rounded:02d}s"
return f"{total_seconds:.2f}s"
def _calculate_total_tokens(
self, haystack_sessions: list[list[dict[str, str]]]
) -> int:
"""Calculate total tokens from all messages in all sessions.
Args:
haystack_sessions: List of sessions, each containing messages
Returns:
Total number of tokens across all messages
"""
tokenizer = tiktoken.get_encoding("cl100k_base")
total_tokens = 0
for session_messages in haystack_sessions:
for msg in session_messages:
content = msg.get("content", "")
total_tokens += len(tokenizer.encode(content))
return total_tokens
def _format_haystack_as_context(
self,
haystack_sessions: list[list[dict[str, str]]],
haystack_dates: list[str],
) -> str:
"""Format haystack sessions as context for the LLM.
Args:
haystack_sessions: List of sessions, each containing messages
haystack_dates: List of dates corresponding to each session
Returns:
Formatted context string
"""
context_parts: list[str] = []
for session_idx, (session_messages, session_date) in enumerate(
zip(haystack_sessions, haystack_dates, strict=True)
):
context_parts.append(
f"\n--- Conversation {session_idx + 1} ({session_date}) ---\n"
)
for msg in session_messages:
role = msg["role"]
content = msg["content"]
role_label = "User" if role == "user" else "Assistant"
context_parts.append(f"{role_label}: {content}\n")
return "".join(context_parts)
def load_test_file(self, test_file: Path) -> list[dict[str, Any]]:
"""
Load longmemeval test definitions from a JSON file.
Args:
test_file: Path to the JSON test file
Returns:
List of test question dictionaries
"""
with open(test_file) as f:
return json.load(f)
async def judge_response(
self, question: str, expected_answer: str, actual_response: str
) -> dict[str, Any]:
"""
Use an LLM to judge if the actual response matches the expected answer.
Args:
question: The question asked
expected_answer: Expected answer from the test
actual_response: Actual response from the LLM
Returns:
Judgment result with pass/fail and reasoning
"""
try:
system_prompt = """
You are an expert judge evaluating AI responses to memory questions. Your task is to determine if an actual response contains the correct answer from long-term memory.
CRITICAL JUDGING PRINCIPLES:
1. SEMANTIC UNDERSTANDING: Focus on whether the actual response conveys the same core factual information as expected, even if expressed differently
2. FLEXIBLE INTERPRETATION: Accept responses that are longer, more detailed, or use different phrasing as long as they contain the correct answer
3. MEMORY ACCURACY: The key is whether the AI correctly recalled and stated the factual information from memory
4. PARTIAL CREDIT: If the response shows the AI accessed relevant memories but made minor errors in details, consider partial credit
5. IMPLICIT vs EXPLICIT: Accept responses that clearly imply the correct answer through context
ONLY FAIL when:
- The core factual answer is demonstrably wrong
- The response shows no evidence of accessing the relevant memory
- The AI explicitly states incorrect information that contradicts the expected answer
Always respond with valid JSON: {"passed": boolean, "reasoning": "short (1-3 sentences) explanation of why the response is correct or incorrect"}"""
user_prompt = f"""Question: "{question}"
Expected answer: "{expected_answer}"
Actual response: "{actual_response}"
Evaluate whether the actual response correctly answers the question based on the expected answer. Focus on factual accuracy and evidence that the AI accessed the correct memory."""
response = await self.anthropic_client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=300,
temperature=0.0,
system=system_prompt,
messages=[
{
"role": "user",
"content": user_prompt,
}
],
)
if not response.content:
raise ValueError("Anthropic returned empty response")
content_block = response.content[0]
judgment_text = getattr(content_block, "text", None)
if judgment_text is None:
raise ValueError(
f"No text content in response block: {type(content_block)}"
)
# Extract JSON from the response if it's wrapped in markdown
if "```json" in judgment_text:
json_start = judgment_text.find("```json") + 7
json_end = judgment_text.find("```", json_start)
judgment_text = judgment_text[json_start:json_end].strip()
elif "```" in judgment_text:
json_start = judgment_text.find("```") + 3
json_end = judgment_text.find("```", json_start)
judgment_text = judgment_text[json_start:json_end].strip()
judgment = json.loads(judgment_text)
return judgment
except Exception as e:
self.logger.error(f"Error judging response: {e}")
# Fallback to simple string matching
is_correct = expected_answer.lower() in actual_response.lower()
return {
"passed": is_correct,
"reasoning": f"Fallback string matching due to error: {'Match found' if is_correct else 'No match found'}",
}
async def execute_question(self, question_data: dict[str, Any]) -> TestResult:
"""
Execute a single longmemeval question.
Args:
question_data: Dictionary containing question data
Returns:
Test execution results
"""
question_id = question_data["question_id"]
question_type = question_data["question_type"]
question = question_data["question"]
expected_answer = question_data["answer"]
question_date = question_data.get("question_date", "")
question_with_date = (
f"[{question_date}] {question}" if question_date else question
)
output_lines: list[str] = []
output_lines.append(
f"\033[1mExecuting question {question_id} ({question_type})\033[0m"
)
output_lines.append(f"Question: {question_with_date}")
output_lines.append(f"Expected: {expected_answer}")
workspace_id = f"{question_id}_{question_type}"
results: TestResult = {
"question_id": question_id,
"question_type": question_type,
"workspace_id": workspace_id,
"sessions_created": [],
"query_executed": None,
"passed": False,
"error": None,
"start_time": time.time(),
"end_time": 0.0,
"duration_seconds": 0.0,
"output_lines": output_lines,
}
try:
haystack_dates = question_data.get("haystack_dates", [])
haystack_sessions = question_data.get("haystack_sessions", [])
# Validate alignment
if len(haystack_dates) != len(haystack_sessions):
raise ValueError(
f"Misaligned data: {len(haystack_dates)} dates but {len(haystack_sessions)} sessions"
)
haystack_total_messages = sum(len(session) for session in haystack_sessions)
total_available_tokens = self._calculate_total_tokens(haystack_sessions)
print(
f"[{workspace_id}] processing {len(haystack_sessions)} sessions with {haystack_total_messages} total messages ({total_available_tokens} total tokens)"
)
# Track session info for results
for session_idx, session_messages in enumerate(haystack_sessions):
results["sessions_created"].append(
SessionResult(
name=f"session_{session_idx}",
message_count=len(session_messages),
)
)
# Format haystack as context
context = self._format_haystack_as_context(
haystack_sessions, haystack_dates
)
# Build prompt with context and question
full_prompt = f"""You are an AI assistant with access to conversation history. Based on the following past conversations, please answer the question that follows.
{context}
Now, please answer this question based on the conversation history above:
{question_with_date}
Provide a direct, concise answer based on the information from the conversations."""
# Calculate tokens used (approximation)
tokenizer = tiktoken.get_encoding("cl100k_base")
tokens_used = len(tokenizer.encode(full_prompt))
output_lines.append(f"\nAsking question: {question_with_date}")
try:
# Call the LLM directly
response = await honcho_llm_call(
provider=self.provider,
model=self.model,
prompt=full_prompt,
max_tokens=512,
enable_retry=True,
retry_attempts=3,
)
token_efficiency = None
if total_available_tokens > 0:
efficiency_ratio = tokens_used / total_available_tokens
token_efficiency = {
"total_available_tokens": total_available_tokens,
"tokens_used": tokens_used,
"efficiency_ratio": efficiency_ratio,
}
output_lines.append(
f" token efficiency: {efficiency_ratio:.4f} ({tokens_used}/{total_available_tokens} tokens, {efficiency_ratio * 100:.2f}%)"
)
judgment = await self.judge_response(
question_with_date, expected_answer, response.content
)
query_result: QueryResult = {
"question": question_with_date,
"expected_answer": expected_answer,
"actual_response": response.content,
"judgment": judgment,
"token_efficiency": token_efficiency,
}
results["query_executed"] = query_result
results["passed"] = judgment["passed"]
output_lines.append(
" judgment: \033[1m\033[32mPASS\033[0m"
if judgment["passed"]
else " judgment: \033[1m\033[31mFAIL\033[0m"
)
if not judgment["passed"]:
output_lines.append(
f" got response: \033[3m{response.content}\033[0m"
)
output_lines.append(f" expected: {expected_answer}")
output_lines.append(f" reasoning: {judgment['reasoning']}")
except Exception as e:
self.logger.error(f"Error executing question: {e}")
query_result = QueryResult(
question=question_with_date,
expected_answer=expected_answer,
actual_response=f"ERROR: {e}",
judgment={
"passed": False,
"reasoning": f"Question execution failed: {e}",
},
token_efficiency=None,
)
results["query_executed"] = query_result
results["passed"] = False
results["end_time"] = time.time()
results["duration_seconds"] = results["end_time"] - results["start_time"]
output_lines.append(
f"\nQuestion {question_id} completed. Status: {'PASS' if results['passed'] else 'FAIL'} (Duration: {self._format_duration(results['duration_seconds'])})"
)
except Exception as e:
self.logger.error(f"Error executing question {question_id}: {e}")
results["error"] = str(e)
results["passed"] = False
results["end_time"] = time.time()
results["duration_seconds"] = results["end_time"] - results["start_time"]
output_lines.append(f"Error executing question {question_id}: {e}")
return results
async def run_all_questions(
self, test_file: Path, batch_size: int = 10
) -> tuple[list[TestResult], float]:
"""
Run all questions in a longmemeval test file.
Args:
test_file: Path to the longmemeval JSON file
batch_size: Number of questions to run concurrently in each batch
Returns:
Tuple of (list of test results, total duration)
"""
questions = self.load_test_file(test_file)
print(
f"found {len(questions)} {'question' if len(questions) == 1 else 'questions'} in {test_file}"
)
overall_start = time.time()
# Process questions in batches
all_results: list[TestResult] = []
for i in range(0, len(questions), batch_size):
batch = questions[i : i + batch_size]
batch_num = (i // batch_size) + 1
total_batches = (len(questions) + batch_size - 1) // batch_size
print(f"\n{'=' * 60}")
print(
f"Processing batch {batch_num}/{total_batches} ({len(batch)} questions)"
)
print(f"{'=' * 60}")
# Run questions in current batch concurrently
batch_results: list[TestResult] = await asyncio.gather(
*[self.execute_question(q) for q in batch]
)
# Print detailed per-question outputs for this batch
for result in batch_results:
print(f"\n{'=' * 60}")
print("\n".join(result.get("output_lines", [])))
print(f"{'=' * 60}\n")
all_results.extend(batch_results)
overall_end = time.time()
overall_duration = overall_end - overall_start
return all_results, overall_duration
def print_summary(
self, results: list[TestResult], total_elapsed_seconds: float | None = None
) -> None:
"""
Print a summary of all test results.
Args:
results: List of test results
total_elapsed_seconds: Total elapsed time
"""
print(f"\n{'=' * 80}")
print("LONGMEMEVAL TEST EXECUTION SUMMARY (Direct LLM, No Honcho)")
print(f"{'=' * 80}")
total_questions = len(results)
passed_questions = sum(1 for r in results if r.get("passed", False))
failed_questions = total_questions - passed_questions
total_test_time = (
total_elapsed_seconds
if total_elapsed_seconds is not None
else sum(r["duration_seconds"] for r in results)
)
print(f"Total Questions: {total_questions}")
print(f"Passed: {passed_questions}")
print(f"Failed: {failed_questions}")
print(f"Success Rate: {(passed_questions / total_questions) * 100:.1f}%")
print(f"Total Test Time: {self._format_duration(total_test_time)}")
efficiency_ratios: list[float] = []
for result in results:
query = result.get("query_executed")
if query:
token_eff = query.get("token_efficiency")
if token_eff:
efficiency_ratios.append(token_eff["efficiency_ratio"])
if efficiency_ratios:
avg_efficiency = sum(efficiency_ratios) / len(efficiency_ratios)
min_efficiency = min(efficiency_ratios)
max_efficiency = max(efficiency_ratios)
print("\nToken Efficiency:")
print(
f" Average: {avg_efficiency:.4f} ({avg_efficiency * 100:.2f}% of available tokens used)"
)
print(f" Min: {min_efficiency:.4f} ({min_efficiency * 100:.2f}%)")
print(f" Max: {max_efficiency:.4f} ({max_efficiency * 100:.2f}%)")
print("\nDetailed Results:")
print(
f"{'Question ID':<15} {'Type':<20} {'Status':<8} {'Duration':<10} {'Workspace ID':<30}"
)
print(f"{'-' * 15} {'-' * 20} {'-' * 8} {'-' * 10} {'-' * 30}")
for result in results:
question_id = result["question_id"]
question_type = result["question_type"]
status = "PASS" if result.get("passed", False) else "FAIL"
duration = self._format_duration(result["duration_seconds"])
workspace = result["workspace_id"]
print(
f"{question_id:<15} {question_type:<20} {status:<8} {duration:<10} {workspace:<30}"
)
print(f"{'=' * 80}")
def generate_json_summary(
self,
results: list[TestResult],
test_file: Path,
total_elapsed_seconds: float,
output_file: Path | None = None,
) -> None:
"""
Generate a comprehensive JSON summary of test results for analytics.
Args:
results: List of test results
test_file: Path to the test file that was executed
total_elapsed_seconds: Total elapsed time for all tests
output_file: Optional path to write JSON output to
"""
total_questions = len(results)
passed_questions = sum(1 for r in results if r.get("passed", False))
failed_questions = total_questions - passed_questions
# Calculate statistics by question type
type_stats: dict[str, dict[str, int | float]] = {}
for result in results:
q_type = result["question_type"]
if q_type not in type_stats:
type_stats[q_type] = {"total": 0, "passed": 0, "failed": 0}
type_stats[q_type]["total"] += 1
if result.get("passed", False):
type_stats[q_type]["passed"] += 1
else:
type_stats[q_type]["failed"] += 1
# Add success rates to type stats
for q_type in type_stats:
stats = type_stats[q_type]
stats["success_rate"] = (
(stats["passed"] / stats["total"]) * 100 if stats["total"] > 0 else 0
)
# Calculate timing statistics
durations = [r["duration_seconds"] for r in results]
timing_stats = {
"total_duration_seconds": total_elapsed_seconds,
"individual_test_durations": {
"min_seconds": min(durations) if durations else 0,
"max_seconds": max(durations) if durations else 0,
"mean_seconds": sum(durations) / len(durations) if durations else 0,
"median_seconds": sorted(durations)[len(durations) // 2]
if durations
else 0,
},
}
# Calculate token efficiency statistics
efficiency_ratios: list[float] = []
total_available_tokens_list: list[int] = []
tokens_used_list: list[int] = []
for result in results:
query = result.get("query_executed")
if query:
eff = query.get("token_efficiency")
if eff:
efficiency_ratios.append(eff["efficiency_ratio"])
total_available_tokens_list.append(eff["total_available_tokens"])
tokens_used_list.append(eff["tokens_used"])
token_efficiency_stats = None
if efficiency_ratios:
token_efficiency_stats = {
"mean_efficiency_ratio": sum(efficiency_ratios)
/ len(efficiency_ratios),
"min_efficiency_ratio": min(efficiency_ratios),
"max_efficiency_ratio": max(efficiency_ratios),
"median_efficiency_ratio": sorted(efficiency_ratios)[
len(efficiency_ratios) // 2
],
"mean_tokens_available": sum(total_available_tokens_list)
/ len(total_available_tokens_list),
"mean_tokens_used": sum(tokens_used_list) / len(tokens_used_list),
"total_questions_with_metrics": len(efficiency_ratios),
}
# Create the full summary
summary = {
"metadata": {
"test_file": str(test_file),
"execution_timestamp": datetime.now().isoformat(),
"runner_version": "1.0.0-no-honcho",
"provider": self.provider,
"model": self.model,
},
"summary_statistics": {
"total_questions": total_questions,
"passed": passed_questions,
"failed": failed_questions,
"success_rate_percent": (passed_questions / total_questions) * 100
if total_questions > 0
else 0,
"statistics_by_type": type_stats,
},
"timing": timing_stats,
"token_efficiency": token_efficiency_stats,
"detailed_results": [
{
"question_id": result["question_id"],
"question_type": result["question_type"],
"workspace_id": result["workspace_id"],
"passed": result.get("passed", False),
"duration_seconds": result["duration_seconds"],
"start_time": result["start_time"],
"end_time": result["end_time"],
"error": result.get("error"),
"query_executed": result.get("query_executed"),
}
for result in results
],
}
if output_file:
output_file.parent.mkdir(parents=True, exist_ok=True)
with open(output_file, "w") as f:
json.dump(summary, f, indent=2, default=str)
print(f"\nJSON summary written to: {output_file}")
async def main() -> int:
"""
Main entry point for the direct longmemeval test runner.
"""
parser = argparse.ArgumentParser(
description="Run longmemeval tests directly against an LLM (no Honcho)",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
%(prog)s --test-file tests/bench/longmemeval_data/longmemeval_s.json # Run longmemeval tests
""",
)
parser.add_argument(
"--test-file",
type=Path,
required=True,
help="Path to longmemeval JSON file (required)",
)
parser.add_argument(
"--anthropic-api-key",
type=str,
help="Anthropic API key for response judging (optional)",
)
parser.add_argument(
"--batch-size",
type=int,
default=10,
help="Number of questions to run concurrently in each batch (default: 10)",
)
parser.add_argument(
"--json-output",
type=Path,
help="Path to write JSON summary results for analytics (optional)",
)
args = parser.parse_args()
# Validate arguments
if not args.test_file.exists():
print(f"Error: Test file {args.test_file} does not exist")
return 1
if args.batch_size <= 0:
print(f"Error: Batch size must be positive, got {args.batch_size}")
return 1
# Get provider and model from environment
provider = os.getenv("DIALECTIC_PROVIDER", "NOT_GIVEN")
model = os.getenv("DIALECTIC_MODEL", "NOT_GIVEN")
print(f"Using provider: {provider}, model: {model}")
# Create test runner
runner = DirectLongMemEvalRunner(
provider=provider,
model=model,
anthropic_api_key=args.anthropic_api_key,
)
try:
# Run all questions
results, total_elapsed = await runner.run_all_questions(
args.test_file, args.batch_size
)
runner.print_summary(results, total_elapsed_seconds=total_elapsed)
# Generate JSON output if requested
if args.json_output:
runner.generate_json_summary(
results, args.test_file, total_elapsed, args.json_output
)
else:
# Always generate a default JSON output file with timestamp
default_output = Path(
f"tests/bench/eval_results/longmemeval_no_honcho_results_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
)
runner.generate_json_summary(
results, args.test_file, total_elapsed, default_output
)
# Return exit code based on results
all_passed = all(r.get("passed", False) for r in results)
return 0 if all_passed else 1
except KeyboardInterrupt:
print("\nTest execution interrupted by user")
return 1
except Exception as e:
print(f"Error running tests: {e}")
return 1
if __name__ == "__main__":
exit_code = asyncio.run(main())
exit(exit_code)