Coverage for src/local_deep_research/benchmarks/datasets/xbench_deepsearch.py: 94%
91 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2xbench-DeepSearch dataset implementation.
4This module provides a class for the xbench-DeepSearch benchmark dataset,
5which evaluates deep research and search capabilities.
6"""
8import base64
9from typing import Any, Dict, List
10from loguru import logger
11from .base import BenchmarkDataset
14class XBenchDeepSearchDataset(BenchmarkDataset):
15 """xbench-DeepSearch benchmark dataset for deep research evaluation."""
17 @staticmethod
18 def xor_decrypt(data: bytes, key: str) -> bytes:
19 """XOR decrypt data with a key."""
20 key_bytes = key.encode("utf-8")
21 key_length = len(key_bytes)
22 return bytes(
23 [data[i] ^ key_bytes[i % key_length] for i in range(len(data))]
24 )
26 @classmethod
27 def get_dataset_info(cls) -> Dict[str, str]:
28 """Get basic information about the dataset."""
29 return {
30 "id": "xbench_deepsearch",
31 "name": "xbench-DeepSearch",
32 "description": "Deep research and search capability evaluation (100 questions)",
33 "url": "https://huggingface.co/datasets/xbench/DeepSearch",
34 }
36 @classmethod
37 def get_default_dataset_path(cls) -> str:
38 """Get the default path for the dataset."""
39 return "xbench/DeepSearch" # Hugging Face dataset identifier
41 def load(self) -> List[Dict[str, Any]]:
42 """Override load to handle HuggingFace datasets directly.
44 Keeps the base-class contract: callers configure sampling via the
45 constructor (``num_examples``/``seed``) and call ``load()`` with no
46 arguments, as ``DatasetRegistry.load_dataset()`` does.
48 Returns:
49 List of processed dataset examples
50 """
51 import random
53 if self._is_loaded:
54 return self.examples
56 # Load the data
57 data = self.load_data(self.dataset_path)
59 # Sample if requested
60 if self.num_examples and len(data) > self.num_examples:
61 # Security: seeded random for reproducible benchmark sampling, not security-sensitive
62 random.seed(self.seed)
63 data = random.sample(data, self.num_examples)
65 # Process each example
66 self.examples = [self.process_example(item) for item in data]
67 self._is_loaded = True
68 return self.examples
70 def load_data(
71 self,
72 dataset_path: str = None,
73 ) -> List[Dict[str, Any]]:
74 """Load the full xbench-DeepSearch dataset from Hugging Face.
76 Sampling happens in load(), driven by the constructor's
77 num_examples/seed.
79 Args:
80 dataset_path: Path to dataset (defaults to Hugging Face)
82 Returns:
83 List of questions from xbench-DeepSearch
84 """
85 try:
86 from datasets import load_dataset
87 except ImportError:
88 logger.exception(
89 "datasets library not installed. Run: pip install datasets"
90 )
91 # Fallback to direct download
92 return self._load_from_url()
94 dataset_path = dataset_path or self.get_default_dataset_path()
96 try:
97 logger.info(
98 f"Loading xbench-DeepSearch dataset from {dataset_path}"
99 )
101 # Load the dataset from Hugging Face (no authentication needed)
102 # Operator-selected benchmark records are data inputs; datasets >=5 disables remote loading scripts.
103 dataset = load_dataset(dataset_path, split="train") # nosec B615
105 # Format for our benchmark system and decrypt
106 formatted_questions = []
107 for item in dataset:
108 # Get the canary key for decryption
109 canary = item.get("canary", "")
111 # Decrypt prompt and answer if they're encrypted
112 prompt = item.get("prompt", "")
113 answer = item.get("answer", "")
115 try:
116 # Try to decrypt if it looks like base64
117 if prompt and all(
118 c
119 in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/="
120 for c in prompt[:100]
121 ):
122 decrypted_prompt = self.xor_decrypt(
123 base64.b64decode(prompt), canary
124 ).decode("utf-8")
125 prompt = decrypted_prompt
127 if answer and all( 127 ↛ 132line 127 didn't jump to line 132 because the condition on line 127 was never true
128 c
129 in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/="
130 for c in answer[:100]
131 ):
132 decrypted_answer = self.xor_decrypt(
133 base64.b64decode(answer), canary
134 ).decode("utf-8")
135 answer = decrypted_answer
136 except Exception:
137 logger.warning("Failed to decrypt item")
139 formatted_item = {
140 "id": item.get("id", f"xbench_{len(formatted_questions)}"),
141 "problem": prompt,
142 "answer": answer,
143 "reference_steps": item.get("reference_steps", ""),
144 "canary": canary,
145 }
146 formatted_questions.append(formatted_item)
148 logger.info(
149 f"Loaded {len(formatted_questions)} questions from xbench-DeepSearch"
150 )
151 return formatted_questions
153 except Exception:
154 logger.warning("Failed to load via datasets library")
155 logger.info("Falling back to direct download")
156 return self._load_from_url()
158 def _load_from_url(self) -> List[Dict[str, Any]]:
159 """Load the full dataset directly from URL without datasets library.
161 Returns:
162 List of questions from xbench-DeepSearch
163 """
164 import pandas as pd
166 try:
167 # Direct URL to the CSV file on Hugging Face
168 url = "https://huggingface.co/datasets/xbench/DeepSearch/resolve/main/data/train-00000-of-00001.parquet"
170 logger.info(f"Downloading xbench-DeepSearch from {url}")
171 df = pd.read_parquet(url)
173 # Convert to list of dicts and decrypt
174 questions = []
175 for _, row in df.iterrows():
176 # Get the canary key for decryption
177 canary = row.get("canary", "")
179 # Decrypt prompt and answer
180 prompt = row.get("prompt", "")
181 answer = row.get("answer", "")
183 try:
184 if prompt and all(
185 c
186 in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/="
187 for c in prompt[:100]
188 ):
189 decrypted_prompt = self.xor_decrypt(
190 base64.b64decode(prompt), canary
191 ).decode("utf-8")
192 prompt = decrypted_prompt
194 if answer and all( 194 ↛ 199line 194 didn't jump to line 199 because the condition on line 194 was never true
195 c
196 in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/="
197 for c in answer[:100]
198 ):
199 decrypted_answer = self.xor_decrypt(
200 base64.b64decode(answer), canary
201 ).decode("utf-8")
202 answer = decrypted_answer
203 except Exception:
204 logger.warning("Failed to decrypt item")
206 questions.append(
207 {
208 "id": row.get("id", f"xbench_{len(questions)}"),
209 "problem": prompt,
210 "answer": answer,
211 "reference_steps": row.get("reference_steps", ""),
212 "canary": canary,
213 }
214 )
216 logger.info(
217 f"Loaded {len(questions)} questions via direct download"
218 )
219 return questions
221 except Exception:
222 logger.exception("Failed to load dataset")
223 return []
225 def process_example(self, example: Dict[str, Any]) -> Dict[str, Any]:
226 """Process a single example from the dataset.
228 xbench-DeepSearch questions are designed for deep research evaluation.
229 """
230 processed = dict(example)
232 # Add evaluation metadata
233 processed["requires_deep_search"] = True
234 processed["expected_iterations"] = (
235 4 # Deep search questions need multiple iterations
236 )
238 # Evaluation criteria for research questions
239 processed["evaluation_criteria"] = {
240 "accuracy": 0.4,
241 "completeness": 0.3,
242 "reasoning": 0.2,
243 "sources": 0.1, # Credit for citing sources
244 }
246 return processed