| |
| |
| |
|
|
| 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 {} |
|
|
