liqiang
b119135836
Build latest book artifacts / build (push) Canceled after 0s
dependency resolution / resolve (3.11) (push) Canceled after 0s
dependency resolution / resolve (3.13) (push) Canceled after 0s
deploy-pages / build (push) Canceled after 0s
deploy-pages / deploy (push) Canceled after 0s
i18n consistency check / check (push) Canceled after 0s
provider adoption tests / test (chapter2/context-compression) (push) Canceled after 0s
provider adoption tests / test (chapter2/prompt-injection) (push) Canceled after 0s
provider adoption tests / test (chapter2/system-hint) (push) Canceled after 0s
provider adoption tests / test (chapter3/log-sanitization) (push) Canceled after 0s
web-search-agent tests / test (push) Canceled after 0s
web-search-agent tests / agentbook (push) Canceled after 0s
763 lines
32 KiB
Python
763 lines
32 KiB
Python
"""Contextual Memory Evaluator with Dual Memory System
|
|
|
|
This evaluator tests the combined system of:
|
|
1. Contextual RAG for conversation chunks
|
|
2. Advanced JSON cards for structured facts
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import logging
|
|
import time
|
|
import uuid
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Dict, List, Any, Optional, Tuple
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
|
|
from config import Config
|
|
from contextual_indexer import ContextualMemoryIndexer
|
|
from contextual_agent import ContextualUserMemoryAgent
|
|
from advanced_memory_manager import AdvancedMemoryManager, AdvancedMemoryCard
|
|
from chunker import ConversationChunker, ConversationChunk
|
|
|
|
|
|
def _reasoning_safe_temperature(model, requested=1.0):
|
|
"""Reasoning models (Kimi K3, GPT-5, ...) only accept temperature=1.
|
|
Return 1 for those; otherwise the requested value so non-reasoning
|
|
providers (Doubao, DeepSeek, older Moonshot) are unchanged."""
|
|
m = str(model or "").lower().replace("/", "-")
|
|
return 1 if ("kimi-k3" in m or "gpt-5" in m) else requested
|
|
|
|
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Import LLM evaluator from user-memory-evaluation project
|
|
LLMEvaluator = None
|
|
EvalTestCase = None
|
|
ConversationHistory = None
|
|
EvalMessage = None
|
|
MessageRole = None
|
|
|
|
# Try to import from week2/user-memory-evaluation
|
|
eval_project_path = Path(__file__).parent.parent.parent / "week2" / "user-memory-evaluation"
|
|
if eval_project_path.exists():
|
|
sys.path.insert(0, str(eval_project_path))
|
|
try:
|
|
from llm_evaluator import LLMEvaluator
|
|
from test_case import TestCase as EvalTestCase, ConversationHistory, Message as EvalMessage, MessageRole
|
|
logger.info("Successfully imported LLM evaluation modules")
|
|
except ImportError as e:
|
|
logger.warning(f"Could not import LLM evaluation modules: {e}")
|
|
logger.info("LLM evaluation will not be available")
|
|
else:
|
|
logger.warning(f"user-memory-evaluation project not found at {eval_project_path}")
|
|
logger.info("LLM evaluation will not be available")
|
|
|
|
|
|
@dataclass
|
|
class TestCase:
|
|
"""Enhanced test case with support for advanced memory"""
|
|
test_id: str
|
|
category: str # layer1, layer2, layer3
|
|
title: str
|
|
description: str
|
|
conversation_histories: List[Dict[str, Any]] # Conversation data directly from YAML
|
|
user_question: str
|
|
evaluation_criteria: str
|
|
expected_behavior: Optional[str] = None
|
|
expected_memory_cards: List[Dict[str, Any]] = field(default_factory=list)
|
|
metadata: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
@classmethod
|
|
def from_yaml(cls, data: Dict[str, Any], test_dir: Path) -> 'TestCase':
|
|
"""Create test case from YAML data"""
|
|
# The conversation_histories field directly contains the conversation data
|
|
conversation_histories = data.get('conversation_histories', [])
|
|
|
|
return cls(
|
|
test_id=data.get('test_id', data.get('id', '')),
|
|
category=data.get('category', ''),
|
|
title=data.get('title', ''),
|
|
description=data.get('description', ''),
|
|
conversation_histories=conversation_histories,
|
|
user_question=data.get('user_question', ''),
|
|
evaluation_criteria=data.get('evaluation_criteria', ''),
|
|
expected_behavior=data.get('expected_behavior'),
|
|
expected_memory_cards=data.get('expected_memory_cards', []),
|
|
metadata=data.get('metadata', {})
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class EvaluationResult:
|
|
"""Enhanced evaluation result with dual memory tracking"""
|
|
test_id: str
|
|
success: bool
|
|
agent_answer: str
|
|
evaluation_criteria: str
|
|
iterations: int
|
|
tool_calls: int
|
|
memory_cards_used: List[str]
|
|
chunks_retrieved: List[str]
|
|
contextual_chunks_count: int
|
|
processing_time: float
|
|
indexing_time: float
|
|
context_generation_time: float
|
|
llm_evaluation: Optional[Dict[str, Any]] = None
|
|
error: Optional[str] = None
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"test_id": self.test_id,
|
|
"success": self.success,
|
|
"agent_answer": self.agent_answer,
|
|
"evaluation_criteria": self.evaluation_criteria,
|
|
"iterations": self.iterations,
|
|
"tool_calls": self.tool_calls,
|
|
"memory_cards_used": self.memory_cards_used,
|
|
"chunks_retrieved": self.chunks_retrieved,
|
|
"contextual_chunks_count": self.contextual_chunks_count,
|
|
"processing_time": self.processing_time,
|
|
"indexing_time": self.indexing_time,
|
|
"context_generation_time": self.context_generation_time,
|
|
"llm_evaluation": self.llm_evaluation,
|
|
"error": self.error
|
|
}
|
|
|
|
|
|
class ContextualMemoryEvaluator:
|
|
"""Evaluates the dual memory system on test cases"""
|
|
|
|
def __init__(self, config: Optional[Config] = None):
|
|
"""
|
|
Initialize the evaluator
|
|
|
|
Args:
|
|
config: Configuration object
|
|
"""
|
|
self.config = config or Config.from_env()
|
|
self.test_cases: Dict[str, TestCase] = {}
|
|
self.results: Dict[str, EvaluationResult] = {}
|
|
|
|
# Initialize components
|
|
self.chunker = ConversationChunker(self.config.chunking)
|
|
self.indexer: Optional[ContextualMemoryIndexer] = None
|
|
self.agent: Optional[ContextualUserMemoryAgent] = None
|
|
|
|
# Initialize LLM evaluator if available
|
|
self.llm_evaluator = None
|
|
logger.info("Checking LLM Evaluator availability...")
|
|
logger.info(f"LLMEvaluator module: {LLMEvaluator}")
|
|
logger.info(f"EvalTestCase module: {EvalTestCase}")
|
|
|
|
if LLMEvaluator:
|
|
try:
|
|
logger.info("Attempting to initialize LLM Evaluator...")
|
|
self.llm_evaluator = LLMEvaluator()
|
|
logger.info("✅ LLM Evaluator initialized successfully for automatic evaluation")
|
|
except Exception as e:
|
|
logger.warning(f"Could not initialize LLM evaluator: {e}")
|
|
logger.info("Automatic LLM evaluation will be skipped")
|
|
else:
|
|
logger.info("LLM Evaluator not available - automatic evaluation will be skipped")
|
|
|
|
# Evaluation framework path
|
|
self.eval_framework_path = Path("../user-memory-evaluation/test_cases")
|
|
if not self.eval_framework_path.exists():
|
|
self.eval_framework_path = Path("../../week2/user-memory-evaluation/test_cases")
|
|
|
|
logger.info("Initialized ContextualMemoryEvaluator")
|
|
|
|
def load_test_cases(self, category: Optional[str] = None) -> List[str]:
|
|
"""
|
|
Load test cases from the evaluation framework
|
|
|
|
Args:
|
|
category: Optional category filter (layer1, layer2, layer3)
|
|
|
|
Returns:
|
|
List of loaded test case IDs
|
|
"""
|
|
import yaml
|
|
|
|
test_cases = []
|
|
test_dirs = ["layer1", "layer2", "layer3"] if not category else [category]
|
|
|
|
for test_dir in test_dirs:
|
|
dir_path = self.eval_framework_path / test_dir
|
|
if not dir_path.exists():
|
|
logger.warning(f"Test directory not found: {dir_path}")
|
|
continue
|
|
|
|
# Load all YAML files
|
|
for yaml_file in dir_path.glob("*.yaml"):
|
|
try:
|
|
with open(yaml_file, 'r', encoding='utf-8') as f:
|
|
data = yaml.safe_load(f)
|
|
|
|
test_case = TestCase.from_yaml(data, self.eval_framework_path)
|
|
test_case.category = test_dir # Ensure category is set
|
|
|
|
self.test_cases[test_case.test_id] = test_case
|
|
test_cases.append(test_case.test_id)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error loading test case {yaml_file}: {e}")
|
|
|
|
logger.info(f"Loaded {len(test_cases)} test cases")
|
|
return test_cases
|
|
|
|
def _prepare_memory_cards(self, test_case: TestCase) -> List[AdvancedMemoryCard]:
|
|
"""
|
|
Generate memory cards for a test case
|
|
|
|
For Layer 3 tests, this creates cards that enable proactive service
|
|
"""
|
|
cards = []
|
|
|
|
# Generate cards based on test metadata
|
|
if test_case.category == "layer3":
|
|
# Layer 3 needs structured facts for proactive service
|
|
if "travel" in test_case.test_id.lower():
|
|
cards.append(AdvancedMemoryCard(
|
|
category="travel",
|
|
card_key="tokyo_trip",
|
|
backstory="User booked a trip to Tokyo in previous conversations",
|
|
date_created=datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
|
person="Jessica Thompson (primary)",
|
|
relationship="primary account holder",
|
|
data={
|
|
"destination": "Tokyo, Japan",
|
|
"departure_date": "2025-01-25",
|
|
"return_date": "2025-02-01",
|
|
"purpose": "business conference"
|
|
}
|
|
))
|
|
|
|
cards.append(AdvancedMemoryCard(
|
|
category="travel",
|
|
card_key="passport_jessica",
|
|
backstory="User's passport expiration was mentioned when booking travel",
|
|
date_created=datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
|
person="Jessica Thompson (primary)",
|
|
relationship="primary account holder",
|
|
data={
|
|
"passport_number": "XXXXX1234",
|
|
"expiration_date": "2025-02-18",
|
|
"issuing_country": "USA",
|
|
"needs_renewal": True
|
|
}
|
|
))
|
|
|
|
# Add any expected cards from test case
|
|
for card_data in test_case.expected_memory_cards:
|
|
cards.append(AdvancedMemoryCard(
|
|
category=card_data.get("category", "general"),
|
|
card_key=card_data.get("key", str(uuid.uuid4())),
|
|
backstory=card_data.get("backstory", "From test case"),
|
|
date_created=datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
|
person=card_data.get("person", "User"),
|
|
relationship=card_data.get("relationship", "primary"),
|
|
data=card_data.get("data", {})
|
|
))
|
|
|
|
return cards
|
|
|
|
def evaluate_test_case(self, test_id: str) -> EvaluationResult:
|
|
"""
|
|
Evaluate a single test case with dual memory system
|
|
|
|
Args:
|
|
test_id: Test case ID
|
|
|
|
Returns:
|
|
Evaluation result
|
|
"""
|
|
if test_id not in self.test_cases:
|
|
raise ValueError(f"Test case {test_id} not found")
|
|
|
|
test_case = self.test_cases[test_id]
|
|
logger.info(f"Evaluating test case: {test_id} - {test_case.title}")
|
|
|
|
start_time = time.time()
|
|
indexing_start = time.time()
|
|
|
|
try:
|
|
# Step 1: Initialize indexer with user ID
|
|
user_id = f"test_user_{test_id}"
|
|
self.indexer = ContextualMemoryIndexer(
|
|
user_id=user_id,
|
|
index_config=self.config.index,
|
|
chunking_config=self.config.chunking,
|
|
use_contextual=True # Enable contextual chunking
|
|
)
|
|
|
|
# Step 2: Process conversations from test case
|
|
all_chunks = []
|
|
for conv_data in test_case.conversation_histories:
|
|
# Extract conversation data from the test case
|
|
conv_id = conv_data.get('conversation_id', f'conv_{test_id}')
|
|
messages = conv_data.get('messages', [])
|
|
|
|
# Chunk the conversation
|
|
chunks = self.chunker.chunk_conversation(
|
|
messages=messages,
|
|
conversation_id=conv_id,
|
|
test_id=test_id
|
|
)
|
|
all_chunks.extend(chunks)
|
|
|
|
logger.info(f"Created {len(all_chunks)} basic chunks")
|
|
|
|
# Step 3: Process with contextual chunking and indexing
|
|
processing_result = self.indexer.process_conversation_history(
|
|
chunks=all_chunks,
|
|
conversation_id=test_id,
|
|
generate_summary_cards=True # Generate cards from conversations
|
|
)
|
|
|
|
# Step 4: Add pre-defined memory cards for the test
|
|
memory_cards = self._prepare_memory_cards(test_case)
|
|
for card in memory_cards:
|
|
self.indexer.memory_manager.add_card(card)
|
|
|
|
# Debug: Print all memory cards
|
|
logger.info("="*60)
|
|
logger.info("DEBUG: All Memory Cards in System")
|
|
logger.info("="*60)
|
|
# Access cards directly from categories attribute
|
|
for category, cards in self.indexer.memory_manager.categories.items():
|
|
for card_key, card in cards.items():
|
|
logger.info(f"\n[{category}.{card_key}]")
|
|
logger.info(json.dumps(card.to_dict(), indent=2, ensure_ascii=False))
|
|
total_cards = sum(len(cards) for cards in self.indexer.memory_manager.categories.values())
|
|
logger.info(f"\nTotal Memory Cards: {total_cards}")
|
|
logger.info("="*60)
|
|
|
|
indexing_time = time.time() - indexing_start
|
|
|
|
# Step 5: Initialize agent with dual memory
|
|
self.agent = ContextualUserMemoryAgent(
|
|
indexer=self.indexer,
|
|
memory_manager=self.indexer.memory_manager,
|
|
config=self.config
|
|
)
|
|
|
|
# Step 6: Answer the question
|
|
trajectory = self.agent.answer_question(
|
|
question=test_case.user_question,
|
|
test_id=test_id,
|
|
max_iterations=self.config.evaluation.max_iterations,
|
|
stream=False
|
|
)
|
|
|
|
processing_time = time.time() - start_time
|
|
|
|
# Get context generation statistics
|
|
chunker_stats = self.indexer.contextual_chunker.get_statistics()
|
|
context_gen_time = chunker_stats.get("total_generation_time", 0)
|
|
|
|
# Step 7: Evaluate the answer with LLM (automatic if available)
|
|
llm_evaluation = None
|
|
logger.info("="*60)
|
|
logger.info("LLM Judge Evaluation")
|
|
logger.info("="*60)
|
|
|
|
if not self.llm_evaluator:
|
|
logger.warning("LLM Judge not available - skipping automatic evaluation")
|
|
logger.info("To enable LLM Judge, ensure the llm_evaluator module is properly imported")
|
|
elif not trajectory.final_answer:
|
|
logger.warning("No final answer from agent - cannot evaluate")
|
|
else:
|
|
logger.info("Running LLM Judge evaluation...")
|
|
|
|
if self.llm_evaluator and trajectory.final_answer:
|
|
try:
|
|
# Convert test case to evaluation format
|
|
eval_histories = []
|
|
for hist in test_case.conversation_histories:
|
|
eval_messages = []
|
|
for msg in hist.get("messages", []):
|
|
role_str = msg.get("role", "user")
|
|
role = MessageRole.USER if role_str == "user" else MessageRole.ASSISTANT
|
|
eval_messages.append(EvalMessage(
|
|
role=role,
|
|
content=msg.get("content", "")
|
|
))
|
|
|
|
eval_histories.append(ConversationHistory(
|
|
conversation_id=hist.get("conversation_id", f"conv_{uuid.uuid4().hex[:8]}"),
|
|
messages=eval_messages,
|
|
metadata=hist.get("metadata", {})
|
|
))
|
|
|
|
# Create evaluation test case
|
|
eval_test_case = EvalTestCase(
|
|
test_id=test_case.test_id,
|
|
category=test_case.category,
|
|
title=test_case.title,
|
|
description=test_case.description,
|
|
conversation_histories=eval_histories,
|
|
user_question=test_case.user_question,
|
|
evaluation_criteria=test_case.evaluation_criteria if test_case.evaluation_criteria else "The agent should provide a relevant and accurate response based on the conversation history.",
|
|
expected_behavior=test_case.expected_behavior
|
|
)
|
|
|
|
# Run LLM evaluation
|
|
llm_result = self.llm_evaluator.evaluate(
|
|
test_case=eval_test_case,
|
|
agent_response=trajectory.final_answer,
|
|
extracted_memory=None
|
|
)
|
|
|
|
llm_evaluation = {
|
|
"reward": llm_result.reward,
|
|
"passed": llm_result.passed if llm_result.passed is not None else llm_result.reward >= 0.6,
|
|
"reasoning": llm_result.reasoning,
|
|
"required_info_found": llm_result.required_info_found if hasattr(llm_result, 'required_info_found') else {},
|
|
"suggestions": llm_result.suggestions if hasattr(llm_result, 'suggestions') else None
|
|
}
|
|
|
|
# Log evaluation results
|
|
logger.info("-"*60)
|
|
logger.info(f"LLM Evaluation Reward: {llm_result.reward:.3f}/1.000")
|
|
logger.info(f"Passed: {'Yes' if llm_evaluation['passed'] else 'No'}")
|
|
logger.info("-"*60)
|
|
logger.info(f"Evaluation Reasoning:")
|
|
logger.info(llm_result.reasoning)
|
|
logger.info("-"*60)
|
|
|
|
if llm_result.required_info_found:
|
|
logger.info("Required Information Found:")
|
|
for key, found in llm_result.required_info_found.items():
|
|
status = "✓" if found else "✗"
|
|
logger.info(f" {status} {key}")
|
|
logger.info("-"*60)
|
|
|
|
except Exception as e:
|
|
logger.warning(f"LLM evaluation failed: {e}")
|
|
logger.debug("Full error:", exc_info=True)
|
|
else:
|
|
# Fallback: Use direct LLM API for evaluation if module not available
|
|
if trajectory.final_answer:
|
|
logger.info("Attempting fallback LLM evaluation...")
|
|
llm_evaluation = self._fallback_llm_evaluation(test_case, trajectory.final_answer)
|
|
|
|
# Create result
|
|
result = EvaluationResult(
|
|
test_id=test_id,
|
|
success=trajectory.success,
|
|
agent_answer=trajectory.final_answer or "",
|
|
evaluation_criteria=test_case.evaluation_criteria,
|
|
iterations=len(trajectory.iterations),
|
|
tool_calls=len(trajectory.tool_calls),
|
|
memory_cards_used=trajectory.memory_cards_used,
|
|
chunks_retrieved=trajectory.chunks_retrieved,
|
|
contextual_chunks_count=processing_result.get("contextual_chunks", 0),
|
|
processing_time=processing_time,
|
|
indexing_time=indexing_time,
|
|
context_generation_time=context_gen_time,
|
|
llm_evaluation=llm_evaluation
|
|
)
|
|
|
|
# Check if answer meets criteria
|
|
# Use LLM evaluation result if available, otherwise fall back to keyword check
|
|
if llm_evaluation and 'passed' in llm_evaluation:
|
|
result.success = llm_evaluation['passed']
|
|
logger.info(f"Using LLM evaluation result: {'Success' if result.success else 'Failed'}")
|
|
elif self._check_answer_criteria(trajectory.final_answer, test_case.evaluation_criteria):
|
|
result.success = True
|
|
logger.info(f"Using keyword-based evaluation: Success")
|
|
else:
|
|
logger.info(f"Using keyword-based evaluation: Failed")
|
|
|
|
self.results[test_id] = result
|
|
logger.info(f"Evaluation complete for {test_id}: {'Success' if result.success else 'Failed'}")
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error evaluating test case {test_id}: {e}")
|
|
|
|
result = EvaluationResult(
|
|
test_id=test_id,
|
|
success=False,
|
|
agent_answer="",
|
|
evaluation_criteria=test_case.evaluation_criteria,
|
|
iterations=0,
|
|
tool_calls=0,
|
|
memory_cards_used=[],
|
|
chunks_retrieved=[],
|
|
contextual_chunks_count=0,
|
|
processing_time=time.time() - start_time,
|
|
indexing_time=indexing_time if 'indexing_time' in locals() else 0,
|
|
context_generation_time=0,
|
|
error=str(e)
|
|
)
|
|
|
|
self.results[test_id] = result
|
|
return result
|
|
|
|
def _load_conversations(self, conv_file: str) -> Dict[str, List[Dict[str, str]]]:
|
|
"""Load conversations from a JSON file"""
|
|
conversations = {}
|
|
|
|
try:
|
|
with open(conv_file, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
|
|
# Handle different formats
|
|
if isinstance(data, dict):
|
|
if "conversations" in data:
|
|
conversations = data["conversations"]
|
|
else:
|
|
# Assume it's already in the right format
|
|
conversations = data
|
|
elif isinstance(data, list):
|
|
# Convert list to dict with generated IDs
|
|
for i, conv in enumerate(data):
|
|
conversations[f"conv_{i}"] = conv
|
|
|
|
return conversations
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error loading conversations from {conv_file}: {e}")
|
|
return {}
|
|
|
|
def _check_answer_criteria(self, answer: Optional[str], criteria: str) -> bool:
|
|
"""Simple check if answer meets criteria"""
|
|
if not answer:
|
|
return False
|
|
|
|
answer_lower = answer.lower()
|
|
criteria_lower = criteria.lower()
|
|
|
|
# Extract key terms from criteria
|
|
key_terms = []
|
|
for word in criteria_lower.split():
|
|
if len(word) > 4 and word not in ["should", "must", "need", "have"]:
|
|
key_terms.append(word)
|
|
|
|
# Check if key terms appear in answer
|
|
matches = sum(1 for term in key_terms if term in answer_lower)
|
|
|
|
return matches >= min(3, len(key_terms) // 2)
|
|
|
|
def _evaluate_with_llm(self, test_case: TestCase, agent_answer: Optional[str]) -> Dict[str, Any]:
|
|
"""Use LLM to evaluate if the answer meets criteria"""
|
|
if not agent_answer:
|
|
return {
|
|
"passed": False,
|
|
"reasoning": "No answer provided",
|
|
"reward": 0.0
|
|
}
|
|
|
|
# This is a simplified version - implement full LLM evaluation as needed
|
|
return {
|
|
"passed": self._check_answer_criteria(agent_answer, test_case.evaluation_criteria),
|
|
"reasoning": "Basic criteria check",
|
|
"reward": 0.5
|
|
}
|
|
|
|
def _fallback_llm_evaluation(self, test_case: TestCase, agent_answer: str) -> Dict[str, Any]:
|
|
"""Fallback LLM evaluation using direct API call when llm_evaluator module isn't available"""
|
|
try:
|
|
from openai import OpenAI
|
|
|
|
config = Config.from_env()
|
|
client_config, model = config.llm.get_client_config()
|
|
base_url = client_config.pop("base_url", None)
|
|
|
|
if base_url:
|
|
client = OpenAI(base_url=base_url, **client_config)
|
|
else:
|
|
client = OpenAI(**client_config)
|
|
|
|
# Create evaluation prompt
|
|
eval_prompt = f"""Evaluate the agent's response based on the test criteria.
|
|
|
|
Test Question: {test_case.user_question}
|
|
|
|
Agent's Answer: {agent_answer}
|
|
|
|
Evaluation Criteria: {test_case.evaluation_criteria}
|
|
|
|
Provide a JSON evaluation with:
|
|
1. "reward": A score from 0.0 to 1.0 (0.6+ is passing)
|
|
2. "passed": Boolean indicating if the answer meets the criteria
|
|
3. "reasoning": Explanation for the score (2-3 sentences)
|
|
4. "required_info_found": Object with boolean values for each required piece of information
|
|
|
|
Respond with valid JSON only."""
|
|
|
|
response = client.chat.completions.create(
|
|
model=model,
|
|
messages=[
|
|
{"role": "system", "content": "You are an evaluation judge. Evaluate if the agent's answer correctly addresses the user's question based on the criteria."},
|
|
{"role": "user", "content": eval_prompt}
|
|
],
|
|
temperature=_reasoning_safe_temperature(model, 0.1),
|
|
response_format={"type": "json_object"}
|
|
)
|
|
|
|
try:
|
|
result = json.loads(response.choices[0].message.content)
|
|
except json.JSONDecodeError:
|
|
result = {}
|
|
|
|
# Ensure required fields
|
|
if 'reward' not in result:
|
|
result['reward'] = 0.5
|
|
if 'passed' not in result:
|
|
result['passed'] = result['reward'] >= 0.6
|
|
if 'reasoning' not in result:
|
|
result['reasoning'] = "Evaluation completed"
|
|
if 'required_info_found' not in result:
|
|
result['required_info_found'] = {}
|
|
|
|
# Log evaluation results
|
|
logger.info("="*60)
|
|
logger.info("Fallback LLM Evaluation Results")
|
|
logger.info("="*60)
|
|
logger.info(f"Reward: {result['reward']:.3f}/1.000")
|
|
logger.info(f"Passed: {'Yes' if result['passed'] else 'No'}")
|
|
logger.info(f"Reasoning: {result['reasoning']}")
|
|
|
|
if result['required_info_found']:
|
|
logger.info("Required Information Found:")
|
|
for key, found in result['required_info_found'].items():
|
|
status = "✓" if found else "✗"
|
|
logger.info(f" {status} {key}")
|
|
logger.info("="*60)
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"Fallback LLM evaluation failed: {e}")
|
|
return None
|
|
|
|
def evaluate_batch(self, test_ids: Optional[List[str]] = None) -> Dict[str, EvaluationResult]:
|
|
"""Evaluate multiple test cases"""
|
|
if test_ids is None:
|
|
test_ids = list(self.test_cases.keys())
|
|
|
|
for test_id in test_ids:
|
|
try:
|
|
self.evaluate_test_case(test_id)
|
|
except Exception as e:
|
|
logger.error(f"Failed to evaluate {test_id}: {e}")
|
|
|
|
return self.results
|
|
|
|
def generate_report(self) -> str:
|
|
"""Generate evaluation report"""
|
|
if not self.results:
|
|
return "No evaluation results available"
|
|
|
|
report = ["=" * 80]
|
|
report.append("CONTEXTUAL MEMORY EVALUATION REPORT")
|
|
report.append("=" * 80)
|
|
report.append(f"Generated: {datetime.now().isoformat()}")
|
|
report.append(f"Total Test Cases: {len(self.results)}")
|
|
report.append("")
|
|
|
|
# Summary statistics
|
|
successful = sum(1 for r in self.results.values() if r.success)
|
|
report.append(f"Success Rate: {successful}/{len(self.results)} ({100*successful/len(self.results):.1f}%)")
|
|
report.append("")
|
|
|
|
# Statistics by category
|
|
categories = {}
|
|
for test_id, result in self.results.items():
|
|
if test_id in self.test_cases:
|
|
cat = self.test_cases[test_id].category
|
|
if cat not in categories:
|
|
categories[cat] = {"total": 0, "success": 0}
|
|
categories[cat]["total"] += 1
|
|
if result.success:
|
|
categories[cat]["success"] += 1
|
|
|
|
report.append("Results by Category:")
|
|
for cat in sorted(categories.keys()):
|
|
stats = categories[cat]
|
|
rate = 100 * stats["success"] / stats["total"] if stats["total"] > 0 else 0
|
|
report.append(f" {cat}: {stats['success']}/{stats['total']} ({rate:.1f}%)")
|
|
report.append("")
|
|
|
|
# Average metrics
|
|
avg_iterations = sum(r.iterations for r in self.results.values()) / len(self.results)
|
|
avg_tool_calls = sum(r.tool_calls for r in self.results.values()) / len(self.results)
|
|
avg_time = sum(r.processing_time for r in self.results.values()) / len(self.results)
|
|
avg_context_time = sum(r.context_generation_time for r in self.results.values()) / len(self.results)
|
|
|
|
report.append("Average Metrics:")
|
|
report.append(f" Iterations: {avg_iterations:.1f}")
|
|
report.append(f" Tool Calls: {avg_tool_calls:.1f}")
|
|
report.append(f" Processing Time: {avg_time:.2f}s")
|
|
report.append(f" Context Generation Time: {avg_context_time:.2f}s")
|
|
report.append("")
|
|
|
|
# Memory usage statistics
|
|
total_cards_used = sum(len(r.memory_cards_used) for r in self.results.values())
|
|
total_chunks_retrieved = sum(len(r.chunks_retrieved) for r in self.results.values())
|
|
|
|
report.append("Memory Usage:")
|
|
report.append(f" Total Memory Cards Used: {total_cards_used}")
|
|
report.append(f" Total Chunks Retrieved: {total_chunks_retrieved}")
|
|
report.append(f" Avg Cards per Query: {total_cards_used/len(self.results):.1f}")
|
|
report.append(f" Avg Chunks per Query: {total_chunks_retrieved/len(self.results):.1f}")
|
|
report.append("")
|
|
|
|
# Individual test results
|
|
report.append("-" * 80)
|
|
report.append("INDIVIDUAL TEST RESULTS")
|
|
report.append("-" * 80)
|
|
|
|
for test_id, result in sorted(self.results.items()):
|
|
test_case = self.test_cases.get(test_id)
|
|
report.append(f"\nTest: {test_id}")
|
|
if test_case:
|
|
report.append(f" Title: {test_case.title}")
|
|
report.append(f" Category: {test_case.category}")
|
|
report.append(f" Status: {'✓ Success' if result.success else '✗ Failed'}")
|
|
report.append(f" Iterations: {result.iterations}")
|
|
report.append(f" Tool Calls: {result.tool_calls}")
|
|
report.append(f" Memory Cards Used: {len(result.memory_cards_used)}")
|
|
report.append(f" Chunks Retrieved: {len(result.chunks_retrieved)}")
|
|
report.append(f" Contextual Chunks: {result.contextual_chunks_count}")
|
|
report.append(f" Processing Time: {result.processing_time:.2f}s")
|
|
if result.error:
|
|
report.append(f" Error: {result.error}")
|
|
|
|
return "\n".join(report)
|
|
|
|
def save_results(self, output_file: str):
|
|
"""Save evaluation results to file"""
|
|
results_data = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"config": {
|
|
"use_contextual": True,
|
|
"llm_provider": self.config.llm.provider,
|
|
"llm_model": self.config.llm.model,
|
|
"chunking_strategy": self.config.chunking.strategy.value,
|
|
"index_mode": self.config.index.mode.value
|
|
},
|
|
"summary": {
|
|
"total_tests": len(self.results),
|
|
"successful": sum(1 for r in self.results.values() if r.success),
|
|
"failed": sum(1 for r in self.results.values() if not r.success)
|
|
},
|
|
"results": {
|
|
test_id: result.to_dict()
|
|
for test_id, result in self.results.items()
|
|
}
|
|
}
|
|
|
|
with open(output_file, 'w', encoding='utf-8') as f:
|
|
json.dump(results_data, f, indent=2, ensure_ascii=False)
|
|
|
|
logger.info(f"Results saved to {output_file}")
|