Praxis CI
114 lines
3.9 KiB
Python
114 lines
3.9 KiB
Python
"""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
|
|
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"] |