"""Context accumulation and management for conversation tracking.""" from dataclasses import dataclass, field from datetime import datetime from typing import Optional @dataclass class TranscriptSegment: """A segment of transcribed text.""" text: str timestamp: datetime = field(default_factory=datetime.now) @dataclass class SpeakerInfo: """Information about the speaker.""" name: str = "" background: str = "" expertise: list[str] = field(default_factory=list) recent_activity: str = "" @dataclass class ResearchResult: """A research result from web search.""" query: str summary: str sources: list[dict] = field(default_factory=list) timestamp: datetime = field(default_factory=datetime.now) class ConversationContext: """Manages accumulated context from a talk/conversation.""" def __init__(self): self.transcript_segments: list[TranscriptSegment] = [] self.speaker_info: Optional[SpeakerInfo] = None self.research_results: list[ResearchResult] = [] self.identified_topics: list[str] = [] self.key_claims: list[str] = [] self.generated_questions: list[dict] = [] self.session_start: datetime = datetime.now() def add_transcript(self, text: str) -> None: """Add a new transcript segment.""" if text.strip(): self.transcript_segments.append(TranscriptSegment(text=text.strip())) def get_full_transcript(self) -> str: """Get the complete transcript as a single string.""" return " ".join(seg.text for seg in self.transcript_segments) def get_recent_transcript(self, num_segments: int = 5) -> str: """Get the most recent transcript segments.""" recent = self.transcript_segments[-num_segments:] return " ".join(seg.text for seg in recent) def set_speaker(self, name: str, background: str = "", expertise: list[str] = None, recent_activity: str = "") -> None: """Set speaker information.""" self.speaker_info = SpeakerInfo( name=name, background=background, expertise=expertise or [], recent_activity=recent_activity ) def add_research(self, query: str, summary: str, sources: list[dict] = None) -> None: """Add a research result.""" self.research_results.append(ResearchResult( query=query, summary=summary, sources=sources or [] )) def add_topic(self, topic: str) -> None: """Add an identified topic.""" if topic not in self.identified_topics: self.identified_topics.append(topic) def add_claim(self, claim: str) -> None: """Add a key claim from the talk.""" if claim not in self.key_claims: self.key_claims.append(claim) def add_question(self, question: str, category: str = "general", reasoning: str = "") -> None: """Add a generated question.""" self.generated_questions.append({ "question": question, "category": category, "reasoning": reasoning, "timestamp": datetime.now().isoformat() }) def get_context_summary(self) -> dict: """Get a summary of all accumulated context.""" return { "transcript_length": len(self.get_full_transcript()), "num_segments": len(self.transcript_segments), "speaker": self.speaker_info.name if self.speaker_info else None, "topics": self.identified_topics, "claims_count": len(self.key_claims), "research_count": len(self.research_results), "questions_generated": len(self.generated_questions), "session_duration": (datetime.now() - self.session_start).seconds } def has_enough_context(self, min_words: int = 100) -> bool: """Check if we have enough context to start generating questions.""" transcript = self.get_full_transcript() word_count = len(transcript.split()) return word_count >= min_words def to_prompt_context(self) -> str: """Format context for use in LLM prompts.""" parts = [] # Transcript transcript = self.get_full_transcript() if transcript: parts.append(f"## Transcript\n{transcript}") # Speaker info if self.speaker_info and self.speaker_info.name: speaker_section = f"## Speaker: {self.speaker_info.name}" if self.speaker_info.background: speaker_section += f"\nBackground: {self.speaker_info.background}" if self.speaker_info.expertise: speaker_section += f"\nExpertise: {', '.join(self.speaker_info.expertise)}" if self.speaker_info.recent_activity: speaker_section += f"\nRecent Activity: {self.speaker_info.recent_activity}" parts.append(speaker_section) # Topics if self.identified_topics: parts.append(f"## Topics Discussed\n" + "\n".join(f"- {t}" for t in self.identified_topics)) # Key claims if self.key_claims: parts.append(f"## Key Claims Made\n" + "\n".join(f"- {c}" for c in self.key_claims)) # Research if self.research_results: research_section = "## Research Findings" for r in self.research_results[-3:]: # Last 3 research results research_section += f"\n\n### {r.query}\n{r.summary}" parts.append(research_section) return "\n\n".join(parts) def clear(self) -> None: """Reset all context.""" self.transcript_segments = [] self.speaker_info = None self.research_results = [] self.identified_topics = [] self.key_claims = [] self.generated_questions = [] self.session_start = datetime.now()