Coverage for src/local_deep_research/news/utils/topic_generator.py: 97%
88 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Topic generation utilities for news items.
3Uses LLM to extract relevant topics/tags from news content.
4"""
6from loguru import logger
7from typing import List, Optional
9from ...utilities.json_utils import extract_json, get_llm_response_text
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.
22 The synchronous entry point uses ``invoke`` without running the async
23 core on a temporary event loop. Callers already on an event loop can
24 await the corresponding ``_async`` entry point. Resource cleanup uses
25 the shared close helper; it is separate from invocation dispatch.
27 Args:
28 query: The search query or research question
29 findings: The research findings/content
30 category: The news category (if available)
31 max_topics: Maximum number of topics to generate
32 settings_snapshot: User settings snapshot threaded to get_llm so
33 the LLM PEP (evaluate_llm_endpoint) fires under
34 llm.require_local_endpoint. Without it get_llm takes its
35 snapshot-less path, which fails closed with
36 PolicyDeniedError for any non-local provider. Mirrors
37 headline_generator.py.
39 Returns:
40 List of topic strings
41 """
42 # Try LLM generation first
43 topics = _generate_with_llm(
44 query, findings, category, max_topics, settings_snapshot
45 )
47 return _finalize_topics(topics, max_topics)
50async def generate_topics_async(
51 query: str,
52 findings: str = "",
53 category: str = "",
54 max_topics: int = 5,
55 settings_snapshot=None,
56) -> List[str]:
57 """Async core of generate_topics: one awaited ainvoke (#5854).
59 For callers that already run on an event loop.
60 """
61 topics = await _generate_with_llm_async(
62 query, findings, category, max_topics, settings_snapshot
63 )
65 return _finalize_topics(topics, max_topics)
68def _finalize_topics(topics: List[str], max_topics: int) -> List[str]:
69 """Apply the failure sentinel and validation shared by both paths."""
70 # No fallback - if LLM fails, mark as missing
71 if not topics:
72 topics = ["[Topic generation failed]"]
74 # Ensure we have valid topics
75 return _validate_topics(topics, max_topics)
78def _build_topic_prompt(
79 query: str,
80 findings: str,
81 category: str,
82 max_topics: int,
83) -> str:
84 """Build the topic-extraction prompt."""
85 # Prepare context
86 query_preview = query[:500] if len(query) > 500 else query
87 findings_preview = (
88 findings[:1000] if findings and len(findings) > 1000 else findings
89 )
91 return f"""Extract relevant topics/tags from this news content.
93Query: {query_preview}
94{f"Content: {findings_preview}" if findings_preview else ""}
95{f"Category: {category}" if category else ""}
97Generate {max_topics} specific, relevant topics that would help categorize and filter this news item.
99Requirements:
100- Each topic should be 1-3 words
101- Topics should be specific and meaningful
102- Include geographic regions if mentioned
103- Include key entities (countries, organizations, people)
104- Include event types (conflict, economy, disaster, etc.)
105- Topics should be diverse and cover different aspects
107Return ONLY a JSON array of topic strings, like: ["Topic 1", "Topic 2", "Topic 3"]"""
110def _parse_topics_response(
111 content: str, max_topics: int
112) -> Optional[List[str]]:
113 """Parse the LLM's topics response; None when nothing usable came back."""
114 # Try to parse the JSON response
115 topics = extract_json(content, expected_type=list)
117 if topics is not None:
118 # Clean and validate each topic
119 cleaned_topics = []
120 for topic in topics:
121 if isinstance(topic, str):
122 cleaned = topic.strip()
123 if cleaned and len(cleaned) <= 30: # Max topic length
124 cleaned_topics.append(cleaned)
126 logger.debug(f"Generated topics: {cleaned_topics}")
127 return cleaned_topics[:max_topics]
129 # Try to extract topics from plain text response
130 logger.debug(f"Failed to parse LLM topics as JSON: {content}")
131 if "," in content:
132 topics = [t.strip().strip("\"'") for t in content.split(",")]
133 return [t for t in topics if t and len(t) <= 30][:max_topics]
135 return None
138def _generate_with_llm(
139 query: str,
140 findings: str,
141 category: str,
142 max_topics: int,
143 settings_snapshot=None,
144) -> List[str]:
145 """Generate topics using LLM (sync path)."""
146 try:
147 from ...config.llm_config import get_llm
149 # (#6938: the category can come from a research request's
150 # metadata, so it is previewed rather than f-stringed, and only
151 # when a DEBUG sink takes the line.)
152 from ...web.routers.notes import _log_value_preview
154 logger.opt(lazy=True).debug(
155 "Topic generation - findings length: {}, category: {}",
156 lambda: len(findings) if findings else 0,
157 lambda: _log_value_preview(category),
158 )
160 # Use the configured model for topic generation
161 llm = get_llm(temperature=0.5, settings_snapshot=settings_snapshot)
163 try:
164 prompt = _build_topic_prompt(query, findings, category, max_topics)
166 response = llm.invoke(prompt)
167 content = get_llm_response_text(response)
169 parsed = _parse_topics_response(content, max_topics)
170 if parsed is not None:
171 return parsed
172 finally:
173 from ...utilities.resource_utils import safe_close
175 safe_close(llm, "topic LLM")
177 except Exception as e:
178 logger.debug(f"LLM topic generation failed: {e}")
180 return []
183async def _generate_with_llm_async(
184 query: str,
185 findings: str,
186 category: str,
187 max_topics: int,
188 settings_snapshot=None,
189) -> List[str]:
190 """Generate topics using LLM (async path).
192 Mirrors :func:`_generate_with_llm`; only the LLM call line differs.
193 """
194 try:
195 from ...config.llm_config import get_llm
197 # (#6938: the category can come from a research request's
198 # metadata, so it is previewed rather than f-stringed, and only
199 # when a DEBUG sink takes the line.)
200 from ...web.routers.notes import _log_value_preview
202 logger.opt(lazy=True).debug(
203 "Topic generation - findings length: {}, category: {}",
204 lambda: len(findings) if findings else 0,
205 lambda: _log_value_preview(category),
206 )
208 # Use the configured model for topic generation
209 llm = get_llm(temperature=0.5, settings_snapshot=settings_snapshot)
211 try:
212 prompt = _build_topic_prompt(query, findings, category, max_topics)
214 response = await llm.ainvoke(prompt)
215 content = get_llm_response_text(response)
217 parsed = _parse_topics_response(content, max_topics)
218 if parsed is not None: 218 ↛ 221line 218 didn't jump to line 221 because the condition on line 218 was always true
219 return parsed
220 finally:
221 from ...utilities.resource_utils import safe_close
223 safe_close(llm, "topic LLM")
225 except Exception as e:
226 logger.debug(f"LLM topic generation failed: {e}")
228 return []
231def _validate_topics(topics: List[str], max_topics: int) -> List[str]:
232 """Validate and clean topics."""
233 valid_topics = []
234 seen = set()
236 for topic in topics:
237 if not topic:
238 continue
240 # Clean the topic
241 cleaned = topic.strip()
243 # Skip if too short or too long
244 if len(cleaned) < 2 or len(cleaned) > 30:
245 continue
247 # Skip duplicates (case-insensitive)
248 normalized = cleaned.lower()
249 if normalized in seen:
250 continue
251 seen.add(normalized)
253 # Convert to lowercase as djpetti suggested
254 valid_topics.append(normalized)
256 if len(valid_topics) >= max_topics:
257 break
259 # Don't add default topics - show what actually happened
260 if not valid_topics:
261 valid_topics = ["[No valid topics]"]
263 return valid_topics