feat(topics): two-level topic lists, and segment filters read from the call payload

app/ai/voice/agents/breeze_buddy/services/conversation_analysis/topics/evaluator.py

@@ -22,6 +22,7 @@
     TopicModelResponseError,
     extract_topics,
 )
+from .segments import resolve_segments, validate_segments
 
 _ANALYSIS_TIMEOUT_SECONDS = 240
 _ANALYSIS_MAX_ATTEMPTS = 2
@@ -167,6 +168,20 @@ async def analyze_topics(
             )
             return failure == MODEL_BAD_RESPONSE
 
+    entries = evaluation.get("topics") or []
+    two_level = bool(entries) and all("." in str(entry) for entry in entries)
+
+    configuration = evaluation.get("configuration") or {}
+    if isinstance(configuration, str):
+        configuration = json.loads(configuration)
+    segments = resolve_segments(
+        validate_segments(configuration.get("segments")),
+        context.get("payload") or {},
+    )
+    if segments:
+        for topic in topics:
+            topic["segments"] = segments
+
     await save_evaluation_results(
         str(evaluation["id"]),
         EvaluationType.TOPIC.value,
@@ -187,7 +202,7 @@ async def analyze_topics(
             if str(topic.get("label") or "").strip()
         }.values()
     )
-    if labels:
+    if labels and not two_level:
         await add_discovered_topics(str(context["template_id"]), labels)
     logger.bind(
         outcome="saved",

app/ai/voice/agents/breeze_buddy/services/conversation_analysis/topics/extractor.py

@@ -24,6 +24,8 @@
     Message,
 )
 
+from .segments import validate_segments
+
 _FIRST_TOKEN_TIMEOUT_SECONDS = 30
 
 
@@ -122,6 +124,7 @@ def resolve_topic_evaluation_configuration(
             "include_agent_prompt": include_agent_prompt,
             "stream": stream,
         },
+        "segments": validate_segments(raw.get("segments")),
     }
 
 
@@ -293,9 +296,9 @@ def topic_labels_to_catalog(labels: Optional[List[str]]) -> List[Dict[str, str]]
     catalog: List[Dict[str, str]] = []
     seen = set()
     for raw_label in labels or []:
-        label = normalize_topic_label(str(raw_label))
+        label = re.sub(r"\s+", " ", str(raw_label).strip()).strip(" .,:;!?-_/")[:120]
         topic_type = normalize_topic_type(label)
-        identity = (topic_type, label)
+        identity = (topic_type, label.lower())
         if not topic_type or not label or identity in seen:
             continue
         seen.add(identity)
@@ -313,12 +316,17 @@ def normalize_topics(
     catalog_by_label: Dict[str, tuple[str, str]] = {}
     for existing in existing_topics or []:
         topic_type = normalize_topic_type(str(existing.get("type") or ""))
-        label = normalize_topic_label(str(existing.get("label") or ""))
+        display = str(existing.get("label") or "").strip()
+        label = normalize_topic_label(display)
         if not topic_type or not label:
             continue
-        catalog_by_type[topic_type] = label
-        catalog_by_label[label] = (topic_type, label)
+        catalog_by_type[topic_type] = display
+        catalog_by_label[label] = (topic_type, display)
 
+    two_level = bool(catalog_by_type) and all(
+        "." in display for display in catalog_by_type.values()
+    )
+    parents = set()
     normalized: List[Dict[str, Any]] = []
     seen = set()
     for topic in parsed.topics:
@@ -333,6 +341,11 @@ def normalize_topics(
 
         if not topic_type or not label or topic_type in seen:
             continue
+        parent, dot, _ = label.partition(".")
+        if two_level and dot:
+            if parent.strip().lower() in parents:
+                continue
+            parents.add(parent.strip().lower())
         seen.add(topic_type)
         normalized.append(
             {

app/ai/voice/agents/breeze_buddy/services/conversation_analysis/topics/segments.py

@@ -0,0 +1,132 @@
+"""Segment values of a conversation, read from its payload."""
+
+import bisect
+import re
+from typing import Any, Dict, List, Mapping
+
+OTHER = "other"
+NOT_KNOWN = "not_known"
+MAX_SEGMENTS = 5
+MAX_VALUES = 100
+MAX_RANGES = 10
+_SEGMENT_KEYS = {"key", "field", "split", "values", "ranges"}
+_KEY_FORMAT = re.compile(r"^[a-z][a-z0-9_]{0,31}$")
+_NUMBER = re.compile(r"-?\d+(\.\d+)?")
+
+
+def validate_segments(segments: Any) -> List[Dict[str, Any]]:
+    """Check the segments of a topic configuration and return them cleaned."""
+    if segments is None:
+        return []
+    if not isinstance(segments, list) or len(segments) > MAX_SEGMENTS:
+        raise ValueError(
+            f"evaluation_config.segments must be a list of at most {MAX_SEGMENTS}"
+        )
+    cleaned: List[Dict[str, Any]] = []
+    seen = set()
+    for segment in segments:
+        if not isinstance(segment, Mapping):
+            raise ValueError("each evaluation_config.segments item must be an object")
+        key = segment.get("key")
+        if not isinstance(key, str) or not _KEY_FORMAT.match(key) or key in seen:
+            raise ValueError(
+                f"segment key {key!r} must be unique, lowercase snake_case, "
+                "up to 32 characters"
+            )
+        seen.add(key)
+        unknown = set(segment) - _SEGMENT_KEYS
+        if unknown:
+            raise ValueError(f"segment {key}: unknown keys {sorted(unknown)}")
+        field = segment.get("field")
+        if not isinstance(field, str) or not field.strip():
+            raise ValueError(f"segment {key}: field must name a payload key")
+        split = segment.get("split")
+        if split is not None and (not isinstance(split, str) or not split):
+            raise ValueError(f"segment {key}: split must be a non-empty string")
+        ranges = segment.get("ranges")
+        if ranges is not None:
+            if (
+                not isinstance(ranges, list)
+                or not 1 <= len(ranges) <= MAX_RANGES
+                or not all(
+                    isinstance(edge, (int, float)) and not isinstance(edge, bool)
+                    for edge in ranges
+                )
+                or sorted(set(ranges)) != ranges
+                or split is not None
+            ):
+                raise ValueError(
+                    f"segment {key}: ranges must be 1 to {MAX_RANGES} increasing "
+                    "numbers, without split"
+                )
+            edges = [f"{edge:.15g}" for edge in ranges]
+            bands = (
+                [f"<{edges[0]}"]
+                + [f"{low}-{high}" for low, high in zip(edges, edges[1:])]
+                + [f"{edges[-1]}+"]
+            )
+            if segment.get("values") not in (None, bands):
+                raise ValueError(
+                    f"segment {key}: values come from ranges, leave them out"
+                )
+            cleaned.append(
+                {"key": key, "field": field.strip(), "ranges": ranges, "values": bands}
+            )
+            continue
+        values = segment.get("values")
+        if (
+            not isinstance(values, list)
+            or not 1 <= len(values) <= MAX_VALUES
+            or not all(isinstance(value, str) and value.strip() for value in values)
+        ):
+            raise ValueError(
+                f"segment {key}: values must be 1 to {MAX_VALUES} non-empty strings"
+            )
+        values = [value.strip() for value in values]
+        lowered = [value.lower() for value in values]
+        if len(set(lowered)) != len(lowered) or {OTHER, NOT_KNOWN} & set(lowered):
+            raise ValueError(
+                f"segment {key}: values must be unique and must not be "
+                f"{OTHER!r} or {NOT_KNOWN!r}"
+            )
+        cleaned.append(
+            {"key": key, "field": field.strip(), "split": split, "values": values}
+        )
+    return cleaned
+
+
+def resolve_segments(
+    segments: List[Dict[str, Any]],
+    payload: Mapping[str, Any],
+) -> Dict[str, List[str]]:
+    """Map each segment's payload value onto the values its config lists."""
+    resolved = {}
+    for segment in segments:
+        raw = payload.get(segment["field"])
+        if segment.get("ranges"):
+            text = str(raw).replace(",", "").strip()
+            band = NOT_KNOWN
+            if _NUMBER.fullmatch(text):
+                band = segment["values"][
+                    bisect.bisect_right(segment["ranges"], float(text))
+                ]
+            resolved[segment["key"]] = [band]
+            continue
+        parts = raw if isinstance(raw, list) else [raw]
+        if segment["split"]:
+            pieces: List[str] = []
+            for part in parts:
+                if part is not None:
+                    pieces.extend(str(part).split(segment["split"]))
+            parts = pieces
+        allowed = {value.lower(): value for value in segment["values"]}
+        found: List[str] = []
+        for part in parts:
+            text = "" if part is None else str(part).strip()
+            if not text:
+                continue
+            value = allowed.get(text.lower(), OTHER)
+            if value not in found:
+                found.append(value)
+        resolved[segment["key"]] = found or [NOT_KNOWN]
+    return resolved

app/ai/voice/agents/breeze_buddy/services/conversation_analysis/worker.py

@@ -86,10 +86,10 @@ async def get_analysis_context(
             "template_id": template_id,
             "started_at": lead.call_initiated_time or lead.created_at,
             "transcript": transcript,
+            "payload": lead.payload or {},
             # The agent's own evals (not live yet, see queue.py) also need
             # what an eval judge sees beside the transcript; topics ignore it:
             # "channel": job.channel.value,
-            # "payload": lead.payload or {},
             # "meta_data": metadata,
             # "recorded_outcome": lead.outcome,
         }
@@ -111,6 +111,7 @@ async def get_analysis_context(
             "merchant_id": session.merchant_id,
             "template_id": template_id,
             "started_at": session.created_at,
+            "payload": session.metadata or {},
             "transcript": [
                 {
                     "idx": message.idx,

app/api/routers/breeze_buddy/analytics/handlers.py

@@ -40,6 +40,7 @@
 from app.database.accessor.breeze_buddy.analytics.evaluation_result import (
     get_topic_conversations,
     get_topic_dashboard,
+    get_topic_segment_counts,
 )
 from app.database.accessor.breeze_buddy.chat_analytics import (
     get_chat_summary_from_db,
@@ -111,6 +112,7 @@ def _format_topic_summary_rows(rows: List[Dict[str, Any]]) -> List[Dict[str, Any
             "underlying_topic_count": int(row["underlying_topic_count"] or 0),
             "conversation_count": int(row["conversation_count"] or 0),
             "conversation_share": float(row["conversation_share"] or 0),
+            "conversation_total": int(row["conversation_total"]),
         }
         for row in rows
     ]
@@ -372,6 +374,9 @@ async def get_topic_dashboard_analytics(
     limit = options.get("limit", 50)
     top_topics = limit if limit <= 25 else 10
     rows = await get_topic_dashboard(filters, top_topics)
+    segment_counts = (
+        await get_topic_segment_counts(filters) if filters.get("template_id") else {}
+    )
 
     return {
         "type": "topic-dashboard",
@@ -406,6 +411,7 @@ async def get_topic_dashboard_analytics(
             for row in rows
             if row["result_type"] == "topic"
         ],
+        "segment_counts": segment_counts,
     }
 
 

app/api/routers/breeze_buddy/topics/handlers.py

@@ -5,6 +5,9 @@
 from app.ai.voice.agents.breeze_buddy.services.conversation_analysis.topics.extractor import (
     resolve_topic_evaluation_configuration,
 )
+from app.ai.voice.agents.breeze_buddy.services.conversation_analysis.topics.segments import (
+    validate_segments,
+)
 from app.api.routers.breeze_buddy.templates.rbac import validate_template_access
 from app.database.accessor.breeze_buddy.evaluation_config import (
     add_discovered_topics,
@@ -20,18 +23,25 @@
     TopicCatalogResponse,
     TopicConfigurationResponse,
     TopicEvaluationSettingsRequest,
+    TopicSegment,
     UpdateTopicConfigurationRequest,
 )
 from app.schemas.breeze_buddy.evals import EvaluationType
+from app.utils.common import parse_json
 
 
 def _evaluation_catalog_response(
     template_id: str, config: Optional[Dict[str, Any]]
 ) -> TopicCatalogResponse:
+    configuration = (parse_json(config, "configuration") or {}) if config else {}
     return TopicCatalogResponse(
         template_id=template_id,
         enabled=bool(config and config.get("enabled")),
         topics=list(config.get("topics") or []) if config else [],
+        segments=[
+            TopicSegment(key=segment["key"], values=segment["values"])
+            for segment in validate_segments(configuration.get("segments"))
+        ],
     )
 
 

app/database/accessor/breeze_buddy/analytics/evaluation_result.py

@@ -10,6 +10,7 @@
 from app.database.queries.breeze_buddy.analytics.evaluation_result import (
     get_topic_conversations_query,
     get_topic_dashboard_rows_query,
+    get_topic_segment_counts_query,
     get_topics_for_source_query,
 )
 
@@ -49,22 +50,24 @@ def _topic_counts(
 
     by_topic: Dict[str, Dict[str, Any]] = {}
     for (_, topic_type), count in counts.items():
+        parent, dot, _ = count["label"].partition(".")
+        count["group"] = f"{parent.strip().lower()}." if dot else topic_type
         total = by_topic.setdefault(
-            topic_type, {"calls": 0, "first_seen_at": count["first_seen_at"]}
+            count["group"], {"calls": 0, "first_seen_at": count["first_seen_at"]}
         )
         total["calls"] += len(count["sources"])
         total["first_seen_at"] = min(total["first_seen_at"], count["first_seen_at"])
     ranked = sorted(
         by_topic,
-        key=lambda topic_type: (
-            -by_topic[topic_type]["calls"],
-            by_topic[topic_type]["first_seen_at"],
-            topic_type,
+        key=lambda group: (
+            -by_topic[group]["calls"],
+            by_topic[group]["first_seen_at"],
+            group,
         ),
     )
-    rank_of = {topic_type: rank for rank, topic_type in enumerate(ranked, 1)}
-    for (_, topic_type), count in counts.items():
-        count["rank"] = rank_of[topic_type]
+    rank_of = {group: rank for rank, group in enumerate(ranked, 1)}
+    for count in counts.values():
+        count["rank"] = rank_of[count["group"]]
     return counts
 
 
@@ -123,6 +126,7 @@ def _summary_rows(
                 "underlying_topic_count": len(underlying),
                 "conversation_count": conversation_count,
                 "conversation_share": round(conversation_count * 100 / total, 2),
+                "conversation_total": total,
             }
         )
     return sorted(
@@ -209,6 +213,19 @@ async def get_topic_dashboard(
     )
 
 
+async def get_topic_segment_counts(
+    filters: Dict[str, Any],
+) -> Dict[str, Dict[str, int]]:
+    query, values = get_topic_segment_counts_query(filters)
+    rows = await run_reader_query(query, values)
+    counts: Dict[str, Dict[str, int]] = {}
+    for row in rows or []:
+        counts.setdefault(row["segment_key"], {})[row["segment_value"]] = row[
+            "conversation_count"
+        ]
+    return counts
+
+
 async def get_topic_conversations(
     filters: Dict[str, Any],
     limit: int,