claude-os/app/core/insight_extractor.py

"""
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)