feat(evals): run custom evals after voice calls; keep the agent's outcome

app/ai/voice/agents/breeze_buddy/agent/transfer.py

@@ -93,7 +93,7 @@ async def apply_transfer(bot: "Agent", transfer: PendingAgentTransfer) -> None:
             template_id=str(transfer.template.id),
         )
         # The in-memory lead names the template its row now names: the end
-        # of the call reads it (evaluations, the post-call outcome check).
+        # of the call reads it (evaluations, the end-of-call outcome check).
         # Only these fields — the rest of bot.lead (metaData) is newer here.
         if persisted is not None:
             bot.lead.template = persisted.template

app/ai/voice/agents/breeze_buddy/crm_mirror.py

@@ -40,6 +40,9 @@
 from app.ai.voice.agents.breeze_buddy.services.conversation_analysis.preset.outcome_eval import (
     checked_outcome,
 )
+from app.ai.voice.agents.breeze_buddy.services.conversation_analysis.queue import (
+    enqueue_conversation_evaluation,
+)
 from app.core.concurrency import spawn_background_task
 from app.core.logger import logger
 from app.crm.identity.contracts import resolve as crm_resolve
@@ -49,6 +52,7 @@
     template as template_accessor,
 )
 from app.schemas import CallDirection, LeadCallStatus, LeadCallTracker
+from app.schemas.breeze_buddy.conversation_analysis import ConversationChannel
 
 SOURCE_LEAD_API = "lead-api"
 SOURCE_TELEPHONY = "telephony"
@@ -315,16 +319,25 @@ def _finished_lead_tap(lead: LeadCallTracker) -> None:
     failures, IVR, daily, demo, aborts) funnel through the two accessors
     that fire this hook, so no ending can be forgotten. Dedupe on
     topic:call_id keeps an accidental double-FINISH to one event.
+
+    It also queues the call's post-call evaluation job (its topics and
+    custom evals) once the end-of-call outcome check is done, for every
+    talked-through call, phone or web, CRM traffic or not: the one place a
+    call's job is queued. A web call's bot process waits for this before it
+    exits (services/daily/bot_runner.py).
     """
     try:
-        if is_non_customer_lead(lead.execution_mode, lead.metaData):
-            return
-        if not lead.merchant_id:
+        to_crm = bool(lead.merchant_id) and not is_non_customer_lead(
+            lead.execution_mode, lead.metaData
+        )
+        # Only a call that was talked through has anything to evaluate.
+        to_evaluate = bool(
+            lead.template_id and (lead.metaData or {}).get("transcription")
+        )
+        if not to_crm and not to_evaluate:
             return
 
-        async def _tap() -> None:
-            # the agent's outcome, or the post-call eval's correction of it
-            outcome = await checked_outcome(lead)
+        async def _mirror_completed(outcome: Optional[str]) -> None:
             await mirror_to_crm(
                 "call.completed",
                 merchant_id=lead.merchant_id,
@@ -348,6 +361,22 @@ async def _tap() -> None:
                 declared=await call_facts(lead),
             )
 
+        async def _tap() -> None:
+            try:
+                if to_crm:
+                    # the agent's outcome, or the outcome check's correction
+                    # of it (the check runs on CRM traffic only)
+                    await _mirror_completed(await checked_outcome(lead))
+            finally:
+                # once the outcome is final, so the custom evals' judge
+                # always sees the corrected word
+                if to_evaluate:
+                    await enqueue_conversation_evaluation(
+                        str(lead.id),
+                        ConversationChannel.VOICE,
+                        str(lead.template_id),
+                    )
+
         spawn_background_task(
             _tap(), name=f"crm-call-completed-{lead.call_id or lead.id}"
         )

app/ai/voice/agents/breeze_buddy/handlers/internal/end_conversation.py

@@ -10,9 +10,6 @@
 from app.ai.voice.agents.breeze_buddy.observability.tracing_setup import (
     update_span_with_evaluation_data,
 )
-from app.ai.voice.agents.breeze_buddy.services.conversation_analysis.queue import (
-    enqueue_conversation_evaluation,
-)
 from app.ai.voice.agents.breeze_buddy.template.context import TemplateContext
 from app.ai.voice.agents.breeze_buddy.utils.hold_transfer import (
     publish_hold_transfer_result,
@@ -23,7 +20,6 @@
 from app.database.accessor.breeze_buddy.chat_session import (
     flip_chat_session_to_chat,
 )
-from app.schemas.breeze_buddy.conversation_analysis import ConversationChannel
 
 callback_map = {
     "service_callback": service_callback,
@@ -306,12 +302,9 @@ async def end_conversation(context: TemplateContext, args, transition_to=None):
         else:
             logger.warning("No call_sid or lead found, skipping database update")
 
-        if updated_lead and updated_lead.template_id:
-            await enqueue_conversation_evaluation(
-                str(updated_lead.id),
-                ConversationChannel.VOICE,
-                str(updated_lead.template_id),
-            )
+        # The call's post-call evaluation job (topics, custom evals) is queued
+        # by crm_mirror's finished tap, fired by the write above, once the
+        # end-of-call outcome check is done; for a web call too.
 
         # Update OpenTelemetry span with evaluation data AFTER DB write,
         # so context.lead.outcome reflects the final persisted value.

app/ai/voice/agents/breeze_buddy/services/conversation_analysis/custom/agent_evals.py

@@ -1,80 +1,106 @@
-"""An agent's own evals for a finished call: the post-call worker's "evals"
-job, queued by ``queue.enqueue_call_evaluations`` once the call's outcome is
-final.
+"""An agent's own evals for a finished call, run by the post-call worker in
+the call's job, with its topics (worker.py). The job is queued once the
+end-of-call outcome check is done (crm_mirror's finished tap), so the judge
+sees the final outcome.
 
 The evals run in batches, one judge request per engine and model
-(``app/services/evals/custom/batch.py``). If a batch fails, the whole job goes back
-to the end of the queue, up to ``_MAX_DELIVERIES`` deliveries; each delivery
-already includes the provider's own retries. A call stores one result per
-eval, so a batch that succeeded on an earlier delivery is not stored twice.
-After the last delivery each eval still failing gets a FAILED result row.
+(``app/services/evals/custom/batch.py``). Each eval whose verdict is stored is
+marked on the job (``job.evals_done``). A batch the provider failed (a
+``ProviderError``: it has already retried what it could, and a missing key or
+URL never succeeds) gets a FAILED result row per eval at once and is not run
+again. Any other failure puts the job back at the head of the queue and pauses
+every consumer with the topics' backoff (worker.py), up to its last delivery;
+a retry runs only the evals not marked, so none is judged or stored twice. After the last one, each eval still failing gets a FAILED result
+row (``save_eval_failures``).
+
+They only store results: no custom eval changes the call's outcome (the
+end-of-call outcome check, conversation_analysis/preset/outcome_eval.py,
+is the one eval that may).
 
 Voice only for now; chat sessions can run the same evals later.
 """
 
-# Not live yet: the agent's own evals (merchant evals) come next, and this
-# is kept for them. See conversation_analysis/queue.py for what enabling
-# them takes.
-#
-# import asyncio
-# from typing import Any, Dict, List, Mapping
-#
-# from app.core.logger import logger
-# from app.database.accessor.breeze_buddy.evaluation_result import (
-#     save_evaluation_failure,
-# )
-# from app.schemas.breeze_buddy.conversation_analysis import (
-#     ConversationChannel,
-#     ConversationEvaluationJob,
-# )
-# from app.services.evals.custom.batch import agent_evals, batches, run_batch
-#
-# from ..queue import EVALS_LOG_COMPONENT, requeue_evals
-#
-# _MAX_DELIVERIES = 5
-#
-#
-# async def run_agent_evals(
-#     job: ConversationEvaluationJob,
-#     context: Dict[str, Any],
-#     evaluations: List[Dict[str, Any]],
-# ) -> None:
-#     """Run the agent's own evals on ``context`` (the worker's), retrying the
-#     job on a failed batch and recording a failure after the last try."""
-#     log = logger.bind(component=EVALS_LOG_COMPONENT, deliveries=job.deliveries)
-#     rows = agent_evals(evaluations)
-#     if not rows or job.channel is not ConversationChannel.VOICE:
-#         return
-#     failed: List[Mapping[str, Any]] = []
-#     for batch in batches(rows):
-#         try:
-#             await run_batch(batch, context, job.channel)
-#         except asyncio.CancelledError:
-#             raise
-#         except Exception as exc:
-#             names = [row["name"] for row in batch]
-#             log.error(f"Agent evals {names} for {job.source_id} failed: {exc!r}")
-#             failed.extend(batch)
-#     if not failed:
-#         log.info(f"Agent evals for {job.source_id} completed")
-#         return
-#     deliveries = job.deliveries + 1
-#     if deliveries < _MAX_DELIVERIES:
-#         await requeue_evals(job.model_copy(update={"deliveries": deliveries}))
-#         log.warning(f"Agent evals for {job.source_id} re-queued (try {deliveries})")
-#         return
-#     for row in failed:
-#         await save_evaluation_failure(
-#             str(row["id"]),
-#             str(row["evaluation_type"]),
-#             str(context["source_id"]),
-#             str(context["reseller_id"]),
-#             context.get("merchant_id"),
-#             str(context["template_id"]),
-#             context["started_at"],
-#             f"failed after {deliveries} deliveries",
-#         )
-#     log.error(
-#         f"Agent evals for {job.source_id} gave up after {deliveries} "
-#         f"deliveries: FAILED rows saved"
-#     )
+import asyncio
+from typing import Any, Dict, List, Mapping, Optional
+
+from app.core.logger import logger
+from app.database.accessor.breeze_buddy.evaluation_result import (
+    save_evaluation_failure,
+)
+from app.schemas.breeze_buddy.conversation_analysis import (
+    ConversationChannel,
+    ConversationEvaluationJob,
+)
+from app.services.evals.custom.batch import agent_evals, batches, run_batch
+from app.services.model_provider import ProviderError
+
+from ..queue import EVALS_LOG_COMPONENT
+
+
+async def run_agent_evals(
+    job: ConversationEvaluationJob,
+    context: Dict[str, Any],
+    evaluations: List[Dict[str, Any]],
+) -> List[Mapping[str, Any]]:
+    """Run the agent's own evals on ``context`` (the worker's), but those the
+    job has marked passed. Marks each one stored on ``job.evals_done``, a
+    verdict or, for a batch the provider failed, a FAILED row; returns the
+    others that failed, for the worker to retry the job."""
+    log = logger.bind(component=EVALS_LOG_COMPONENT, deliveries=job.deliveries)
+    rows = [
+        row for row in agent_evals(evaluations) if row["name"] not in job.evals_done
+    ]
+    if not rows or job.channel is not ConversationChannel.VOICE:
+        return []
+    log.info(f"Agent evals for {job.source_id} started: {[r['name'] for r in rows]}")
+    failed: List[Mapping[str, Any]] = []
+    for batch in batches(rows):
+        try:
+            await run_batch(batch, context, job.channel, job.evals_done)
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:
+            names = [row["name"] for row in batch]
+            log.error(f"Agent evals {names} for {job.source_id} failed: {exc!r}")
+            # a batch can fail part way through storing: what it stored passed
+            failed_evals = [row for row in batch if row["name"] not in job.evals_done]
+            if isinstance(exc, ProviderError):
+                # the provider already retried what it could: FAILED rows now,
+                # and the job never runs these again
+                await save_eval_failures(
+                    job, context, failed_evals, f"provider error: {exc}"
+                )
+                job.evals_done.extend(row["name"] for row in failed_evals)
+            else:
+                failed.extend(failed_evals)
+    if not failed:
+        log.info(f"Agent evals for {job.source_id} completed")
+    return failed
+
+
+async def save_eval_failures(
+    job: ConversationEvaluationJob,
+    context: Dict[str, Any],
+    failed: List[Mapping[str, Any]],
+    error: Optional[str] = None,
+) -> None:
+    """A FAILED row for each eval in ``failed``: after the job's last delivery,
+    or at once with ``error`` for a batch the provider failed."""
+    error = error or f"failed after {job.deliveries} deliveries"
+    for row in failed:
+        await save_evaluation_failure(
+            str(row["id"]),
+            str(row["evaluation_type"]),
+            str(context["source_id"]),
+            str(context["reseller_id"]),
+            context.get("merchant_id"),
+            str(context["template_id"]),
+            context["started_at"],
+            error,
+        )
+    logger.bind(
+        component=EVALS_LOG_COMPONENT, outcome="gave_up", deliveries=job.deliveries
+    ).error(
+        f"Agent evals {[row['name'] for row in failed]} for {job.source_id} gave "
+        f"up ({error}): FAILED rows saved"
+    )