Coverage for src/local_deep_research/advanced_search_system/parallel_search.py: 100%

1"""Parallel search execution helper. 

2 

3Provides :func:`run_parallel_searches`, a small helper that runs a sequence 

4of search queries concurrently using :class:`~concurrent.futures.ThreadPoolExecutor` 

5while preserving the research context expected by worker threads. 

6 

7The helper is intentionally generic: 

8 

9* ``search_fn`` is a callable accepting the query string and returning 

10 whatever per-query payload the caller needs (a list of results, a dict 

11 with metadata, etc.). Callers wrap it with context-preserving 

12 decorators (e.g. :func:`preserve_research_context`) before passing it 

13 in; this helper stays focused on the concurrency concern. 

14 

15Returns a list of ``(query, payload)`` tuples in completion order. 

16Callers that need a question-keyed dict build it from this list; callers 

17that only need a flat list of results flatten it directly. 

18""" 

19 

20from __future__ import annotations 

21 

22import concurrent.futures 

23from typing import Callable, List, Optional, Tuple, TypeVar 

24 

25from loguru import logger 

26 

27T = TypeVar("T") 

28 

29 

30def run_parallel_searches( 

31 queries: List[str], 

32 search_fn: Callable[[str], T], 

33 max_workers: Optional[int] = None, 

34) -> List[Tuple[str, T]]: 

35 """Run ``search_fn`` for each query in parallel. 

36 

37 Args: 

38 queries: Queries to search. If empty, returns an empty list 

39 immediately (and logs a warning). 

40 search_fn: Callable invoked as ``search_fn(query)`` inside a worker 

41 thread. Callers are responsible for wrapping it with any 

42 context-preserving decorators (e.g. 

43 :func:`preserve_research_context`) before passing it in, and 

44 for their own error handling (the callable should never raise 

45 — return an empty payload on failure instead, matching the 

46 pre-existing contract of the strategies this was extracted 

47 from). 

48 max_workers: Size of the thread pool. Defaults to ``len(queries)`` 

49 when ``None``, matching the historical behavior of the 

50 source-based, focused-iteration, and progressive strategies. 

51 

52 Returns: 

53 A list of ``(query, payload)`` tuples in completion order. Each 

54 ``payload`` is whatever ``search_fn`` returned for that query. 

55 """ 

56 if not queries: 

57 logger.warning("No queries provided for parallel search") 

58 return [] 

59 

60 if max_workers is None: 

61 max_workers = len(queries) 

62 

63 def _worker(query: str) -> Tuple[str, T]: 

64 return (query, search_fn(query)) 

65 

66 with concurrent.futures.ThreadPoolExecutor( 

67 max_workers=max_workers 

68 ) as executor: 

69 futures = [executor.submit(_worker, q) for q in queries] 

70 return [ 

71 future.result() 

72 for future in concurrent.futures.as_completed(futures) 

73 ]