python-services/summary.py

# summary.py
# Compresses old message histories asynchronously into dense systemic memory logs.
# Prevents context window inflation and keeps voice processing latencies low.

import json
import logging
import re
import asyncio
from typing import List, Dict, Any, Optional
from client import groq

logger = logging.getLogger("summary")

NAME_STOP_WORDS = {"he", "hai", "hun", "hoon", "ho", "and", "ye", "yeh", "the", "a"}


def _build_summary_system_prompt(caller_fields_spec: Optional[List[Dict[str, Any]]]) -> str:
    if not caller_fields_spec:
        return (
            "You are a conversation summarizer. Summarize the key facts, topics discussed, "
            "decisions made, and pending actions from the conversation. "
            "Do NOT mention or infer caller personal details (name, phone, ID, etc.). "
            "Merge with existing summary. Keep under 3 sentences."
        )
    return (
        "You are a conversation summarizer. Summarize the key facts, topics discussed, "
        "decisions made, and pending actions from the conversation. "
        "Use the provided CALLER DATA block as the authoritative source for caller identity "
        "fields (name, phone, custom fields). Do NOT re-extract or guess caller identity "
        "from transcript wording. Roman Urdu copulas (he, hai, hun, hoon) are NOT names — "
        "never record them as caller names. If a caller field is null or empty in CALLER DATA, "
        "omit it from the summary — do not invent a value. "
        "Merge with existing summary. Keep under 3 sentences."
    )


def _sanitize_summary(result: str, caller_info: Optional[Dict[str, Any]]) -> str:
    """Strip hallucinated stop-word names when no authoritative name exists."""
    if not result or not caller_info:
        return result or ""
    if caller_info.get("name"):
        return result
    cleaned = result
    for stop in NAME_STOP_WORDS:
        cleaned = re.sub(
            rf"(caller\s+)?name[:\s]+{re.escape(stop)}\b",
            "",
            cleaned,
            flags=re.I,
        )
    return cleaned.strip()


async def generate_summary(
    existing_summary: str,
    messages_to_process: List[Dict[str, Any]],
    caller_info: Optional[Dict[str, Any]] = None,
    caller_fields_spec: Optional[List[Dict[str, Any]]] = None,
) -> str:
    """
    Generate a condensed structural summary of earlier conversation message arrays.
    Runs asynchronously in an isolated thread executor pool to prevent voice line stutters.
    """
    if not messages_to_process:
        return existing_summary or ""

    conversation_string = "\n".join(
        f"{'User' if m.get('role') == 'user' else 'Assistant'}: {m.get('content', '')}"
        for m in messages_to_process
    )

    caller_info = caller_info or {}
    caller_fields_spec = caller_fields_spec or []
    system_prompt = _build_summary_system_prompt(caller_fields_spec)

    user_parts = []
    if caller_fields_spec:
        user_parts.append(
            "CALLER DATA (authoritative — do not override):\n"
            + json.dumps(caller_info, ensure_ascii=False)
        )
    user_parts.append(f"Existing Summary: {existing_summary or 'None yet.'}")
    user_parts.append(f"New messages:\n{conversation_string}")
    user_content = "\n\n".join(user_parts)

    try:
        loop = asyncio.get_running_loop()

        response = await loop.run_in_executor(
            None,
            lambda: groq.chat.completions.create(
                model="llama-3.1-8b-instant",
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_content},
                ],
                temperature=0.2,
                max_tokens=150,
            ),
        )

        result = response.choices[0].message.content
        if result:
            cleaned = _sanitize_summary(result.strip(), caller_info)
            logger.info("Summary updated successfully")
            return cleaned

        return existing_summary or ""

    except Exception as e:
        logger.error(f"Summary generation failed: {e}")
        return existing_summary or ""


async def extract_caller_data_from_transcript(
    transcript: str,
    fields_spec: List[Dict[str, Any]]
) -> Dict[str, Any]:
    """
    Call Groq to extract structured caller fields from the conversation transcript
    based on the fields specification. Returns a dictionary of extracted values.
    """
    if not fields_spec or not transcript.strip():
        return {}

    system_prompt = (
        "You are a structured data extractor. Your task is to extract information from "
        "the provided conversation transcript matching the schema specification. "
        "Return ONLY a clean JSON object containing the extracted keys. "
        "If a field is not mentioned or cannot be inferred, set its value to null. "
        "Do not invent or assume any values. Output valid JSON, nothing else."
    )

    schema_desc = {f["key"]: {"label": f.get("label", f["key"]), "type": f.get("type", "text")} for f in fields_spec}
    user_content = (
        f"Schema Spec:\n{json.dumps(schema_desc, ensure_ascii=False)}\n\n"
        f"Transcript:\n{transcript}"
    )

    try:
        loop = asyncio.get_running_loop()
        response = await loop.run_in_executor(
            None,
            lambda: groq.chat.completions.create(
                model="llama-3.1-8b-instant",
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_content},
                ],
                temperature=0.0,
                response_format={"type": "json_object"},
                max_tokens=250,
            ),
        )
        result = response.choices[0].message.content
        if result:
            return json.loads(result)
        return {}
    except Exception as e:
        logger.error(f"LLM caller extraction failed: {e}")
        return {}