Coverage for src/local_deep_research/benchmarks/metrics/calculation.py: 99%
142 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +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
214 Raises:
215 ValueError: If num_runs is less than 1.
216 """
217 if num_runs < 1:
218 # Reject the call BEFORE the try/except below, which would otherwise
219 # swallow the exception and return a dict carrying the same message
220 # as 'error' (a confusing shape — the caller asked for an
221 # average over N runs, and N=0 is not a configuration, it's a bug
222 # in the caller). The downstream `total_time / num_runs` would
223 # raise ZeroDivisionError, which the broad except turns into
224 # `{'error': '... division by zero'}`; the explicit guard makes
225 # the failure mode actionable (#6664).
226 raise ValueError(f"num_runs must be >= 1, got: {num_runs!r}")
228 from local_deep_research.search_system import AdvancedSearchSystem
229 from local_deep_research.config.llm_config import get_llm
230 from local_deep_research.config.search_config import get_search
232 if search_tool:
233 system_config["search_tool"] = search_tool
235 # Configure system — pre-initialize so finally can clean up partial init
236 llm = None
237 search_engine = None
238 system = None
240 try:
241 llm = get_llm()
242 search_engine = get_search(
243 system_config.get("search_tool", "searxng"),
244 llm_instance=llm,
245 )
246 system = AdvancedSearchSystem(
247 llm=llm,
248 search=search_engine,
249 max_iterations=system_config.get("iterations", 2),
250 questions_per_iteration=system_config.get(
251 "questions_per_iteration", 2
252 ),
253 strategy_name=system_config.get("search_strategy", "source-based"),
254 )
256 # Run multiple times and calculate average
257 total_time = 0.0
258 times: list[float] = []
260 for i in range(num_runs):
261 logger.info(f"Executing speed test run {i + 1}/{num_runs}")
262 start_time = time.time()
263 system.search(query, full_response=False)
264 end_time = time.time()
265 run_time = end_time - start_time
266 times.append(run_time)
267 total_time += run_time
269 # Calculate metrics
270 average_time = total_time / num_runs
272 # Calculate speed score (0-1 scale, lower times are better)
273 # Using sigmoid-like normalization where:
274 # - Times around 30s get ~0.5 score
275 # - Times under 10s get >0.8 score
276 # - Times over 2min get <0.2 score
277 speed_score = 1.0 / (1.0 + (average_time / 30.0))
279 return {
280 "average_time": average_time,
281 "min_time": min(times),
282 "max_time": max(times),
283 "speed_score": speed_score,
284 }
286 except Exception as e:
287 logger.exception("Error in speed measurement")
288 return {"average_time": 0.0, "speed_score": 0.0, "error": str(e)}
289 finally:
290 from local_deep_research.utilities.resource_utils import safe_close
292 safe_close(system, "benchmark search system", allow_none=True)
293 safe_close(search_engine, "benchmark search engine", allow_none=True)
294 safe_close(llm, "benchmark LLM", allow_none=True)
297def calculate_quality_metrics(
298 system_config: Dict[str, Any],
299 num_examples: int = 2, # Reduced for quicker demo
300 output_dir: Optional[str] = None,
301) -> Dict[str, Any]:
302 """
303 Calculate quality-related metrics for a configuration.
305 Args:
306 system_config: Configuration parameters to evaluate
307 num_examples: Number of benchmark examples to use
308 output_dir: Directory to save results (temporary if None)
310 Returns:
311 Dictionary with quality metrics
312 """
313 # Run quality evaluation
314 quality_results = evaluate_benchmark_quality(
315 system_config=system_config,
316 num_examples=num_examples,
317 output_dir=output_dir,
318 )
320 # Return normalized quality score
321 return {
322 "quality_score": quality_results.get("quality_score", 0.0),
323 "accuracy": quality_results.get("accuracy", 0.0),
324 }
327def calculate_speed_metrics(
328 system_config: Dict[str, Any],
329 query: str = "test query",
330 search_tool: Optional[str] = None,
331 num_runs: int = 1,
332) -> Dict[str, Any]:
333 """
334 Calculate speed-related metrics for a configuration.
336 Args:
337 system_config: Configuration parameters to evaluate
338 query: Query to use for timing tests
339 search_tool: Override search tool
340 num_runs: Number of runs to average time over
342 Returns:
343 Dictionary with speed metrics
344 """
345 # Run speed measurement
346 speed_results = measure_execution_time(
347 system_config=system_config,
348 query=query,
349 search_tool=search_tool,
350 num_runs=num_runs,
351 )
353 # Return normalized speed score
354 return {
355 "speed_score": speed_results.get("speed_score", 0.0),
356 "average_time": speed_results.get("average_time", 0.0),
357 }
360def calculate_resource_metrics(
361 system_config: Dict[str, Any],
362 query: str = "test query",
363 search_tool: Optional[str] = None,
364) -> Dict[str, Any]:
365 """
366 Calculate resource usage metrics for a configuration.
368 Args:
369 system_config: Configuration parameters to evaluate
370 query: Query to use for resource tests
371 search_tool: Override search tool
373 Returns:
374 Dictionary with resource metrics
375 """
376 # This is a simplified version - in a real implementation,
377 # you would measure memory usage, API call counts, etc.
379 # For now, we'll use a heuristic based on configuration values
380 iterations = system_config.get("iterations", 2)
381 questions = system_config.get("questions_per_iteration", 2)
382 max_results = system_config.get("max_results", 50)
384 # Simple heuristic: more iterations, questions, and results = more resources
385 complexity = iterations * questions * (max_results / 50)
387 # Normalize to 0-1 scale (lower is better)
388 resource_score = 1.0 / (1.0 + (complexity / 4.0))
390 return {
391 "resource_score": resource_score,
392 "estimated_complexity": complexity,
393 }
396def calculate_combined_score(
397 metrics: Dict[str, Dict[str, float]],
398 weights: Optional[Dict[str, float]] = None,
399) -> float:
400 """
401 Calculate a combined optimization score from multiple metrics.
403 Args:
404 metrics: Dictionary of metric categories and their values
405 weights: Dictionary of weights for each metric category
407 Returns:
408 Combined score between 0 and 1
409 """
410 # Default weights if not provided
411 if weights is None:
412 weights = {"quality": 0.6, "speed": 0.3, "resource": 0.1}
414 # Normalize weights to sum to 1
415 total_weight = sum(weights.values())
416 if total_weight == 0:
417 return 0.0
419 norm_weights = {k: v / total_weight for k, v in weights.items()}
421 # Calculate weighted score
422 score = 0.0
424 # Quality component
425 if "quality" in metrics and "quality" in norm_weights:
426 quality_score = metrics["quality"].get("quality_score", 0.0)
427 score += quality_score * norm_weights["quality"]
429 # Speed component
430 if "speed" in metrics and "speed" in norm_weights:
431 speed_score = metrics["speed"].get("speed_score", 0.0)
432 score += speed_score * norm_weights["speed"]
434 # Resource component
435 if "resource" in metrics and "resource" in norm_weights:
436 resource_score = metrics["resource"].get("resource_score", 0.0)
437 score += resource_score * norm_weights["resource"]
439 return score