Coverage for src/local_deep_research/news/utils/topic_generator.py: 100%

1""" 

2Topic generation utilities for news items. 

3Uses LLM to extract relevant topics/tags from news content. 

4""" 

5 

6from loguru import logger 

7from typing import List 

8 

9from ...utilities.json_utils import extract_json, get_llm_response_text 

10 

11 

12def generate_topics( 

13 query: str, 

14 findings: str = "", 

15 category: str = "", 

16 max_topics: int = 5, 

17 settings_snapshot=None, 

18) -> List[str]: 

19 """ 

20 Generate relevant topics/tags from news content. 

21 

22 Args: 

23 query: The search query or research question 

24 findings: The research findings/content 

25 category: The news category (if available) 

26 max_topics: Maximum number of topics to generate 

27 settings_snapshot: User settings snapshot threaded to get_llm so 

28 the LLM PEP (evaluate_llm_endpoint) fires under 

29 llm.require_local_endpoint. Mirrors headline_generator.py. 

30 

31 Returns: 

32 List of topic strings 

33 """ 

34 # Try LLM generation first 

35 topics = _generate_with_llm( 

36 query, findings, category, max_topics, settings_snapshot 

37 ) 

38 

39 # No fallback - if LLM fails, mark as missing 

40 if not topics: 

41 topics = ["[Topic generation failed]"] 

42 

43 # Ensure we have valid topics 

44 return _validate_topics(topics, max_topics) 

45 

46 

47def _generate_with_llm( 

48 query: str, 

49 findings: str, 

50 category: str, 

51 max_topics: int, 

52 settings_snapshot=None, 

53) -> List[str]: 

54 """Generate topics using LLM.""" 

55 try: 

56 from ...config.llm_config import get_llm 

57 

58 logger.debug( 

59 f"Topic generation - findings length: {len(findings) if findings else 0}, category: {category}" 

60 ) 

61 

62 # Use the configured model for topic generation 

63 llm = get_llm(temperature=0.5, settings_snapshot=settings_snapshot) 

64 

65 try: 

66 # Prepare context 

67 query_preview = query[:500] if len(query) > 500 else query 

68 findings_preview = ( 

69 findings[:1000] 

70 if findings and len(findings) > 1000 

71 else findings 

72 ) 

73 

74 prompt = f"""Extract relevant topics/tags from this news content. 

75 

76Query: {query_preview} 

77{f"Content: {findings_preview}" if findings_preview else ""} 

78{f"Category: {category}" if category else ""} 

79 

80Generate {max_topics} specific, relevant topics that would help categorize and filter this news item. 

81 

82Requirements: 

83- Each topic should be 1-3 words 

84- Topics should be specific and meaningful 

85- Include geographic regions if mentioned 

86- Include key entities (countries, organizations, people) 

87- Include event types (conflict, economy, disaster, etc.) 

88- Topics should be diverse and cover different aspects 

89 

90Return ONLY a JSON array of topic strings, like: ["Topic 1", "Topic 2", "Topic 3"]""" 

91 

92 response = llm.invoke(prompt) 

93 content = get_llm_response_text(response) 

94 

95 # Try to parse the JSON response 

96 topics = extract_json(content, expected_type=list) 

97 

98 if topics is not None: 

99 # Clean and validate each topic 

100 cleaned_topics = [] 

101 for topic in topics: 

102 if isinstance(topic, str): 

103 cleaned = topic.strip() 

104 if cleaned and len(cleaned) <= 30: # Max topic length 

105 cleaned_topics.append(cleaned) 

106 

107 logger.debug(f"Generated topics: {cleaned_topics}") 

108 return cleaned_topics[:max_topics] 

109 

110 # Try to extract topics from plain text response 

111 logger.debug(f"Failed to parse LLM topics as JSON: {content}") 

112 if "," in content: 

113 topics = [t.strip().strip("\"'") for t in content.split(",")] 

114 return [t for t in topics if t and len(t) <= 30][:max_topics] 

115 finally: 

116 from ...utilities.resource_utils import safe_close 

117 

118 safe_close(llm, "topic LLM") 

119 

120 except Exception as e: 

121 logger.debug(f"LLM topic generation failed: {e}") 

122 

123 return [] 

124 

125 

126def _validate_topics(topics: List[str], max_topics: int) -> List[str]: 

127 """Validate and clean topics.""" 

128 valid_topics = [] 

129 seen = set() 

130 

131 for topic in topics: 

132 if not topic: 

133 continue 

134 

135 # Clean the topic 

136 cleaned = topic.strip() 

137 

138 # Skip if too short or too long 

139 if len(cleaned) < 2 or len(cleaned) > 30: 

140 continue 

141 

142 # Skip duplicates (case-insensitive) 

143 normalized = cleaned.lower() 

144 if normalized in seen: 

145 continue 

146 seen.add(normalized) 

147 

148 # Convert to lowercase as djpetti suggested 

149 valid_topics.append(normalized) 

150 

151 if len(valid_topics) >= max_topics: 

152 break 

153 

154 # Don't add default topics - show what actually happened 

155 if not valid_topics: 

156 valid_topics = ["[No valid topics]"] 

157 

158 return valid_topics