Coverage for src/local_deep_research/benchmarks/metrics/calculation.py: 99%
140 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""
2Unified metrics calculation module.
4This module provides functions for calculating metrics for both
5standard benchmarks and optimization tasks.
6"""
8import json
9from loguru import logger
10from pathlib import Path
11import tempfile
12import time
13from datetime import datetime, UTC
14from typing import Any, Dict, Optional
17def calculate_metrics(results_file: str) -> Dict[str, Any]:
18 """
19 Calculate evaluation metrics from results.
21 Args:
22 results_file: Path to results file
24 Returns:
25 Dictionary of metrics
26 """
27 # Load results
28 results = []
29 try:
30 with open(results_file, "r", encoding="utf-8") as f:
31 for line in f:
32 if line.strip():
33 results.append(json.loads(line))
34 except Exception as e:
35 logger.exception("Error loading results file")
36 return {"error": str(e)}
38 if not results:
39 return {"error": "No results found"}
41 from .statistics import wilson_score_interval
43 # Calculate accuracy
44 graded_results = [r for r in results if "is_correct" in r]
45 correct_count = sum(1 for r in graded_results if r.get("is_correct", False))
46 total_graded = len(graded_results)
47 accuracy = correct_count / total_graded if total_graded else 0
49 # Wilson score confidence interval for accuracy
50 accuracy_ci = wilson_score_interval(correct_count, total_graded)
52 # Calculate average processing time if available
53 processing_times = [
54 r.get("processing_time", 0) for r in results if "processing_time" in r
55 ]
56 avg_time = (
57 sum(processing_times) / len(processing_times) if processing_times else 0
58 )
60 # Average confidence if available
61 confidence_values = []
62 for r in results:
63 if r.get("confidence"):
64 try:
65 confidence_values.append(int(r["confidence"]))
66 except (ValueError, TypeError):
67 pass
69 avg_confidence = (
70 sum(confidence_values) / len(confidence_values)
71 if confidence_values
72 else 0
73 )
75 # Calculate error rate
76 error_count = sum(1 for r in results if "error" in r)
77 error_rate = error_count / len(results) if results else 0
79 # Basic metrics
80 metrics = {
81 "total_examples": len(results),
82 "graded_examples": total_graded,
83 "correct": correct_count,
84 "accuracy": accuracy,
85 "accuracy_ci": accuracy_ci,
86 "average_processing_time": avg_time,
87 "average_confidence": avg_confidence,
88 "error_count": error_count,
89 "error_rate": error_rate,
90 "timestamp": datetime.now(UTC).isoformat(),
91 }
93 # If we have category information, calculate per-category metrics
94 categories = {}
95 for r in graded_results:
96 if "category" in r:
97 category = r["category"]
98 if category not in categories:
99 categories[category] = {"total": 0, "correct": 0}
100 categories[category]["total"] += 1
101 if r.get("is_correct", False):
102 categories[category]["correct"] += 1
104 if categories:
105 category_metrics = {}
106 for category, counts in categories.items():
107 cat_accuracy = (
108 counts["correct"] / counts["total"] if counts["total"] else 0
109 )
110 category_metrics[category] = {
111 "total": counts["total"],
112 "correct": counts["correct"],
113 "accuracy": cat_accuracy,
114 "accuracy_ci": wilson_score_interval(
115 counts["correct"], counts["total"]
116 ),
117 }
118 metrics["categories"] = category_metrics
120 return metrics
123def evaluate_benchmark_quality(
124 system_config: Dict[str, Any],
125 num_examples: int = 10,
126 output_dir: Optional[str] = None,
127) -> Dict[str, Any]:
128 """
129 Evaluate quality using SimpleQA benchmark.
131 Args:
132 system_config: Configuration parameters to evaluate
133 num_examples: Number of benchmark examples to use
134 output_dir: Directory to save results (temporary if None)
136 Returns:
137 Dictionary with benchmark metrics
138 """
139 from ..runners import run_simpleqa_benchmark
141 # Create temporary directory if not provided
142 temp_dir = None
143 if output_dir is None:
144 temp_dir = tempfile.mkdtemp(prefix="ldr_benchmark_")
145 output_dir = temp_dir
147 try:
148 # Create search configuration from system config
149 search_config = {
150 "iterations": system_config.get("iterations", 2),
151 "questions_per_iteration": system_config.get(
152 "questions_per_iteration", 2
153 ),
154 "search_strategy": system_config.get(
155 "search_strategy", "source-based"
156 ),
157 "search_tool": system_config.get("search_tool", "searxng"),
158 "model_name": system_config.get("model_name"),
159 "provider": system_config.get("provider"),
160 }
162 # Run benchmark
163 logger.info(f"Running SimpleQA benchmark with {num_examples} examples")
164 benchmark_results = run_simpleqa_benchmark(
165 num_examples=num_examples,
166 output_dir=output_dir,
167 search_config=search_config,
168 run_evaluation=True,
169 )
171 # Extract key metrics
172 metrics = benchmark_results.get("metrics", {})
173 accuracy = metrics.get("accuracy", 0.0)
175 # Return only the most relevant metrics
176 return {
177 "accuracy": accuracy,
178 "quality_score": accuracy, # Map accuracy directly to quality score
179 }
181 except Exception as e:
182 logger.exception("Error in benchmark evaluation")
183 return {"accuracy": 0.0, "quality_score": 0.0, "error": str(e)}
185 finally:
186 # Clean up temporary directory if we created it
187 if temp_dir and Path(temp_dir).exists():
188 import shutil
190 try:
191 shutil.rmtree(temp_dir)
192 except Exception:
193 logger.warning("Failed to clean up temporary directory")
196def measure_execution_time(
197 system_config: Dict[str, Any],
198 query: str = "test query",
199 search_tool: Optional[str] = None,
200 num_runs: int = 1,
201) -> Dict[str, Any]:
202 """
203 Measure execution time for a given configuration.
205 Args:
206 system_config: Configuration parameters to evaluate
207 query: Query to use for timing tests
208 search_tool: Override search tool
209 num_runs: Number of runs to average time over
211 Returns:
212 Dictionary with speed metrics
213 """
214 from local_deep_research.search_system import AdvancedSearchSystem
215 from local_deep_research.config.llm_config import get_llm
216 from local_deep_research.config.search_config import get_search
218 if search_tool:
219 system_config["search_tool"] = search_tool
221 # Configure system — pre-initialize so finally can clean up partial init
222 llm = None
223 search_engine = None
224 system = None
226 try:
227 llm = get_llm()
228 search_engine = get_search(
229 system_config.get("search_tool", "searxng"),
230 llm_instance=llm,
231 )
232 system = AdvancedSearchSystem(
233 llm=llm,
234 search=search_engine,
235 max_iterations=system_config.get("iterations", 2),
236 questions_per_iteration=system_config.get(
237 "questions_per_iteration", 2
238 ),
239 strategy_name=system_config.get("search_strategy", "source-based"),
240 )
242 # Run multiple times and calculate average
243 total_time = 0.0
244 times: list[float] = []
246 for i in range(num_runs):
247 logger.info(f"Executing speed test run {i + 1}/{num_runs}")
248 start_time = time.time()
249 system.search(query, full_response=False)
250 end_time = time.time()
251 run_time = end_time - start_time
252 times.append(run_time)
253 total_time += run_time
255 # Calculate metrics
256 average_time = total_time / num_runs
258 # Calculate speed score (0-1 scale, lower times are better)
259 # Using sigmoid-like normalization where:
260 # - Times around 30s get ~0.5 score
261 # - Times under 10s get >0.8 score
262 # - Times over 2min get <0.2 score
263 speed_score = 1.0 / (1.0 + (average_time / 30.0))
265 return {
266 "average_time": average_time,
267 "min_time": min(times),
268 "max_time": max(times),
269 "speed_score": speed_score,
270 }
272 except Exception as e:
273 logger.exception("Error in speed measurement")
274 return {"average_time": 0.0, "speed_score": 0.0, "error": str(e)}
275 finally:
276 from local_deep_research.utilities.resource_utils import safe_close
278 safe_close(system, "benchmark search system", allow_none=True)
279 safe_close(search_engine, "benchmark search engine", allow_none=True)
280 safe_close(llm, "benchmark LLM", allow_none=True)
283def calculate_quality_metrics(
284 system_config: Dict[str, Any],
285 num_examples: int = 2, # Reduced for quicker demo
286 output_dir: Optional[str] = None,
287) -> Dict[str, Any]:
288 """
289 Calculate quality-related metrics for a configuration.
291 Args:
292 system_config: Configuration parameters to evaluate
293 num_examples: Number of benchmark examples to use
294 output_dir: Directory to save results (temporary if None)
296 Returns:
297 Dictionary with quality metrics
298 """
299 # Run quality evaluation
300 quality_results = evaluate_benchmark_quality(
301 system_config=system_config,
302 num_examples=num_examples,
303 output_dir=output_dir,
304 )
306 # Return normalized quality score
307 return {
308 "quality_score": quality_results.get("quality_score", 0.0),
309 "accuracy": quality_results.get("accuracy", 0.0),
310 }
313def calculate_speed_metrics(
314 system_config: Dict[str, Any],
315 query: str = "test query",
316 search_tool: Optional[str] = None,
317 num_runs: int = 1,
318) -> Dict[str, Any]:
319 """
320 Calculate speed-related metrics for a configuration.
322 Args:
323 system_config: Configuration parameters to evaluate
324 query: Query to use for timing tests
325 search_tool: Override search tool
326 num_runs: Number of runs to average time over
328 Returns:
329 Dictionary with speed metrics
330 """
331 # Run speed measurement
332 speed_results = measure_execution_time(
333 system_config=system_config,
334 query=query,
335 search_tool=search_tool,
336 num_runs=num_runs,
337 )
339 # Return normalized speed score
340 return {
341 "speed_score": speed_results.get("speed_score", 0.0),
342 "average_time": speed_results.get("average_time", 0.0),
343 }
346def calculate_resource_metrics(
347 system_config: Dict[str, Any],
348 query: str = "test query",
349 search_tool: Optional[str] = None,
350) -> Dict[str, Any]:
351 """
352 Calculate resource usage metrics for a configuration.
354 Args:
355 system_config: Configuration parameters to evaluate
356 query: Query to use for resource tests
357 search_tool: Override search tool
359 Returns:
360 Dictionary with resource metrics
361 """
362 # This is a simplified version - in a real implementation,
363 # you would measure memory usage, API call counts, etc.
365 # For now, we'll use a heuristic based on configuration values
366 iterations = system_config.get("iterations", 2)
367 questions = system_config.get("questions_per_iteration", 2)
368 max_results = system_config.get("max_results", 50)
370 # Simple heuristic: more iterations, questions, and results = more resources
371 complexity = iterations * questions * (max_results / 50)
373 # Normalize to 0-1 scale (lower is better)
374 resource_score = 1.0 / (1.0 + (complexity / 4.0))
376 return {
377 "resource_score": resource_score,
378 "estimated_complexity": complexity,
379 }
382def calculate_combined_score(
383 metrics: Dict[str, Dict[str, float]],
384 weights: Optional[Dict[str, float]] = None,
385) -> float:
386 """
387 Calculate a combined optimization score from multiple metrics.
389 Args:
390 metrics: Dictionary of metric categories and their values
391 weights: Dictionary of weights for each metric category
393 Returns:
394 Combined score between 0 and 1
395 """
396 # Default weights if not provided
397 if weights is None:
398 weights = {"quality": 0.6, "speed": 0.3, "resource": 0.1}
400 # Normalize weights to sum to 1
401 total_weight = sum(weights.values())
402 if total_weight == 0:
403 return 0.0
405 norm_weights = {k: v / total_weight for k, v in weights.items()}
407 # Calculate weighted score
408 score = 0.0
410 # Quality component
411 if "quality" in metrics and "quality" in norm_weights:
412 quality_score = metrics["quality"].get("quality_score", 0.0)
413 score += quality_score * norm_weights["quality"]
415 # Speed component
416 if "speed" in metrics and "speed" in norm_weights:
417 speed_score = metrics["speed"].get("speed_score", 0.0)
418 score += speed_score * norm_weights["speed"]
420 # Resource component
421 if "resource" in metrics and "resource" in norm_weights:
422 resource_score = metrics["resource"].get("resource_score", 0.0)
423 score += resource_score * norm_weights["resource"]
425 return score