Coverage for src/qdrant_loader_mcp_server/search/hybrid/components/boosting.py: 78%

1from __future__ import annotations 

2 

3import logging 

4from collections.abc import Iterable 

5from typing import Any 

6 

7 

8class ResultBooster: 

9 """Apply simple multiplicative boosts using a provided function. 

10 

11 The booster accepts a callable that returns a multiplier for a given result. 

12 Default behavior is neutral (multiplier=1.0). 

13 """ 

14 

15 def __init__(self, boost_fn: callable | None = None): 

16 self.boost_fn = boost_fn or (lambda _r: 1.0) 

17 

18 def apply(self, results: Iterable[Any]) -> list[Any]: 

19 """Apply boost multipliers to result scores in-place. 

20 

21 - Mutates: updates each item's `score` attribute in-place when present and numeric. 

22 - Errors: invalid multipliers or score operations are logged and skipped. 

23 """ 

24 logger = logging.getLogger(__name__) 

25 boosted: list[Any] = [] 

26 for r in results: 

27 try: 

28 multiplier = float(self.boost_fn(r)) 

29 except (TypeError, ValueError) as exc: 

30 logger.warning("ResultBooster: invalid multiplier for %r: %s", r, exc) 

31 multiplier = 1.0 

32 

33 if hasattr(r, "score") and isinstance(r.score, int | float): 

34 try: 

35 r.score = float(r.score) * multiplier 

36 except (TypeError, ValueError) as exc: 

37 logger.warning( 

38 "ResultBooster: failed to apply boost for %r: %s", r, exc 

39 ) 

40 boosted.append(r) 

41 return boosted