Coverage for src/local_deep_research/chat/context.py: 89%
106 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""
2ChatContextManager - Custom context building for multi-turn conversations.
4This is DIFFERENT from FollowUpResearchService (single parent-child context).
5Multi-turn chat requires:
6- Rolling window of recent messages
7- Accumulated findings across conversation
8- Source deduplication across turns
9- Context summarization for long conversations
10"""
12from typing import Dict, Any, List, Optional
14from loguru import logger
16from ..config.thread_settings import get_setting_from_snapshot
19class ChatContextManager:
20 """
21 Build context from multi-turn conversation history.
23 Handles: rolling window, summarization, context accumulation.
24 Different from follow-up: accumulates from MULTIPLE previous turns.
25 """
27 MAX_FINDINGS_TO_INCLUDE = 5 # Recent findings to include
29 # Limits for the query-focused conversation summary that becomes the
30 # follow-up prompt's "previous findings" block.
31 CONTEXT_SUMMARY_MAX_SENTENCES = 8
32 CONTEXT_SUMMARY_MAX_CHARS = 2000
33 # Transcript char budget kept below BaseSummarizer.INPUT_TRUNCATE_CHARS
34 # (8000) so the summarizer's own truncation never has to drop the most
35 # recent turns — we trim oldest-first ourselves below.
36 CONTEXT_INPUT_CHAR_BUDGET = 7500
38 # Default for the chat.followup_context_mode setting (summary | raw |
39 # full | none) — what prior context a follow-up turn receives.
40 DEFAULT_FOLLOWUP_CONTEXT_MODE = "summary"
42 def __init__(
43 self,
44 session_id: str,
45 messages: List[Dict[str, Any]],
46 accumulated_context: Optional[Dict[str, Any]] = None,
47 settings_snapshot: Optional[Dict[str, Any]] = None,
48 ):
49 """
50 Initialize context manager.
52 Args:
53 session_id: Chat session ID
54 messages: List of message dictionaries with role, content, etc.
55 accumulated_context: Previously accumulated context from session
56 settings_snapshot: Optional settings to override class-level defaults
57 """
58 self.session_id = session_id
59 # chat_messages no longer contains step rows (they live in
60 # chat_progress_steps), but get_session_messages MERGES both for
61 # client rendering. Filter out steps + non-dict entries here so
62 # accumulated context only reflects durable conversation turns.
63 self.messages = [
64 msg
65 for msg in (messages or [])
66 if isinstance(msg, dict) and msg.get("message_type") != "step"
67 ]
68 self.accumulated_context = accumulated_context or {}
69 # Used by build_research_context to construct the LLM that produces
70 # the query-focused conversation summary.
71 self.settings_snapshot = settings_snapshot
73 # Override class defaults from settings if provided.
74 # Note: settings_snapshot is the 4th keyword arg; passing it positionally
75 # would bind it to the unused `username` param, silently using defaults.
76 if settings_snapshot:
77 self.MAX_FINDINGS_TO_INCLUDE = get_setting_from_snapshot(
78 "chat.max_findings_to_include",
79 self.MAX_FINDINGS_TO_INCLUDE,
80 settings_snapshot=settings_snapshot,
81 )
83 def build_research_context(self, current_query: str = "") -> Dict[str, Any]:
84 """
85 Build context for the next research query.
87 Args:
88 current_query: The user's new message. On a follow-up turn it is
89 used to focus a summary of the whole prior conversation, which
90 becomes the follow-up prompt's "previous findings". On the
91 first turn there is no prior work to summarize.
93 Returns dict with:
94 - session_id: Current session
95 - original_query: The session's first user message
96 - accumulated_findings / past_findings: Prior work for the follow-up
97 (query-focused summary on follow-ups; empty on the first turn)
98 - key_entities: Important entities mentioned
99 - topics: Topics discussed
100 - is_multi_turn: Whether this is a follow-up
101 """
102 # The follow-up strategy reads "original_query" to anchor the prompt on
103 # the topic that started the conversation; without it the contextual
104 # follow-up loses the original question. Use the session's first user
105 # message as that anchor.
106 original_query = next(
107 (
108 m.get("content", "")
109 for m in self.messages
110 if isinstance(m, dict) and m.get("role") == "user"
111 ),
112 "",
113 )
115 is_multi_turn = any(
116 isinstance(m, dict) and m.get("role") == "assistant"
117 for m in self.messages
118 )
120 findings = self._select_prior_findings(current_query, is_multi_turn)
122 return {
123 "session_id": self.session_id,
124 "original_query": original_query,
125 "accumulated_findings": findings,
126 "past_findings": findings, # Research engine expects this key
127 "key_entities": self._get_key_entities(),
128 "topics": self._get_topics(),
129 "is_multi_turn": is_multi_turn,
130 "turn_count": len(self.messages),
131 }
133 def _select_prior_findings(
134 self, current_query: str, is_multi_turn: bool
135 ) -> str:
136 """Pick the follow-up's "previous findings" per chat.followup_context_mode.
138 Modes:
139 - ``summary`` (default): query-focused LLM summary of the conversation
140 - ``raw``: recent research findings, truncated
141 - ``full``: the entire conversation transcript
142 - ``none``: no prior findings
144 Only follow-up turns carry prior work; the first turn returns "".
145 """
146 if not is_multi_turn:
147 return ""
149 mode = self.DEFAULT_FOLLOWUP_CONTEXT_MODE
150 if self.settings_snapshot:
151 mode = get_setting_from_snapshot(
152 "chat.followup_context_mode",
153 mode,
154 settings_snapshot=self.settings_snapshot,
155 )
157 if mode == "none":
158 findings = ""
159 elif mode == "raw":
160 findings = self._extract_findings_from_history()
161 elif mode == "full":
162 findings = self._build_conversation_text()
163 elif current_query:
164 # "summary" with a question to focus on.
165 findings = self._summarize_prior_work(current_query)
166 else:
167 # "summary" with no question (e.g. a no-arg build_research_context
168 # call): fall back to raw recent findings.
169 findings = self._extract_findings_from_history()
171 # Observability: the summary path is otherwise silent (no token-counter
172 # entry, since get_llm runs without a research_id), so a follow-up's
173 # prior-context build looked like an unexplained pause. One line per
174 # follow-up turn records which mode ran and how much context it built.
175 logger.info(
176 "Chat follow-up prior context: mode={}, {} chars",
177 mode,
178 len(findings),
179 )
180 return findings
182 def _build_conversation_text(self) -> str:
183 """Render the prior conversation (both roles) as a plain transcript.
185 Trims oldest-first to ``CONTEXT_INPUT_CHAR_BUDGET`` so the most recent
186 turns survive and the summarizer's input cap never has to truncate.
187 """
188 lines: List[str] = []
189 used = 0
190 for msg in reversed(self.messages):
191 if not isinstance(msg, dict): 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true
192 continue
193 content = (msg.get("content") or "").strip()
194 if not content: 194 ↛ 195line 194 didn't jump to line 195 because the condition on line 194 was never true
195 continue
196 role = (msg.get("role") or "unknown").capitalize()
197 line = f"{role}: {content}"
198 remaining = self.CONTEXT_INPUT_CHAR_BUDGET - used
199 if remaining <= 0: 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true
200 break
201 if len(line) > remaining: 201 ↛ 202line 201 didn't jump to line 202 because the condition on line 201 was never true
202 if lines:
203 # Budget already spent on more recent turns — stop rather
204 # than partially including an older one.
205 break
206 # The most recent turn alone exceeds the budget: keep its head
207 # so the transcript still fits the summarizer's input cap.
208 line = line[:remaining]
209 lines.append(line)
210 used += len(line)
211 lines.reverse()
212 return "\n\n".join(lines)
214 def _summarize_prior_work(self, current_query: str) -> str:
215 """Summarize the prior conversation, focused on ``current_query``.
217 Returns an empty string when there is no prior conversation, the LLM
218 cannot be constructed (e.g. a misconfigured provider), or the LLM call
219 itself fails. The summary is additive context, so a failure here must
220 not crash the follow-up request — the research dispatch that follows
221 surfaces a genuinely-broken LLM through its own error handling.
222 """
223 transcript = self._build_conversation_text()
224 if not transcript: 224 ↛ 225line 224 didn't jump to line 225 because the condition on line 224 was never true
225 return ""
227 from ..config.llm_config import get_llm
228 from ..advanced_search_system.summarization import FocusedSummarizer
230 try:
231 llm = get_llm(settings_snapshot=self.settings_snapshot)
232 except Exception:
233 logger.opt(exception=True).debug(
234 "Could not build LLM for chat context summary; skipping"
235 )
236 return ""
238 return FocusedSummarizer(
239 llm,
240 focus_query=current_query,
241 max_sentences=self.CONTEXT_SUMMARY_MAX_SENTENCES,
242 max_chars=self.CONTEXT_SUMMARY_MAX_CHARS,
243 ).summarize(transcript)
245 def _extract_findings_from_history(self) -> str:
246 """
247 Extract key findings from assistant messages with research.
249 Returns combined findings text, limited in length.
250 """
251 findings = []
253 for msg in self.messages:
254 if msg.get("role") == "assistant" and msg.get("research_id"):
255 content = msg.get("content") or ""
256 # Summarize long responses - take first part
257 if len(content) > 500:
258 # Try to find a natural break point
259 break_point = content.find("\n\n", 300)
260 if break_point == -1 or break_point > 600: 260 ↛ 262line 260 didn't jump to line 262 because the condition on line 260 was always true
261 break_point = 500
262 content = content[:break_point] + "..."
263 findings.append(content)
265 # Keep only recent findings
266 recent_findings = findings[-self.MAX_FINDINGS_TO_INCLUDE :]
267 return "\n\n---\n\n".join(recent_findings)
269 def _get_key_entities(self) -> List[str]:
270 """Get key entities from accumulated context."""
271 entities: List[str] = self.accumulated_context.get("key_entities", [])
272 return entities[:20]
274 def _get_topics(self) -> List[str]:
275 """Get topics from accumulated context."""
276 topics: List[str] = self.accumulated_context.get("topics", [])
277 return topics[:10]
279 def extract_context_updates(self, new_content: str) -> Dict[str, Any]:
280 """
281 Extract context updates from a new research response.
283 Args:
284 new_content: New assistant response content
286 Returns:
287 Dict with entity, topic, and summary updates.
288 """
289 return {
290 "new_entities": [], # Could be enhanced with NLP entity extraction
291 "new_topics": [], # Could be enhanced with NLP topic modeling
292 "summary_addition": self._create_summary(new_content),
293 }
295 def _create_summary(self, content: str) -> str:
296 """
297 Create a brief summary of content for context accumulation.
299 Returns first meaningful paragraph or truncated content.
300 """
301 if not content:
302 return ""
304 # Try to get first paragraph
305 paragraphs = content.split("\n\n")
306 for para in paragraphs:
307 para = para.strip()
308 # Skip headers and very short paragraphs
309 if para and len(para) > 50 and not para.startswith("#"):
310 if len(para) > 300:
311 return para[:300] + "..."
312 return para
314 # Fallback: just truncate
315 if len(content) > 300: 315 ↛ 316line 315 didn't jump to line 316 because the condition on line 315 was never true
316 return content[:300] + "..."
317 return content