Coverage for src/local_deep_research/advanced_search_system/summarization/base.py: 100%

1"""Abstract base for LLM-driven summarizers.""" 

2 

3from abc import ABC, abstractmethod 

4 

5from langchain_core.language_models.chat_models import BaseChatModel 

6from loguru import logger 

7 

8from ...utilities.search_utilities import remove_think_tags 

9from ...utilities.llm_utils import invoke_llm_sync 

10 

11 

12class BaseSummarizer(ABC): 

13 """Common machinery for invoking an LLM to summarize text. 

14 

15 Subclasses provide the prompt via :meth:`_build_prompt`. The base class 

16 handles the model invocation, think-tag stripping, and length truncation. 

17 

18 On LLM-invocation failure (network error, rate limit, malformed response) 

19 :meth:`summarize` returns an empty string. This is a deliberate choice for 

20 callers that aggregate the summary into a larger context update — losing 

21 one summary turn should not abort the rest of the update. 

22 """ 

23 

24 INPUT_TRUNCATE_CHARS = 8000 

25 

26 def __init__( 

27 self, 

28 model: BaseChatModel, 

29 max_sentences: int = 3, 

30 max_chars: int = 300, 

31 ): 

32 self.model = model 

33 self.max_sentences = max_sentences 

34 self.max_chars = max_chars 

35 

36 def summarize(self, content: str) -> str: 

37 if not content: 

38 return "" 

39 

40 prompt = self._build_prompt(content[: self.INPUT_TRUNCATE_CHARS]) 

41 

42 try: 

43 response = invoke_llm_sync(self.model, prompt) 

44 except Exception: 

45 logger.opt(exception=True).debug("LLM summarization failed") 

46 return "" 

47 

48 summary = remove_think_tags(str(response.content)).strip() 

49 return self._truncate(summary) 

50 

51 @abstractmethod 

52 def _build_prompt(self, content: str) -> str: 

53 """Return the prompt sent to the LLM for ``content``.""" 

54 

55 def _truncate(self, summary: str) -> str: 

56 if len(summary) > self.max_chars: 

57 return summary[: self.max_chars].rstrip() + "..." 

58 return summary