Manual Trace System¶
This guide explains how to use the Manual Trace System to evaluate pre-generated LLM responses without making live API calls during verification.
Overview¶
The Manual Trace System enables you to provide pre-generated answer traces directly to the Karenina verification engine, bypassing the real-time LLM answer generation step. This feature is designed for standalone backend usage, allowing you to:
- Evaluate pre-recorded LLM responses (from previous experiments, other systems, or manual collection)
- Compare different answer generation approaches without re-running models
- Test verification/rubric systems with controlled answers
- Integrate external LLM outputs into Karenina's evaluation framework
The system supports both simple string traces and rich LangChain message lists with automatic tool call metrics extraction.
Key Capabilities¶
- Programmatic Trace Management:
ManualTracesclass for managing traces tied to benchmarks - Flexible Registration: Register traces by question hash (MD5) or question text with automatic mapping
- Dual Format Support:
- Simple string traces (plain text answers)
- LangChain message lists (AIMessage, ToolMessage, etc.) with automatic preprocessing
- Agent Metrics Extraction: Automatic tool call counting, failure detection from message lists
- Batch Registration: Efficient bulk trace registration with
register_traces() - Post-Config Population: Populate traces after
ModelConfigcreation for flexible workflows - Preset Compatibility: Manual configs work with preset system (traces excluded from serialization)
- Session-Based Storage: Thread-safe, time-bounded trace storage with automatic cleanup
- Backward Compatible: Maintains compatibility with existing GUI-based manual trace upload
from karenina.adapters.manual import ManualTraces
from karenina.benchmark import Benchmark
from karenina.schemas import ModelConfig, VerificationConfig
# Note: Using finished=True instead of answer_template for notebook examples
# This avoids source code extraction issues with dynamically defined classes
# Create benchmark
benchmark = Benchmark("my_experiment")
benchmark.add_question(question="What is 2+2?", raw_answer="4", finished=True) # Question marked as finished
# Initialize manual traces
manual_traces = ManualTraces(benchmark)
# Register trace by question text
manual_traces.register_trace("What is 2+2?", "The answer is 4. I computed this by adding 2 and 2.", map_to_id=True)
print(f"Registered {len(manual_traces._benchmark._questions_cache)} traces")
print(f"Question text maps to hash: {list(manual_traces._benchmark._questions_cache.keys())[0][:32]}...")
# Create manual config
manual_config = ModelConfig(interface="manual", manual_traces=manual_traces)
# Create judge config
judge_config = ModelConfig(
id="gpt-4.1-mini",
model_provider="openai",
model_name="gpt-4.1-mini",
temperature=0.0,
interface="langchain",
system_prompt="You are an expert judge.",
)
print(f"Manual config ID: {manual_config.id}")
print(f"Manual config interface: {manual_config.interface}")
print(f"Judge config: {judge_config.id}")
# Create verification config
config = VerificationConfig(answering_models=[manual_config], parsing_models=[judge_config])
# Run verification
results = benchmark.run_verification(config)
print("Verification complete!")
print(f"Results: {len(results.results)} question(s) evaluated")
print(f"Passed: {sum(1 for r in results.results if r.template.verify_result)}")
Architecture¶
Core Components¶
1. ManualTraces Class¶
Location: karenina/src/karenina/adapters/manual/traces.py
Purpose: High-level API for managing manual traces for a specific benchmark
Key Methods:
__init__(benchmark)- Initialize with benchmark for question mappingregister_trace(question_identifier, trace, map_to_id=False)- Register single traceregister_traces(traces_dict, map_to_id=False)- Batch register traces_question_text_to_hash(question_text)- Convert text to MD5 hash with validation_preprocess_trace(trace)- Handle both string and LangChain message formats
2. ManualTraceManager Class¶
Location: karenina/src/karenina/adapters/manual/manager.py
Purpose: Session-based thread-safe storage for manual traces
Key Features:
- Thread-safe storage with
threading.RLock() - Session timeout (default: 1 hour) with automatic cleanup
- Storage for both traces and agent metrics
- MD5 hash validation
3. ModelConfig Integration¶
New Field: manual_traces: Any = Field(default=None, exclude=True)
Validation:
- Enforces
manual_tracesrequirement for manual interface - Auto-sets
id="manual"andmodel_name="manual"for manual interface - Validates that MCP tools are not used with manual interface
from karenina.adapters.manual import ManualTraces
from karenina.benchmark import Benchmark
from karenina.schemas import ModelConfig, VerificationConfig
# Create benchmark
benchmark = Benchmark("simple_example")
benchmark.add_question(question="What is 2+2?", raw_answer="4", finished=True)
benchmark.add_question(question="What is 3+3?", raw_answer="6", finished=True)
# Initialize manual traces
manual_traces = ManualTraces(benchmark)
# Register traces by question text
manual_traces.register_trace("What is 2+2?", "The answer is 4. I added 2 and 2 to get 4.", map_to_id=True)
manual_traces.register_trace("What is 3+3?", "The answer is 6. I added 3 and 3 to get 6.", map_to_id=True)
print(f"Registered {len(list(benchmark.get_finished_questions()))} traces")
for q in benchmark.get_finished_questions(ids_only=False):
print(f" - {q['question'][:30]}...")
# Create manual config
manual_config = ModelConfig(interface="manual", manual_traces=manual_traces)
# Create judge config
judge_config = ModelConfig(
id="gpt-4.1-mini", model_provider="openai", model_name="gpt-4.1-mini", temperature=0.0, interface="langchain"
)
# Run verification
config = VerificationConfig(answering_models=[manual_config], parsing_models=[judge_config])
results = benchmark.run_verification(config)
print(f"Evaluated {len(results.results)} questions")
Workflow 2: LangChain Message Lists with Tool Calls¶
When you use LangChain message lists (with AIMessage, ToolMessage, etc.), the system automatically extracts agent metrics like tool call counts and failures.
from langchain_core.messages import AIMessage, ToolMessage
from karenina.adapters.manual import ManualTraces
# Create a new benchmark for this example
benchmark = Benchmark("agent_workflow")
benchmark.add_question(question="What is 6 times 7?", raw_answer="42", finished=True)
# Initialize manual traces
manual_traces = ManualTraces(benchmark)
# Register trace with tool calls
messages = [
AIMessage(content="I need to calculate this"),
ToolMessage(name="calculator", content="Result: 42", tool_call_id="call_calc_001"),
ToolMessage(name="validator", content="Validation passed", tool_call_id="call_valid_002"),
AIMessage(content="The answer is 42. I verified this using a calculator and validator."),
]
manual_traces.register_trace("What is 6 times 7?", messages, map_to_id=True)
print("Registered trace with tool calls")
print("Agent metrics automatically extracted:")
print(" - tool_calls: 2")
print(" - unique_tools_used: 2 (calculator, validator)")
print(" - iterations: 1")
Workflow 3: Batch Registration¶
For efficiency, you can register multiple traces at once using a dictionary.
from langchain_core.messages import AIMessage, ToolMessage
# Create benchmark
benchmark = Benchmark("batch_example")
benchmark.add_question(question="Question 1?", raw_answer="Answer 1", finished=True)
benchmark.add_question(question="Question 2?", raw_answer="Answer 2", finished=True)
benchmark.add_question(question="Question 3?", raw_answer="Answer 3", finished=True)
# Initialize manual traces
manual_traces = ManualTraces(benchmark)
# Prepare traces dictionary
traces = {
"Question 1?": "Answer 1 with explanation.",
"Question 2?": [
AIMessage(content="Thinking..."),
ToolMessage(name="tool", content="data", tool_call_id="call_1"),
AIMessage(content="Answer 2 with context."),
],
"Question 3?": "Answer 3 with details.",
}
# Batch register all at once
manual_traces.register_traces(traces, map_to_id=True)
print(f"Batch registered {len(traces)} traces")
print("All traces now available for verification")
Workflow 4: Register by Question Hash¶
You can also register traces using the MD5 hash directly instead of question text.
# Create benchmark
benchmark = Benchmark("hash_example")
benchmark.add_question(question="What is 2+2?", raw_answer="4", finished=True)
# Initialize manual traces
manual_traces = ManualTraces(benchmark)
# Compute hash manually or get from CSV mapper export
question_hash = hashlib.md5(b"What is 2+2?").hexdigest()
print(f"Question hash: {question_hash}")
# Register by hash (map_to_id=False, default)
manual_traces.register_trace(question_hash, "The answer is 4.", map_to_id=False)
print("Registered trace using hash directly")
Workflow 5: Populate Traces After Config Creation¶
You can create the config structure first and populate traces later, enabling flexible workflows.
# Create benchmark
benchmark = Benchmark("delayed_population")
benchmark.add_question(question="What is 2+2?", raw_answer="4", finished=True)
# 1. Create ManualTraces and ModelConfig upfront
manual_traces = ManualTraces(benchmark)
manual_config = ModelConfig(interface="manual", manual_traces=manual_traces)
print("Created config with empty traces")
# 2. Later, populate traces (e.g., from file, database, API)
def load_traces_from_source():
return {"What is 2+2?": "The answer is 4."}
for question_text, trace_content in load_traces_from_source().items():
manual_traces.register_trace(question_text, trace_content, map_to_id=True)
print("Populated traces from external source")
# 3. Run verification with populated traces
judge_config = ModelConfig(id="gpt-4.1-mini", model_provider="openai", model_name="gpt-4.1-mini", interface="langchain")
config = VerificationConfig(answering_models=[manual_config], parsing_models=[judge_config])
results = benchmark.run_verification(config)
print(f"Verification complete: {len(results.results)} results")
Implementation Details¶
Question Mapping¶
Hash Generation:
- Uses MD5 hash of UTF-8 encoded question text:
hashlib.md5(question_text.encode("utf-8")).hexdigest() - Same algorithm as
Question.idproperty inschemas/domain/question.py - Results in 32-character hexadecimal string
Validation:
- When
map_to_id=True, question text is searched in benchmark's_questions_cache - Raises
ValueErrorif question not found, with computed hash and available count - Exact text matching (case-sensitive, including whitespace)
Trace Format Processing¶
String Traces:
- Stored as-is with no preprocessing
- No agent metrics extracted (
metrics = None) - Simplest format for basic answer evaluation
LangChain Message Lists:
- Validation: Must be list of
BaseMessageobjects (AIMessage, ToolMessage, etc.)- Calls
extract_agent_metrics_from_messages(messages)fromtrace_agent_metrics.py - Calls
_extract_agent_metrics(response)fromverification_utils.py - Extracts: tool calls, tool failures, iterations
- Calls
- Harmonization:
- Calls
harmonize_agent_response(response)fromadapters/langchain/trace.py - Converts message list to unified string format
- Calls
- Storage: Both harmonized trace and metrics stored together
Agent Metrics Structure¶
When using LangChain message lists, the following metrics are automatically extracted:
# Example agent metrics structure
example_metrics = {
"tool_calls": 3, # Number of tool invocations
"unique_tools_used": 2, # Number of unique tools
"failed_tool_calls": 0, # Number of failed invocations
"iterations": 1, # Agent iterations
}
print("Agent Metrics Structure:")
for key, value in example_metrics.items():
print(f" {key}: {value}")
Session-Based Storage¶
Design:
- Global singleton
ManualTraceManagerinstance - Thread-safe with
threading.RLock() - Session timeout: 1 hour (3600 seconds)
- Automatic cleanup of expired traces
Cleanup Strategy:
- Timer-Based:
threading.Timertriggers cleanup after timeout - Activity-Based: Timer resets on any trace access
- Trace-Level: Individual traces have timestamps, expired traces removed
- Session-Level: If no activity for timeout period, entire session clears
ModelConfig Validation¶
Requirements for Manual Interface:
interfacemust be"manual"manual_tracesmust not beNone(raisesValueErrorif missing)iddefaults to"manual"if not providedmodel_namedefaults to"manual"if not providedmcp_urls_dictmust beNone(raisesValueErrorif MCP configured)
Preset Compatibility:
manual_tracesfield marked withField(exclude=True)- Automatically excluded from Pydantic serialization
- Presets save config structure but not trace data
- Traces must be re-populated when loading preset
Best Practices¶
1. Question Text Matching¶
Do:
- Use exact question text from benchmark (case-sensitive, including whitespace)
- Use
map_to_id=Truewhen working with question text - Verify question text matches benchmark before registration
Don't:
- Modify question text (trim whitespace, change case, etc.)
- Assume approximate matching will work
- Register traces for questions not in benchmark
Tip: Export CSV mapper from benchmark to see exact question text and hashes
2. Trace Format Selection¶
Use String Traces When:
- Answers are simple text without tool calls
- No agent metrics needed
- Simplest workflow sufficient
Use LangChain Message Lists When:
- Preserving tool call history is important
- Agent metrics (tool calls, failures) are valuable
- Comparing agent-based vs. non-agent-based approaches
- Debugging tool usage patterns
3. Error Handling¶
Common Errors:
- Question Not Found:
ValueError: Question not found in benchmark: 'What is 2+2?...'
Computed hash: 936dbc8755f623c951d96ea2b03e13bc
Fix: Verify exact question text matches benchmark, check for whitespace/case differences
- Invalid Hash Format:
ManualTraceError: Invalid question hash format: 'short'
Fix: Ensure hash is 32-character hexadecimal MD5
- Missing Manual Traces:
ValueError: manual_traces is required when interface='manual'
Fix: Pass manual_traces to ModelConfig constructor
- MCP Configuration Conflict:
ValueError: MCP tools are not supported with manual interface
Fix: Remove mcp_urls_dict from manual ModelConfig
4. Performance Optimization¶
Batch Registration:
- Use
register_traces()instead of multipleregister_trace()calls - Reduces overhead for large trace sets
- More readable code
Memory Management:
- Monitor trace count with
get_manual_trace_count() - Check memory usage with
get_memory_usage_info() - Clear traces with
clear_manual_traces()when done
Session Cleanup:
- Traces auto-expire after 1 hour of inactivity
- Manual cleanup with
clear_manual_traces()if needed - Activity resets timeout (any trace access)
5. Testing and Validation¶
Before Running Verification, verify traces were registered correctly:
import hashlib
from karenina.adapters.manual import ManualTraces, get_manual_trace, get_manual_trace_count, has_manual_trace
# Create benchmark and register a trace
benchmark = Benchmark("validation_example")
benchmark.add_question(question="What is 2+2?", raw_answer="4", finished=True)
manual_traces = ManualTraces(benchmark)
manual_traces.register_trace("What is 2+2?", "The answer is 4.", map_to_id=True)
# Verify trace was registered
question_hash = hashlib.md5(b"What is 2+2?").hexdigest()
print(f"Question hash: {question_hash}")
print(f"Has trace: {has_manual_trace(question_hash)}")
# Retrieve and inspect trace
trace = get_manual_trace(question_hash)
print(f"Registered trace: {trace[:50]}...")
# Check trace count
expected_count = len(list(benchmark.get_finished_questions()))
actual_count = get_manual_trace_count()
print(f"Expected traces: {expected_count}")
print(f"Actual traces: {actual_count}")
if actual_count != expected_count:
print(f"Warning: Expected {expected_count} traces, have {actual_count}")
else:
print("All traces registered correctly!")
6. Preset Workflow¶
Saving Presets with Manual Configs:
# Example: Saving preset with manual config
benchmark = Benchmark("preset_example")
benchmark.add_question(question="What is 2+2?", raw_answer="4", finished=True)
manual_traces = ManualTraces(benchmark)
manual_traces.register_trace("What is 2+2?", "The answer is 4.", map_to_id=True)
manual_config = ModelConfig(interface="manual", manual_traces=manual_traces)
judge_config = ModelConfig(id="gpt-4.1-mini", model_provider="openai", model_name="gpt-4.1-mini", interface="langchain")
config = VerificationConfig(answering_models=[manual_config], parsing_models=[judge_config])
# Manual traces excluded from serialization
config_dict = config.model_dump()
print("manual_traces in config dict:", "manual_traces" in config_dict)
print("Preset can be saved without trace data")
# Example: Loading preset and re-populating manual traces
# 1. Load preset (manual_traces will be None)
# loaded_config = preset_service.load_preset("my_preset")
# 2. Re-populate manual traces
manual_traces = ManualTraces(benchmark)
traces_dict = {"What is 2+2?": "The answer is 4."}
manual_traces.register_traces(traces_dict, map_to_id=True)
# 3. Update config with traces
# loaded_config.answering_models[0].manual_traces = manual_traces
# 4. Run verification
# benchmark.run_verification(loaded_config)
print("When loading presets with manual configs:")
print("1. Load preset (manual_traces will be None)")
print("2. Re-populate manual traces from your data source")
print("3. Update config with new manual_traces")
print("4. Run verification")
from karenina.adapters.manual import ManualTraces
from karenina.benchmark import Benchmark
from karenina.schemas import ModelConfig, VerificationConfig
# Create benchmark
benchmark = Benchmark("complete_example_1")
benchmark.add_question(question="What is 2+2?", raw_answer="4", finished=True)
benchmark.add_question(question="What is 3+3?", raw_answer="6", finished=True)
# Initialize manual traces
manual_traces = ManualTraces(benchmark)
# Register traces by question text
manual_traces.register_trace("What is 2+2?", "The answer is 4. I added 2 and 2 to get 4.", map_to_id=True)
manual_traces.register_trace("What is 3+3?", "The answer is 6. I added 3 and 3 to get 6.", map_to_id=True)
# Create manual config
manual_config = ModelConfig(interface="manual", manual_traces=manual_traces)
# Create judge config
judge_config = ModelConfig(
id="gpt-4.1-mini", model_provider="openai", model_name="gpt-4.1-mini", temperature=0.0, interface="langchain"
)
# Run verification
config = VerificationConfig(answering_models=[manual_config], parsing_models=[judge_config])
results = benchmark.run_verification(config)
print(f"Example 1 complete: {len(results.results)} results")
for r in results.results:
print(f" - {r.metadata.question_text[:30]}... passed: {r.template.verify_result}")
Example 2: Batch Registration with Mixed Formats¶
from langchain_core.messages import AIMessage, ToolMessage
# Create benchmark
benchmark = Benchmark("complete_example_2")
benchmark.add_question(question="Question 1?", raw_answer="Answer 1", finished=True)
benchmark.add_question(question="Question 2?", raw_answer="Answer 2", finished=True)
benchmark.add_question(question="Question 3?", raw_answer="Answer 3", finished=True)
# Initialize manual traces
manual_traces = ManualTraces(benchmark)
# Prepare traces dictionary with mixed formats
traces = {
"Question 1?": "Answer 1 is a simple string trace.",
"Question 2?": [
AIMessage(content="Let me think about this..."),
ToolMessage(name="search", content="Found relevant data", tool_call_id="call_1"),
AIMessage(content="Answer 2 based on search results."),
],
"Question 3?": "Answer 3 is another string trace.",
}
# Batch register all at once
manual_traces.register_traces(traces, map_to_id=True)
print(f"Example 2: Batch registered {len(traces)} traces")
print(" - Question 1: String trace")
print(" - Question 2: Message list with tool calls")
print(" - Question 3: String trace")
Example 3: Delayed Trace Population¶
# Create benchmark
benchmark = Benchmark("complete_example_3")
benchmark.add_question(question="What is 2+2?", raw_answer="4", finished=True)
# Step 1: Create config structure early
manual_traces = ManualTraces(benchmark)
manual_config = ModelConfig(interface="manual", manual_traces=manual_traces)
# Step 2: Pass config around, set up verification structure
judge_config = ModelConfig(id="gpt-4.1-mini", model_provider="openai", model_name="gpt-4.1-mini", interface="langchain")
config = VerificationConfig(answering_models=[manual_config], parsing_models=[judge_config])
# Step 3: Later, populate traces (e.g., from file load, API call)
def load_traces_from_file(filepath):
return {"What is 2+2?": "The answer is 4."}
traces_data = load_traces_from_file("experiment_traces.json")
for question_text, trace_content in traces_data.items():
manual_traces.register_trace(question_text, trace_content, map_to_id=True)
# Step 4: Run verification with populated traces
results = benchmark.run_verification(config)
print(f"Example 3 complete: {len(results.results)} results")
Summary¶
The Manual Trace System enables:
- Pre-Generated Answer Evaluation - Evaluate LLM responses without making live API calls
- Flexible Trace Formats - Support for both simple strings and rich message lists
- Agent Metrics Extraction - Automatic tool call and failure tracking
- Efficient Workflows - Batch registration, delayed population, preset compatibility
- Production-Ready - Thread-safe, session-based, with automatic cleanup
Key Workflow: Create benchmark → Initialize ManualTraces → Register traces → Create ModelConfig with interface="manual" → Run verification
Related Documentation:
- Quick Start: Basic verification workflow
- Verification: Complete verification documentation
- Configuration: Model and provider configuration
- TaskEval: Evaluate agent workflow traces (similar concept for pre-existing outputs)