Coverage for src/qdrant_loader_mcp_server/search/hybrid/components/helpers.py: 96%

1from __future__ import annotations 

2 

3from typing import Any 

4 

5 

6async def get_embedding(vector_search_service: Any, text: str) -> list[float]: 

7 return await vector_search_service.get_embedding(text) 

8 

9 

10def analyze_query(query_processor: Any, query: str) -> dict[str, Any]: 

11 return query_processor.analyze_query(query) 

12 

13 

14async def expand_query(query_processor: Any, query: str) -> str: 

15 return await query_processor.expand_query(query) 

16 

17 

18async def expand_query_aggressive(query_processor: Any, query: str) -> str: 

19 return await query_processor.expand_query_aggressive(query) 

20 

21 

22async def vector_search( 

23 vector_search_service: Any, query: str, limit: int, project_ids: list[str] | None 

24) -> list[dict[str, Any]]: 

25 return await vector_search_service.vector_search(query, limit, project_ids) 

26 

27 

28async def keyword_search( 

29 keyword_search_service: Any, query: str, limit: int, project_ids: list[str] | None 

30) -> list[dict[str, Any]]: 

31 return await keyword_search_service.keyword_search(query, limit, project_ids) 

32 

33 

34async def combine_results( 

35 result_combiner: Any, 

36 engine_min_score: float, 

37 vector_results: list[dict[str, Any]], 

38 keyword_results: list[dict[str, Any]], 

39 query_context: dict[str, Any], 

40 limit: int, 

41 source_types: list[str] | None, 

42 project_ids: list[str] | None, 

43): 

44 previous_min_score = getattr(result_combiner, "min_score", None) 

45 should_override = ( 

46 previous_min_score is None or previous_min_score > engine_min_score 

47 ) 

48 if should_override: 

49 result_combiner.min_score = engine_min_score 

50 try: 

51 return await result_combiner.combine_results( 

52 vector_results, 

53 keyword_results, 

54 query_context, 

55 limit, 

56 source_types, 

57 project_ids, 

58 ) 

59 finally: 

60 if should_override: 

61 result_combiner.min_score = previous_min_score 

62 

63 

64def build_filter(vector_search_service: Any, project_ids: list[str] | None) -> Any: 

65 # Use public API on the service to avoid relying on private methods 

66 return vector_search_service.build_filter(project_ids)