File size: 6,009 Bytes
52b3394 7e54ab9 52b3394 7e54ab9 52b3394 7e54ab9 52b3394 7e54ab9 52b3394 7e54ab9 52b3394 7e54ab9 52b3394 37523d3 52b3394 7e54ab9 52b3394 7e54ab9 52b3394 3e4e5aa 7e54ab9 52b3394 37523d3 7e54ab9 37523d3 7e54ab9 37523d3 52b3394 37523d3 7e54ab9 e3d8805 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | # 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 {}
|