Coverage for src/local_deep_research/benchmarks/datasets/xbench_deepsearch.py: 94%
91 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +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 dataset = load_dataset(dataset_path, split="train")
104 # Format for our benchmark system and decrypt
105 formatted_questions = []
106 for item in dataset:
107 # Get the canary key for decryption
108 canary = item.get("canary", "")
110 # Decrypt prompt and answer if they're encrypted
111 prompt = item.get("prompt", "")
112 answer = item.get("answer", "")
114 try:
115 # Try to decrypt if it looks like base64
116 if prompt and all(
117 c
118 in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/="
119 for c in prompt[:100]
120 ):
121 decrypted_prompt = self.xor_decrypt(
122 base64.b64decode(prompt), canary
123 ).decode("utf-8")
124 prompt = decrypted_prompt
126 if answer and all( 126 ↛ 131line 126 didn't jump to line 131 because the condition on line 126 was never true
127 c
128 in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/="
129 for c in answer[:100]
130 ):
131 decrypted_answer = self.xor_decrypt(
132 base64.b64decode(answer), canary
133 ).decode("utf-8")
134 answer = decrypted_answer
135 except Exception:
136 logger.warning("Failed to decrypt item")
138 formatted_item = {
139 "id": item.get("id", f"xbench_{len(formatted_questions)}"),
140 "problem": prompt,
141 "answer": answer,
142 "reference_steps": item.get("reference_steps", ""),
143 "canary": canary,
144 }
145 formatted_questions.append(formatted_item)
147 logger.info(
148 f"Loaded {len(formatted_questions)} questions from xbench-DeepSearch"
149 )
150 return formatted_questions
152 except Exception:
153 logger.warning("Failed to load via datasets library")
154 logger.info("Falling back to direct download")
155 return self._load_from_url()
157 def _load_from_url(self) -> List[Dict[str, Any]]:
158 """Load the full dataset directly from URL without datasets library.
160 Returns:
161 List of questions from xbench-DeepSearch
162 """
163 import pandas as pd
165 try:
166 # Direct URL to the CSV file on Hugging Face
167 url = "https://huggingface.co/datasets/xbench/DeepSearch/resolve/main/data/train-00000-of-00001.parquet"
169 logger.info(f"Downloading xbench-DeepSearch from {url}")
170 df = pd.read_parquet(url)
172 # Convert to list of dicts and decrypt
173 questions = []
174 for _, row in df.iterrows():
175 # Get the canary key for decryption
176 canary = row.get("canary", "")
178 # Decrypt prompt and answer
179 prompt = row.get("prompt", "")
180 answer = row.get("answer", "")
182 try:
183 if prompt and all(
184 c
185 in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/="
186 for c in prompt[:100]
187 ):
188 decrypted_prompt = self.xor_decrypt(
189 base64.b64decode(prompt), canary
190 ).decode("utf-8")
191 prompt = decrypted_prompt
193 if answer and all( 193 ↛ 198line 193 didn't jump to line 198 because the condition on line 193 was never true
194 c
195 in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/="
196 for c in answer[:100]
197 ):
198 decrypted_answer = self.xor_decrypt(
199 base64.b64decode(answer), canary
200 ).decode("utf-8")
201 answer = decrypted_answer
202 except Exception:
203 logger.warning("Failed to decrypt item")
205 questions.append(
206 {
207 "id": row.get("id", f"xbench_{len(questions)}"),
208 "problem": prompt,
209 "answer": answer,
210 "reference_steps": row.get("reference_steps", ""),
211 "canary": canary,
212 }
213 )
215 logger.info(
216 f"Loaded {len(questions)} questions via direct download"
217 )
218 return questions
220 except Exception:
221 logger.exception("Failed to load dataset")
222 return []
224 def process_example(self, example: Dict[str, Any]) -> Dict[str, Any]:
225 """Process a single example from the dataset.
227 xbench-DeepSearch questions are designed for deep research evaluation.
228 """
229 processed = dict(example)
231 # Add evaluation metadata
232 processed["requires_deep_search"] = True
233 processed["expected_iterations"] = (
234 4 # Deep search questions need multiple iterations
235 )
237 # Evaluation criteria for research questions
238 processed["evaluation_criteria"] = {
239 "accuracy": 0.4,
240 "completeness": 0.3,
241 "reasoning": 0.2,
242 "sources": 0.1, # Credit for citing sources
243 }
245 return processed