""" Insight Extractor for Claude Code session transcripts. Uses LLM to extract valuable insights (decisions, patterns, solutions, blockers). """ import json import logging from dataclasses import dataclass from typing import List, Dict, Any, Optional import httpx from app.core.config import Config logger = logging.getLogger(__name__) @dataclass class Insight: """An extracted insight from a session.""" type: str # "decision", "pattern", "solution", or "blocker" title: str content: str confidence: float # 0.0 to 1.0 metadata: Optional[Dict[str, Any]] = None # Extraction prompt template EXTRACTION_PROMPT = """Analyze this Claude Code session transcript and extract valuable insights. CRITICAL RULES: 1. ONLY extract insights that are EXPLICITLY mentioned in the transcript below 2. Do NOT invent, assume, or hallucinate any insights 3. If the transcript doesn't contain valuable insights, return an empty list 4. Every insight MUST be directly traceable to specific text in the transcript Look for these types of insights (ONLY if actually present): 1. **Decisions**: Technical choices made with clear reasoning stated in the transcript 2. **Patterns**: Reusable approaches or code patterns that were explicitly discovered/discussed 3. **Solutions**: Bug fixes or error resolutions that were actually implemented 4. **Blockers**: Problems encountered that were explicitly mentioned For each insight you extract: - The title MUST use terminology from the actual transcript - The content MUST describe what actually happened (not what might have happened) - Confidence should be 0.9+ only if explicitly stated, 0.7-0.8 if implied from context - If you cannot point to specific transcript text supporting the insight, DO NOT include it **Extract ONLY insights that someone reading the transcript would clearly see.** Skip routine operations (ls, git status, simple file reads, etc.). If unsure whether something qualifies, leave it out. Session transcript: {transcript} Return valid JSON only (no markdown, no extra text): {{ "insights": [ {{"type": "decision|pattern|solution|blocker", "title": "Title using actual terms from transcript", "content": "What specifically happened based on transcript text...", "confidence": 0.85}} ] }} If no valuable insights found, return: {{"insights": []}} """ class InsightExtractor: """Extract insights from session transcripts using LLM.""" def __init__(self, ollama_base_url: Optional[str] = None, model: Optional[str] = None): """ Initialize insight extractor. Args: ollama_base_url: Ollama API base URL (defaults to Config.OLLAMA_HOST) model: LLM model to use (defaults to Config.OLLAMA_MODEL) """ self.ollama_base_url = ollama_base_url or Config.OLLAMA_HOST self.model = model or Config.get_active_llm_model() logger.info(f"Initialized InsightExtractor with model: {self.model}") async def extract( self, session_summary: str, insight_types: Optional[List[str]] = None ) -> List[Insight]: """ Extract insights from session summary using LLM. Args: session_summary: Condensed session transcript insight_types: Filter to specific types (defaults to all types) Returns: List of Insight objects """ if insight_types is None: insight_types = ["decision", "pattern", "solution", "blocker"] logger.info(f"Extracting insights from session summary ({len(session_summary)} chars)") # Build prompt prompt = EXTRACTION_PROMPT.format(transcript=session_summary) # Call Ollama try: insights = await self._call_ollama(prompt) # Filter by requested types filtered_insights = [ i for i in insights if i.type in insight_types ] logger.info(f"Extracted {len(filtered_insights)} insights (filtered from {len(insights)})") return filtered_insights except Exception as e: logger.error(f"Failed to extract insights: {e}") return [] async def _call_ollama(self, prompt: str) -> List[Insight]: """ Call Ollama API to extract insights. Args: prompt: Extraction prompt with session transcript Returns: List of Insight objects """ url = f"{self.ollama_base_url}/api/generate" payload = { "model": self.model, "prompt": prompt, "stream": False, "format": "json", # Request JSON response "options": { "temperature": 0.3, # Lower temperature for more focused extraction "num_predict": 2048, # Allow longer responses for multiple insights "top_k": 40, "top_p": 0.9 } } try: async with httpx.AsyncClient(timeout=120.0) as client: response = await client.post(url, json=payload) response.raise_for_status() result = response.json() response_text = result.get("response", "") # Parse JSON response insights_data = self._parse_llm_response(response_text) # Convert to Insight objects insights = [] for item in insights_data.get("insights", []): try: insight = Insight( type=item.get("type", "pattern"), title=item.get("title", "Untitled"), content=item.get("content", ""), confidence=float(item.get("confidence", 0.5)) ) insights.append(insight) except (ValueError, KeyError) as e: logger.warning(f"Failed to parse insight item: {e}") continue return insights except httpx.HTTPError as e: logger.error(f"Ollama API error: {e}") raise except Exception as e: logger.error(f"Unexpected error calling Ollama: {e}") raise def _parse_llm_response(self, response_text: str) -> Dict[str, Any]: """ Parse LLM response JSON. Handles cases where LLM might include markdown code blocks or extra text. Args: response_text: Raw LLM response Returns: Parsed JSON dict """ # Clean up response if it has markdown code blocks cleaned_text = response_text.strip() # Remove markdown code blocks if present if cleaned_text.startswith("```json"): cleaned_text = cleaned_text[7:] # Remove ```json elif cleaned_text.startswith("```"): cleaned_text = cleaned_text[3:] # Remove ``` if cleaned_text.endswith("```"): cleaned_text = cleaned_text[:-3] cleaned_text = cleaned_text.strip() try: return json.loads(cleaned_text) except json.JSONDecodeError as e: logger.error(f"Failed to parse LLM JSON response: {e}") logger.debug(f"Response text: {cleaned_text[:500]}") # Return empty insights on parse failure return {"insights": []} def filter_by_confidence( self, insights: List[Insight], min_confidence: float = 0.7 ) -> List[Insight]: """ Filter insights by minimum confidence threshold. Args: insights: List of insights min_confidence: Minimum confidence score (0.0-1.0) Returns: Filtered list of insights """ return [i for i in insights if i.confidence >= min_confidence] def group_by_type(self, insights: List[Insight]) -> Dict[str, List[Insight]]: """ Group insights by type. Args: insights: List of insights Returns: Dict mapping type to list of insights """ grouped: Dict[str, List[Insight]] = { "decision": [], "pattern": [], "solution": [], "blocker": [] } for insight in insights: if insight.type in grouped: grouped[insight.type].append(insight) return grouped def format_for_save(self, insight: Insight, session_id: str) -> str: """ Format insight as markdown for saving to knowledge base. Args: insight: Insight to format session_id: Session ID for metadata Returns: Markdown string """ type_emoji = { "decision": "💎", "pattern": "🔄", "solution": "✅", "blocker": "🚧" } emoji = type_emoji.get(insight.type, "📝") type_label = insight.type.capitalize() markdown = f"""# {emoji} {insight.title} **Type:** {type_label} **Confidence:** {insight.confidence:.2f} **Source:** Session `{session_id}` ## Details {insight.content} --- *Extracted from Claude Code session on {insight.metadata.get('extracted_date', 'unknown') if insight.metadata else 'unknown'}* """ return markdown def extract_insights_sync( session_summary: str, insight_types: Optional[List[str]] = None, min_confidence: float = 0.7 ) -> List[Insight]: """ Synchronous wrapper for extract_insights (for non-async contexts). Args: session_summary: Condensed session transcript insight_types: Filter to specific types min_confidence: Minimum confidence threshold Returns: List of high-confidence Insight objects """ import asyncio extractor = InsightExtractor() # Run async extraction loop = asyncio.get_event_loop() insights = loop.run_until_complete( extractor.extract(session_summary, insight_types) ) # Filter by confidence return extractor.filter_by_confidence(insights, min_confidence)