Coverage for src/local_deep_research/news/utils/headline_generator.py: 97%
57 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Headline generation utilities for news items.
3Uses LLM to generate concise, meaningful headlines from long queries and findings.
4"""
6from typing import Any, Optional
7from loguru import logger
9from ...utilities.json_utils import get_llm_response_text
12def generate_headline(
13 query: str,
14 findings: str = "",
15 max_length: int = 100,
16 settings_snapshot: Optional[dict] = None,
17) -> str:
18 """
19 Generate a concise headline from a query and optional findings.
21 The synchronous entry point uses ``invoke`` without running the async
22 core on a temporary event loop. Callers already on an event loop can
23 await the corresponding ``_async`` entry point. Resource cleanup uses
24 the shared close helper; it is separate from invocation dispatch.
26 Args:
27 query: The search query or research question
28 findings: Optional findings/content to help generate better headline
29 max_length: Maximum length for the headline
30 settings_snapshot: Optional settings snapshot so the LLM call
31 picks up the active egress policy. Background callers
32 should pass this through; without it ``get_llm`` takes its
33 snapshot-less path, which fails closed with
34 ``PolicyDeniedError`` for any non-local provider, so
35 headline generation fails rather than reaching a cloud LLM.
37 Returns:
38 A concise headline string
39 """
40 # Always try LLM generation first for dynamic headlines based on actual content
41 llm_headline = _generate_with_llm(
42 query, findings, max_length, settings_snapshot
43 )
44 return _finalize_headline(llm_headline)
47async def generate_headline_async(
48 query: str,
49 findings: str = "",
50 max_length: int = 100,
51 settings_snapshot: Optional[dict] = None,
52) -> str:
53 """Async core of generate_headline: one awaited ainvoke (#5854).
55 For callers that already run on an event loop. Applies the same
56 failure sentinel as :func:`generate_headline` via
57 :func:`_finalize_headline`.
58 """
59 llm_headline = await _generate_with_llm_async(
60 query, findings, max_length, settings_snapshot
61 )
62 return _finalize_headline(llm_headline)
65def _finalize_headline(llm_headline: Optional[str]) -> str:
66 """Apply the failure sentinel shared by both generate_headline paths."""
67 # No fallback - if LLM fails, indicate failure
68 if llm_headline:
69 return llm_headline
71 return "[Headline generation failed]"
74def _build_headline_prompt(findings: str) -> Optional[str]:
75 """Build the headline prompt, or None when there is nothing to summarize."""
76 # Focus only on the findings/report content, not the query
77 if not findings:
78 logger.debug("No findings provided for headline generation")
79 return None
81 # Use the COMPLETE findings - no character limit
82 findings_preview = findings
83 logger.debug(f"Generating headline with {len(findings)} chars of findings")
85 return f"""Generate a comprehensive news headline that captures the key events from the research report below.
87Research Findings:
88{findings_preview}
90Requirements:
91- Include MULTIPLE major events if several important things happened (e.g., "Earthquake Strikes California While Wildfires Rage; Global Markets Tumble Amid Political Tensions")
92- Capture as much important information as possible in the headline
93- Be specific about locations, impacts, and key details
94- Professional news headline style but can be longer to include more information
95- Focus on the most impactful findings from the report
96- Use semicolons or commas to separate multiple major events
97- No quotes or punctuation at start/end
98- Base the headline ONLY on the actual findings in the report
100Generate only the headline text, nothing else."""
103def _clean_headline(response: Any) -> Optional[str]:
104 """Clean and validate the headline the LLM returned."""
105 headline = get_llm_response_text(response).strip()
107 # Clean up the generated headline
108 headline = headline.strip("\"'.,!?")
110 # Validate the headline
111 if headline:
112 logger.debug(f"Generated headline: {headline}")
113 return headline
115 return None
118def _generate_with_llm(
119 query: str,
120 findings: str,
121 max_length: int,
122 settings_snapshot: Optional[dict] = None,
123) -> Optional[str]:
124 """Generate headline using LLM (sync path)."""
125 try:
126 from ...config.llm_config import get_llm
128 # Use the configured model for headline generation
129 llm = get_llm(temperature=0.3, settings_snapshot=settings_snapshot)
131 try:
132 prompt = _build_headline_prompt(findings)
133 if prompt is None:
134 return None
136 response = llm.invoke(prompt)
137 return _clean_headline(response)
138 finally:
139 from ...utilities.resource_utils import safe_close
141 safe_close(llm, "headline LLM")
143 except Exception as e:
144 logger.debug(f"LLM headline generation failed: {e}")
146 return None
149async def _generate_with_llm_async(
150 query: str,
151 findings: str,
152 max_length: int,
153 settings_snapshot: Optional[dict] = None,
154) -> Optional[str]:
155 """Generate headline using LLM (async path).
157 Mirrors :func:`_generate_with_llm`; only the LLM call line differs.
158 """
159 try:
160 from ...config.llm_config import get_llm
162 # Use the configured model for headline generation
163 llm = get_llm(temperature=0.3, settings_snapshot=settings_snapshot)
165 try:
166 prompt = _build_headline_prompt(findings)
167 if prompt is None: 167 ↛ 168line 167 didn't jump to line 168 because the condition on line 167 was never true
168 return None
170 response = await llm.ainvoke(prompt)
171 return _clean_headline(response)
172 finally:
173 from ...utilities.resource_utils import safe_close
175 safe_close(llm, "headline LLM")
177 except Exception as e:
178 logger.debug(f"LLM headline generation failed: {e}")
180 return None