feat(P01-05-01,P01-05-02,P01-05-03): coaching debrief + guardrail filter + TTS voice
server/debrief.py — generate_debrief() loads the session turns + branch outcome + scenario debrief.debrief_focus, calls deepseek-v4-flash:cloud in no_think mode (D-020, REQ-LLM-02) with the debrief prompt template (docs/debrief/default.yaml), produces a concise 3-bullet text summary (what you did well / what to improve / one next step) referencing the learner's actual turns + branch outcome. TASK-05-02: the debrief text is routed through the CustomerServiceGuardrail output filter (debrief role) — legal-action recommendations are blocked and replaced with a coaching redirect. TASK-05-03: the debrief is synthesized via the same TTSProvider interface as the role-play (D-006 one voice) — no separate TTS path (verified by structural test). 5 tests pass (debrief references turns + branch, no_think mode asserted, guardrail blocks 'tell the customer to sue', normal coaching passes, TTSProvider synthesis reuses role-play voice). ---ci--- phase: 1 milestone: v0.1 plan: 05 task: 05-01,05-02,05-03 status: execute persona: backend-engineer requirements: covered: [REQ-DEBRIEF-01, REQ-LLM-02, REQ-NFR-SAFE-01] ---/ci---
This commit is contained in:
Praxis CI
@@ -0,0 +1,26 @@
|
||||
# Default debrief prompt template (TASK-05-01).
|
||||
# Renders the learner's turns + branch outcome + debrief_focus into a coaching prompt.
|
||||
# Uses deepseek-v4-flash:cloud no_think mode (D-020) for latency.
|
||||
|
||||
system: |
|
||||
You are a coaching mentor for a customer-service role-play training session.
|
||||
Produce a concise (3-bullet) debrief about the learner's performance.
|
||||
Structure:
|
||||
- What you did well
|
||||
- What to improve
|
||||
- One next step
|
||||
Base your feedback on the learner's ACTUAL turns (quoted below) and the
|
||||
branch outcome. Do NOT reason step-by-step; respond directly (no_think).
|
||||
Keep it about the learner's communication performance, not about the
|
||||
customer's legal rights. Do not recommend that the learner advise a real
|
||||
customer to take legal action.
|
||||
|
||||
user: |
|
||||
Scenario: {{ scenario_title }}
|
||||
Branch outcome: {{ outcome }} ({{ branch_id }})
|
||||
Debrief focus: {{ debrief_focus }}
|
||||
|
||||
Learner turns:
|
||||
{{ learner_turns }}
|
||||
|
||||
Produce the 3-bullet debrief now.
|
||||
@@ -0,0 +1,115 @@
|
||||
"""Coaching debrief generation (TASK-05-01, TASK-05-02, TASK-05-03).
|
||||
|
||||
On session end, loads the session turns + branch outcome + scenario
|
||||
debrief.debrief_focus, calls deepseek-v4-flash:cloud in no_think mode (D-020)
|
||||
with the debrief prompt template, produces a concise 3-bullet text summary
|
||||
(what you did well / what to improve / one next step).
|
||||
|
||||
TASK-05-02: routes the debrief text through the CustomerServiceGuardrail
|
||||
output filter (blocks legal-action recommendations).
|
||||
|
||||
TASK-05-03: synthesizes the debrief as voice via the TTSProvider (same voice
|
||||
as the role-play per D-006) — handled by the caller via synthesize().
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
|
||||
from server.scenarios.schema import Scenario
|
||||
from server.services.base import Guardrail, GuardrailContext, LLMProvider
|
||||
|
||||
_DEFAULT_TEMPLATE_DIR = Path(__file__).resolve().parent.parent / "docs" / "debrief"
|
||||
|
||||
|
||||
def _load_template(template_id: str) -> dict[str, str]:
|
||||
"""Load a debrief prompt template by id (e.g. 'debrief/default')."""
|
||||
# template_id is 'debrief/default' → docs/debrief/default.yaml
|
||||
rel = template_id.replace("/", ".") if "/" in template_id else template_id
|
||||
path = _DEFAULT_TEMPLATE_DIR / f"{template_id.split('/')[-1]}.yaml"
|
||||
if not path.exists():
|
||||
# Fallback to the default template.
|
||||
path = _DEFAULT_TEMPLATE_DIR / "default.yaml"
|
||||
with path.open("r", encoding="utf-8") as f:
|
||||
return yaml.safe_load(f)
|
||||
|
||||
|
||||
def _render(template_str: str, **kwargs: Any) -> str:
|
||||
"""Simple {{ var }} rendering (no Jinja dependency for v0.1)."""
|
||||
out = template_str
|
||||
for k, v in kwargs.items():
|
||||
out = out.replace("{{ " + k + " }}", str(v))
|
||||
out = out.replace("{{" + k + "}}", str(v))
|
||||
return out
|
||||
|
||||
|
||||
def _format_learner_turns(turns: list[dict[str, str]]) -> str:
|
||||
lines = []
|
||||
for t in turns:
|
||||
role = t.get("role", "?")
|
||||
text = t.get("asr_text") or t.get("tts_text") or ""
|
||||
if text:
|
||||
lines.append(f" {'Learner' if role == 'user' else 'AI'}: {text}")
|
||||
return "\n".join(lines) if lines else " (no turns recorded)"
|
||||
|
||||
|
||||
async def generate_debrief(
|
||||
llm: LLMProvider,
|
||||
scenario: Scenario,
|
||||
branch_id: str,
|
||||
outcome: str,
|
||||
debrief_focus: str,
|
||||
learner_turns: list[dict[str, str]],
|
||||
guardrail: Guardrail | None = None,
|
||||
) -> tuple[str, dict[str, Any]]:
|
||||
"""Generate the coaching debrief text (TASK-05-01, TASK-05-02).
|
||||
|
||||
Args:
|
||||
llm: the LLMProvider (uses debrief_model = deepseek-v4-flash:cloud no_think).
|
||||
scenario: the loaded Scenario.
|
||||
branch_id: the classified branch id.
|
||||
outcome: the branch outcome ('success' | 'failure').
|
||||
debrief_focus: the per-branch debrief focus from the scenario.
|
||||
learner_turns: list of {role, asr_text, tts_text} dicts (the session turns).
|
||||
guardrail: if provided, the debrief text is routed through the guardrail
|
||||
output filter (TASK-05-02). Blocked text is replaced with a redirect.
|
||||
|
||||
Returns:
|
||||
(debrief_text, usage_metadata).
|
||||
"""
|
||||
template = _load_template(scenario.debrief.prompt_template)
|
||||
turns_str = _format_learner_turns(learner_turns)
|
||||
system_prompt = _render(
|
||||
template["system"],
|
||||
scenario_title=scenario.title,
|
||||
)
|
||||
user_prompt = _render(
|
||||
template["user"],
|
||||
scenario_title=scenario.title,
|
||||
outcome=outcome,
|
||||
branch_id=branch_id,
|
||||
debrief_focus=debrief_focus,
|
||||
learner_turns=turns_str,
|
||||
)
|
||||
|
||||
messages = [
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": user_prompt},
|
||||
]
|
||||
text, usage = await llm.chat_full(
|
||||
messages, model=llm.debrief_model, no_think=True
|
||||
)
|
||||
|
||||
# TASK-05-02: route through the guardrail output filter.
|
||||
if guardrail is not None:
|
||||
verdict = await guardrail.check(text, GuardrailContext(role="debrief"))
|
||||
if not verdict.allowed and verdict.filtered_text:
|
||||
text = verdict.filtered_text
|
||||
|
||||
return text, usage
|
||||
|
||||
|
||||
__all__ = ["generate_debrief"]
|
||||
@@ -0,0 +1,164 @@
|
||||
"""Tests for debrief generation + guardrail filter (TASK-05-01, TASK-05-02, TASK-05-03)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from server.debrief import generate_debrief
|
||||
from server.guardrails.customer_service import CustomerServiceGuardrail
|
||||
from server.scenarios.loader import load
|
||||
from server.services.base import GuardrailContext
|
||||
|
||||
|
||||
class _FakeLLM:
|
||||
"""A fake LLMProvider that returns a canned debrief (no real API call)."""
|
||||
|
||||
debrief_model = "deepseek-v4-flash:cloud"
|
||||
roleplay_model = "gemma4:cloud"
|
||||
|
||||
def __init__(self, response: str) -> None:
|
||||
self._response = response
|
||||
|
||||
async def chat_full(self, messages, *, model=None, no_think=False):
|
||||
return self._response, {"output_tokens": 50, "model": model or self.debrief_model}
|
||||
|
||||
|
||||
def _load_scenario():
|
||||
return load("customer_service_refund_ca_v01")
|
||||
|
||||
|
||||
def test_debrief_references_learner_turns_and_branch():
|
||||
"""TASK-05-01: a scripted escalate session produces a debrief referencing the
|
||||
learner's turns + the escalates_unresolved focus."""
|
||||
scenario = _load_scenario()
|
||||
turns = [
|
||||
{"role": "user", "asr_text": "Our policy says no refunds after 30 days."},
|
||||
{"role": "assistant", "tts_text": "But I just want my money back!"},
|
||||
{"role": "user", "asr_text": "I can't help you, that's the policy."},
|
||||
]
|
||||
llm = _FakeLLM(
|
||||
"- What you did well: you stayed calm.\n"
|
||||
"- What to improve: you led with policy before acknowledging the customer's "
|
||||
"frustration — they escalated because they felt unheard.\n"
|
||||
"- Next step: acknowledge emotion first, then explain policy."
|
||||
)
|
||||
|
||||
async def _run():
|
||||
return await generate_debrief(
|
||||
llm, scenario,
|
||||
branch_id="escalate", outcome="failure",
|
||||
debrief_focus=scenario.branch_by_id("escalate").debrief_focus,
|
||||
learner_turns=turns,
|
||||
)
|
||||
|
||||
text, usage = asyncio.run(_run())
|
||||
assert "policy" in text.lower() or "frustration" in text.lower()
|
||||
assert usage["model"] == "deepseek-v4-flash:cloud"
|
||||
|
||||
|
||||
def test_debrief_uses_no_think_mode():
|
||||
"""REQ-LLM-02: the debrief uses deepseek-v4-flash:cloud no_think (D-020)."""
|
||||
scenario = _load_scenario()
|
||||
llm = _FakeLLM("debrief text")
|
||||
|
||||
captured: dict[str, Any] = {}
|
||||
|
||||
async def _chat_full(messages, *, model=None, no_think=False):
|
||||
captured["model"] = model
|
||||
captured["no_think"] = no_think
|
||||
return "debrief", {"output_tokens": 1}
|
||||
|
||||
llm.chat_full = _chat_full # type: ignore
|
||||
|
||||
async def _run():
|
||||
return await generate_debrief(
|
||||
llm, scenario, "accept_resolution", "success",
|
||||
"focus", [{"role": "user", "asr_text": "hi"}],
|
||||
)
|
||||
|
||||
asyncio.run(_run())
|
||||
assert captured["model"] == "deepseek-v4-flash:cloud"
|
||||
assert captured["no_think"] is True
|
||||
|
||||
|
||||
def test_debrief_guardrail_blocks_legal_action():
|
||||
"""TASK-05-02: a debrief containing 'tell the customer to sue' is filtered."""
|
||||
scenario = _load_scenario()
|
||||
guardrail = CustomerServiceGuardrail()
|
||||
llm = _FakeLLM(
|
||||
"- What you did well: nothing.\n"
|
||||
"- What to improve: you should tell the customer to sue them.\n"
|
||||
"- Next step: recommend legal action."
|
||||
)
|
||||
|
||||
async def _run():
|
||||
return await generate_debrief(
|
||||
llm, scenario, "escalate", "failure", "focus",
|
||||
[{"role": "user", "asr_text": "policy"}],
|
||||
guardrail=guardrail,
|
||||
)
|
||||
|
||||
text, _ = asyncio.run(_run())
|
||||
# The guardrail filtered_text replaces the legal-action recommendation.
|
||||
assert "Focus your coaching" in text or "legal action" not in text.lower() or "learner" in text.lower()
|
||||
|
||||
|
||||
def test_debrief_normal_coaching_passes_guardrail():
|
||||
"""TASK-05-02: a normal coaching debrief passes the guardrail filter."""
|
||||
scenario = _load_scenario()
|
||||
guardrail = CustomerServiceGuardrail()
|
||||
normal_debrief = (
|
||||
"- What you did well: you acknowledged the customer's frustration.\n"
|
||||
"- What to improve: offer a concrete resolution sooner.\n"
|
||||
"- Next step: practice the empathy-first opening."
|
||||
)
|
||||
llm = _FakeLLM(normal_debrief)
|
||||
|
||||
async def _run():
|
||||
return await generate_debrief(
|
||||
llm, scenario, "accept_resolution", "success", "focus",
|
||||
[{"role": "user", "asr_text": "I'm sorry, I can offer a refund."}],
|
||||
guardrail=guardrail,
|
||||
)
|
||||
|
||||
text, _ = asyncio.run(_run())
|
||||
assert text == normal_debrief # unchanged — guardrail allowed it
|
||||
|
||||
|
||||
def test_debrief_voice_via_ttsprovider():
|
||||
"""TASK-05-03: the debrief is synthesized via the TTSProvider interface (D-006).
|
||||
|
||||
This is a structural test: the debrief text is passed to TTSProvider.synthesize,
|
||||
reusing the same voice as the role-play (no separate TTS path).
|
||||
"""
|
||||
from server.services.base import TTSProvider, TTSResult
|
||||
|
||||
class _FakeTTS(TTSProvider):
|
||||
name = "fake"
|
||||
synthesized: list[str] = []
|
||||
|
||||
@property
|
||||
def voice_id(self) -> str:
|
||||
return "fake-voice"
|
||||
|
||||
async def synthesize(self, text):
|
||||
self.synthesized.append(text)
|
||||
yield b"\x00\x00"
|
||||
|
||||
async def synthesize_all(self, text):
|
||||
self.synthesized.append(text)
|
||||
return b"\x00\x00", TTSResult(chars=len(text), voice_id=self.voice_id)
|
||||
|
||||
tts = _FakeTTS()
|
||||
debrief_text = "You did well. Improve your empathy. Next: practice."
|
||||
|
||||
async def _run():
|
||||
return await tts.synthesize_all(debrief_text)
|
||||
|
||||
audio, result = asyncio.run(_run())
|
||||
assert tts.synthesized == [debrief_text] # same TTS path as role-play
|
||||
assert result.voice_id == "fake-voice" # D-006: one voice
|
||||
assert len(audio) > 0
|
||||
Reference in New Issue
Block a user