diff --git a/app/ai/voice/agents/breeze_buddy/services/conversation_analysis/topics/evaluator.py b/app/ai/voice/agents/breeze_buddy/services/conversation_analysis/topics/evaluator.py index 638520c54..1feb3bd5a 100644 --- a/app/ai/voice/agents/breeze_buddy/services/conversation_analysis/topics/evaluator.py +++ b/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", diff --git a/app/ai/voice/agents/breeze_buddy/services/conversation_analysis/topics/extractor.py b/app/ai/voice/agents/breeze_buddy/services/conversation_analysis/topics/extractor.py index a3f5120c1..f77248c10 100644 --- a/app/ai/voice/agents/breeze_buddy/services/conversation_analysis/topics/extractor.py +++ b/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( { diff --git a/app/ai/voice/agents/breeze_buddy/services/conversation_analysis/topics/segments.py b/app/ai/voice/agents/breeze_buddy/services/conversation_analysis/topics/segments.py new file mode 100644 index 000000000..73535e5e0 --- /dev/null +++ b/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 diff --git a/app/ai/voice/agents/breeze_buddy/services/conversation_analysis/worker.py b/app/ai/voice/agents/breeze_buddy/services/conversation_analysis/worker.py index 6e3fa3cdc..eb8789732 100644 --- a/app/ai/voice/agents/breeze_buddy/services/conversation_analysis/worker.py +++ b/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, diff --git a/app/api/routers/breeze_buddy/analytics/handlers.py b/app/api/routers/breeze_buddy/analytics/handlers.py index 2fe7c4dba..3a2ebff9f 100644 --- a/app/api/routers/breeze_buddy/analytics/handlers.py +++ b/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, } diff --git a/app/api/routers/breeze_buddy/topics/handlers.py b/app/api/routers/breeze_buddy/topics/handlers.py index b2080304e..50dc464cd 100644 --- a/app/api/routers/breeze_buddy/topics/handlers.py +++ b/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")) + ], ) diff --git a/app/database/accessor/breeze_buddy/analytics/evaluation_result.py b/app/database/accessor/breeze_buddy/analytics/evaluation_result.py index 42839a113..3bc170921 100644 --- a/app/database/accessor/breeze_buddy/analytics/evaluation_result.py +++ b/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, diff --git a/app/database/queries/breeze_buddy/analytics/evaluation_result.py b/app/database/queries/breeze_buddy/analytics/evaluation_result.py index 031e6904a..84e586af4 100644 --- a/app/database/queries/breeze_buddy/analytics/evaluation_result.py +++ b/app/database/queries/breeze_buddy/analytics/evaluation_result.py @@ -29,6 +29,12 @@ def add(value: Any, sql: str) -> None: template_id = filters.get("template_id") or filters.get("template") if template_id: add(str(template_id), f"{alias}.template_id = ${{index}}::uuid") + for key, wanted in (filters.get("segments") or {}).items(): + values.extend([key, list(wanted)]) + clauses.append( + f"{alias}.metadata -> 'segments' -> ${len(values) - 1}::text " + f"?| ${len(values)}::text[]" + ) if filters.get("date_from"): add(filters["date_from"], f"{alias}.started_at >= ${{index}}::date") if filters.get("date_to"): @@ -84,6 +90,38 @@ def get_topic_dashboard_rows_query( return query, values +def get_topic_segment_counts_query( + filters: Dict[str, Any], +) -> Tuple[str, List[Any]]: + scope_filters = { + key: value + for key, value in filters.items() + if key not in {"topic_type", "topic_types", "segments"} + } + clauses, values = _topic_filters(scope_filters) + for key, wanted in (filters.get("segments") or {}).items(): + values.extend([key, list(wanted)]) + clauses.append( + f"(segment.key = ${len(values) - 1}::text " + f"OR ca.metadata -> 'segments' -> ${len(values) - 1}::text " + f"?| ${len(values)}::text[])" + ) + where = " AND ".join(clauses) + query = f""" + SELECT + segment.key AS segment_key, + item.value AS segment_value, + COUNT(DISTINCT ca.source_id) AS conversation_count + FROM evaluation_result ca + CROSS JOIN LATERAL jsonb_each(ca.metadata -> 'segments') + AS segment(key, items) + CROSS JOIN LATERAL jsonb_array_elements_text(segment.items) AS item(value) + WHERE {where} + GROUP BY segment.key, item.value + """ + return query, values + + def get_topic_conversations_query( filters: Dict[str, Any], limit: int, diff --git a/app/schemas/breeze_buddy/analytics.py b/app/schemas/breeze_buddy/analytics.py index 58c68018e..7ded98ce8 100644 --- a/app/schemas/breeze_buddy/analytics.py +++ b/app/schemas/breeze_buddy/analytics.py @@ -2,7 +2,7 @@ from datetime import date, datetime from enum import Enum -from typing import Any, Dict, List, Literal, Optional +from typing import Annotated, Any, Dict, List, Literal, Optional from pydantic import BaseModel, Field, field_validator @@ -138,6 +138,13 @@ def _template_must_be_uuid(cls, v: Optional[str]) -> Optional[str]: max_length=200, description="Exact underlying stable keys when opening the virtual Other topic group", ) + segments: Optional[ + Dict[str, Annotated[List[str], Field(min_length=1, max_length=100)]] + ] = Field( + None, + max_length=5, + description="Topic analytics: segment key -> values; a conversation matches when it has one of the values of every key", + ) class AnalyticsOptions(BaseModel): diff --git a/app/schemas/breeze_buddy/conversation_analysis.py b/app/schemas/breeze_buddy/conversation_analysis.py index a5f444013..2ab151443 100644 --- a/app/schemas/breeze_buddy/conversation_analysis.py +++ b/app/schemas/breeze_buddy/conversation_analysis.py @@ -43,10 +43,16 @@ class TopicEvaluationSettingsRequest(BaseModel): enabled: bool +class TopicSegment(BaseModel): + key: str + values: List[str] + + class TopicCatalogResponse(BaseModel): template_id: UUID enabled: bool topics: List[str] = Field(default_factory=list) + segments: List[TopicSegment] = Field(default_factory=list) class TopicCatalogChangeRequest(BaseModel): @@ -64,6 +70,7 @@ class UpdateTopicConfigurationRequest(BaseModel): region: Optional[str] = Field(None, max_length=100) system_prompt: Optional[str] = Field(None, max_length=50000) settings: Optional[Dict[str, Any]] = None + segments: Optional[List[Dict[str, Any]]] = None @field_validator( "provider", @@ -72,6 +79,7 @@ class UpdateTopicConfigurationRequest(BaseModel): "region", "system_prompt", "settings", + "segments", mode="before", ) @classmethod @@ -91,6 +99,7 @@ class TopicConfigurationResponse(BaseModel): region: Optional[str] = None system_prompt: str settings: Dict[str, Any] + segments: List[Dict[str, Any]] = Field(default_factory=list) class ConversationTopicsResponse(BaseModel): diff --git a/tests/services/conversation_analysis/test_worker.py b/tests/services/conversation_analysis/test_worker.py index 1759d654e..40632feaa 100644 --- a/tests/services/conversation_analysis/test_worker.py +++ b/tests/services/conversation_analysis/test_worker.py @@ -13,6 +13,7 @@ import openai import pytest from fastapi import HTTPException +from pydantic import ValidationError from app.ai.voice.agents.breeze_buddy.chat import cleanup as chat_cleanup from app.ai.voice.agents.breeze_buddy.services.conversation_analysis import ( @@ -22,16 +23,19 @@ from app.ai.voice.agents.breeze_buddy.services.conversation_analysis.topics import ( evaluator, extractor, + segments, ) from app.ai.voice.agents.breeze_buddy.template.types import ConfigurationModel from app.api.routers.breeze_buddy.analytics.handlers import ( _validate_topic_filters, get_topic_dashboard_analytics, ) +from app.api.routers.breeze_buddy.topics import handlers as topic_handlers from app.core.config import static from app.database.accessor.breeze_buddy.analytics import evaluation_result from app.database.queries.breeze_buddy.analytics.evaluation_result import ( get_topic_dashboard_rows_query, + get_topic_segment_counts_query, ) from app.database.queries.breeze_buddy.evaluation_config import ( add_discovered_topics_query, @@ -46,7 +50,7 @@ save_evaluation_results_query, ) from app.schemas import LeadCallStatus, UserInfo, UserRole -from app.schemas.breeze_buddy.analytics import AnalyticsOptions +from app.schemas.breeze_buddy.analytics import AnalyticsFilters, AnalyticsOptions from app.schemas.breeze_buddy.chat import ChatSessionStatus from app.schemas.breeze_buddy.conversation_analysis import ( ConversationChannel, @@ -177,7 +181,7 @@ async def test_openrouter_topics_go_through_the_model_provider( configuration, ) - assert [topic["label"] for topic in topics] == ["delivery delay"] + assert [topic["label"] for topic in topics] == ["Delivery Delay"] assert generate.await_args is not None (request,) = generate.await_args.args assert request.model == "openai/gpt-4o-mini" @@ -232,6 +236,100 @@ def test_a_setting_of_the_wrong_type_is_refused() -> None: ) +SEGMENTS = [ + { + "key": "lender", + "field": "lender_name", + "split": ",", + "values": ["DMI", "TVS_CREDIT"], + }, + {"key": "stage", "field": "event_name", "values": ["OFFERED", "KYC_COMPLETED"]}, +] + + +def test_segments_come_from_the_payload() -> None: + config = segments.validate_segments(SEGMENTS) + assert segments.resolve_segments( + config, {"lender_name": "TVS_CREDIT, dmi", "event_name": "KYC_COMPLETED"} + ) == {"lender": ["TVS_CREDIT", "DMI"], "stage": ["KYC_COMPLETED"]} + assert segments.resolve_segments( + config, {"lender_name": ["TVS_CREDIT, DMI", None], "event_name": "OFFERED"} + ) == {"lender": ["TVS_CREDIT", "DMI"], "stage": ["OFFERED"]} + assert segments.resolve_segments( + config, {"lender_name": "SBI", "event_name": ""} + ) == {"lender": ["other"], "stage": ["not_known"]} + assert segments.resolve_segments(config, {}) == { + "lender": ["not_known"], + "stage": ["not_known"], + } + + +def test_number_ranges_make_bands() -> None: + config = segments.validate_segments( + [{"key": "order_value", "field": "total_price", "ranges": [500, 2000]}] + ) + assert config[0]["values"] == ["<500", "500-2000", "2000+"] + resolved = extractor.resolve_topic_evaluation_configuration( + {"model": "m", "segments": config} + ) + assert extractor.resolve_topic_evaluation_configuration(resolved) == resolved + assert segments.validate_segments( + [{"key": "limit", "field": "credit_limit", "ranges": [100000, 1000000]}] + )[0]["values"] == ["<100000", "100000-1000000", "1000000+"] + for raw, band in ( + (499, "<500"), + ("500", "500-2000"), + ("1,999.50", "500-2000"), + (2000, "2000+"), + ("free", "not_known"), + (None, "not_known"), + ): + assert segments.resolve_segments(config, {"total_price": raw}) == { + "order_value": [band] + } + + +def test_a_bad_segment_config_is_refused() -> None: + for bad, message in ( + ([SEGMENTS[0], SEGMENTS[0]], "unique"), + ([{**SEGMENTS[0], "values": []}], "values"), + ([{**SEGMENTS[0], "values": ["DMI", "Other"]}], "must not be"), + ([{**SEGMENTS[0], "spilt": ","}], "unknown keys"), + ([{**SEGMENTS[0], "field": ""}], "field"), + ([{"key": "value", "field": "total_price", "ranges": [2000, 500]}], "ranges"), + ([{**SEGMENTS[0], "ranges": [500]}], "ranges"), + ( + [ + { + "key": "value", + "field": "total_price", + "ranges": [500], + "values": ["a"], + } + ], + "values come from ranges", + ), + ): + with pytest.raises(ValueError, match=message): + extractor.resolve_topic_evaluation_configuration( + {"model": "m", "segments": bad} + ) + + +def test_the_catalog_lists_the_agent_segments() -> None: + config = { + "enabled": True, + "topics": ["KYC.Selfie fails"], + "configuration": json.dumps({"model": "m", "segments": SEGMENTS}), + } + catalog = topic_handlers._evaluation_catalog_response(TEMPLATE_ID, config) + assert [segment.model_dump() for segment in catalog.segments] == [ + {"key": "lender", "values": ["DMI", "TVS_CREDIT"]}, + {"key": "stage", "values": ["OFFERED", "KYC_COMPLETED"]}, + ] + assert topic_handlers._evaluation_catalog_response(TEMPLATE_ID, None).segments == [] + + async def test_non_list_transcript_is_ignored( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -278,7 +376,7 @@ async def test_chat_context_uses_existing_accessors( template_id=TEMPLATE_ID, created_at=now, status=ChatSessionStatus.ENDED, - metadata={}, + metadata={"plan": "gold"}, ) ) list_messages = AsyncMock( @@ -302,6 +400,7 @@ async def test_chat_context_uses_existing_accessors( ) assert context and context["transcript"][0]["content"] == "My order is late" + assert context["payload"] == {"plan": "gold"} list_messages.assert_awaited_once_with("session-id") @@ -473,6 +572,51 @@ async def test_analysis_retries_once_after_timeout( ) +async def test_a_two_level_list_never_grows( + monkeypatch: pytest.MonkeyPatch, +) -> None: + context = _context() + found = [{"type": "otp_not_coming", "label": "otp not coming"}] + monkeypatch.setattr(evaluator, "extract_topics", AsyncMock(return_value=found)) + monkeypatch.setattr(evaluator, "save_evaluation_results", AsyncMock()) + add = AsyncMock() + monkeypatch.setattr(evaluator, "add_discovered_topics", add) + evaluation = { + "id": "00000000-0000-0000-0000-000000000010", + "evaluation_type": "TOPIC", + "configuration": {}, + } + + two_level = ["KYC.Location error", "KYC.Other"] + await evaluator.analyze_topics(context, {**evaluation, "topics": two_level}) + add.assert_not_awaited() + + for flat in ([], ["Refund status"], ["KYC.Location error", "Refund status"]): + add.reset_mock() + await evaluator.analyze_topics(context, {**evaluation, "topics": flat}) + add.assert_awaited_once_with(str(context["template_id"]), ["otp not coming"]) + + +def test_a_two_level_list_keeps_one_subtopic_per_topic() -> None: + labels = [ + "KYC.Selfie fails", + "KYC.Camera does not open", + "Auto-pay.UPI not received", + ] + raw = { + "topics": [ + {"type": f"t{index}", "label": label, "phrase": "p", "evidence_turns": [1]} + for index, label in enumerate(labels) + ] + } + two_level = extractor.topic_labels_to_catalog(labels) + kept = extractor.normalize_topics(raw, max_topics=2, existing_topics=two_level) + assert [topic["label"] for topic in kept] == [labels[0], labels[2]] + + flat = extractor.topic_labels_to_catalog(["Refund"]) + assert len(extractor.normalize_topics(raw, None, existing_topics=flat)) == 3 + + async def test_completion_enqueues_source_identity( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -724,6 +868,7 @@ async def test_topic_dashboard_aggregates_rows_after_fetch( summary = [row for row in dashboard if row["result_type"] == "summary"] assert len(summary) == 12 assert not any(row["is_other"] for row in summary) + assert {row["conversation_total"] for row in summary} == {11} async def test_topic_dashboard_limit_picks_named_topics( @@ -820,6 +965,59 @@ async def test_topic_dashboard_limit_is_across_agents( assert shared_on_first["rank"] == 1 +async def test_topic_dashboard_limit_keeps_a_topic_with_all_its_subtopics( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = [ + ("kyc_location_error", "KYC.Location error", 3), + ("kyc_selfie_fail", "KYC.Selfie fail", 1), + ("later_busy", "Later.Busy", 2), + ("refund", "Refund", 1), + ] + rows = [ + { + "source_id": f"{topic_type}-{index}", + "template_id": TEMPLATE_ID, + "template_name": "Agent", + "started_at": datetime(2026, 8, 1, tzinfo=timezone.utc), + "raw_topic_type": topic_type, + "raw_label": label, + } + for topic_type, label, count in calls + for index in range(count) + ] + monkeypatch.setattr( + evaluation_result, + "run_reader_query", + AsyncMock(return_value=rows), + ) + + dashboard = await evaluation_result.get_topic_dashboard( + {"date_from": date(2026, 8, 1), "date_to": date(2026, 8, 2)}, top_topics=1 + ) + + named = [ + row + for row in dashboard + if row["result_type"] == "summary" and not row["is_other"] + ] + assert {row["topic_type"] for row in named} == { + "kyc_location_error", + "kyc_selfie_fail", + } + ranks = { + row["topic_type"]: row["rank"] + for row in dashboard + if row["result_type"] == "topic" + } + assert ranks == { + "kyc_location_error": 1, + "kyc_selfie_fail": 1, + "later_busy": 2, + "refund": 3, + } + + def test_topic_filter_normalizes_template_alias() -> None: filters: dict[str, Any] = { "date_from": date(2026, 8, 1), @@ -835,6 +1033,39 @@ def test_topic_filter_normalizes_template_alias() -> None: _validate_topic_filters(filters, drilldown=True) +def test_topic_filters_match_every_segment() -> None: + query, values = get_topic_dashboard_rows_query( + { + "date_from": date(2026, 10, 1), + "date_to": date(2026, 10, 2), + "segments": {"lender": ["DMI"], "stage": ["OFFERED", "KYC_COMPLETED"]}, + } + ) + assert "ca.metadata -> 'segments' -> $1::text ?| $2::text[]" in query + assert "ca.metadata -> 'segments' -> $3::text ?| $4::text[]" in query + assert values[:4] == ["lender", ["DMI"], "stage", ["OFFERED", "KYC_COMPLETED"]] + for bad in ({"lender": []}, {"lender": ["DMI"] * 101}): + with pytest.raises(ValidationError): + AnalyticsFilters(segments=bad) + + +def test_segment_counts_leave_out_their_own_filter() -> None: + query, values = get_topic_segment_counts_query( + { + "date_from": date(2026, 10, 1), + "date_to": date(2026, 10, 2), + "template_id": TEMPLATE_ID, + "segments": {"lender": ["DMI"]}, + } + ) + assert "jsonb_each(ca.metadata -> 'segments')" in query + assert ( + "(segment.key = $4::text OR ca.metadata -> 'segments' -> $4::text " + "?| $5::text[])" + ) in query + assert values[-2:] == ["lender", ["DMI"]] + + EVALUATION = { "id": "00000000-0000-0000-0000-000000000010", "evaluation_type": "TOPIC", @@ -964,6 +1195,34 @@ async def test_bad_model_response_saves_a_failed_row( ) +async def test_saved_topics_carry_the_segments( + monkeypatch: pytest.MonkeyPatch, +) -> None: + topic = {"type": "kyc_location_blocked", "label": "KYC.Location blocked"} + save = AsyncMock() + monkeypatch.setattr(evaluator, "extract_topics", AsyncMock(return_value=[topic])) + monkeypatch.setattr(evaluator, "save_evaluation_results", save) + monkeypatch.setattr(evaluator, "add_discovered_topics", AsyncMock()) + context = { + **_context(), + "payload": {"lender_name": "TVS_CREDIT, DMI", "event_name": "OFFERED"}, + } + evaluation = { + **EVALUATION, + "configuration": {"model": "open-large-sa", "segments": SEGMENTS}, + } + + assert await evaluator.analyze_topics(context, evaluation) is True + + assert save.await_args is not None + assert save.await_args.args[-1] == [ + { + **topic, + "segments": {"lender": ["TVS_CREDIT", "DMI"], "stage": ["OFFERED"]}, + } + ] + + async def test_our_own_timeout_fails_the_job_without_pausing( monkeypatch: pytest.MonkeyPatch, ) -> None: