From 30051fdfd6f84c3578c3aa65d5856b323fef96e6 Mon Sep 17 00:00:00 2001 From: Praxis CI Date: Sat, 1 Aug 2026 13:11:30 +0000 Subject: [PATCH] feat(P01-03-03,P01-03-07): Pipecat Flows wiring + scenario-driven prompt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit server/scenarios/runtime.py — ScenarioRuntime maps a Scenario to a Pipecat Flows state-machine spec: initial 'conversation' state with the scenario's system prompt + opening line, branch metadata carried in the spec (v0.1 has no in-flight transitions per G-002 — branching is post-hoc; Phase 2+ can fork without schema change). set_branch() resolves the branch outcome from the classifier; debrief_focus() returns the per-branch focus. TASK-03-07: server/pipeline.py build_pipeline() now accepts a scenario_id, loads the runtime, and uses scenario.setup.system_prompt instead of the SLICE-02 hardcoded walking-skeleton prompt. Falls back gracefully if the scenario can't load. server/__main__.py passes PRAXIS_SCENARIO=customer_service_refund_ca_v01 by default. 7 runtime tests pass (system prompt, set_branch accept/escalate, unknown-branch error, debrief_focus per branch, flows spec branches, debrief model config). ---ci--- phase: 1 milestone: v0.1 plan: 03 task: 03-03,03-07 status: execute persona: backend-engineer requirements: covered: [REQ-SCEN-01, REQ-SCEN-FMT-01, REQ-ORCH-02] ---/ci--- --- server/__main__.py | 18 +++++- server/pipeline.py | 55 +++++++++++++++--- server/scenarios/runtime.py | 103 +++++++++++++++++++++++++++++++++ tests/test_scenario_runtime.py | 68 ++++++++++++++++++++++ 4 files changed, 234 insertions(+), 10 deletions(-) create mode 100644 server/scenarios/runtime.py create mode 100644 tests/test_scenario_runtime.py diff --git a/server/__main__.py b/server/__main__.py index cae260d..90afe32 100644 --- a/server/__main__.py +++ b/server/__main__.py @@ -75,7 +75,12 @@ async def health() -> dict[str, Any]: @app.post("/pipecat/webrtc") async def webrtc_offer(offer: WebRTCOffer) -> dict[str, str]: - """Accept a WebRTC offer, start a Pipecat pipeline task, return the answer.""" + """Accept a WebRTC offer, start a Pipecat pipeline task, return the answer. + + Loads the v0.1 scenario (customer_service_refund_ca_v01) so the pipeline + uses the scenario-driven system prompt + opening line (TASK-03-07). + """ + scenario_id = _env("PRAXIS_SCENARIO", "customer_service_refund_ca_v01") try: connection = SmallWebRTCConnection( ice_servers=[{"urls": "stun:stun.l.google.com:19302"}], @@ -84,11 +89,20 @@ async def webrtc_offer(offer: WebRTCOffer) -> dict[str, str]: await connection.accept() answer = connection.get_answer() # Build + run the pipeline for this connection. - pipeline, task, runner, transport = build_pipeline(connection) + pipeline, task, runner, transport, scenario_runtime = build_pipeline( + connection, scenario_id=scenario_id + ) # Run the pipeline task in the background; the runner manages its lifecycle. import asyncio asyncio.create_task(runner.run(task)) + # Play the opening line + disclaimer as the first AI utterance if the + # scenario loaded (the runner handles the actual TTS queueing). + if scenario_runtime is not None: + logger.info( + f"Session starting with scenario {scenario_id!r}; " + f"opening line: {scenario_runtime.opening_line[:60]!r}" + ) return {"sdp": answer["sdp"], "type": answer["type"]} except Exception as exc: logger.error(f"WebRTC offer failed: {exc}") diff --git a/server/pipeline.py b/server/pipeline.py index f4526fa..56bd30e 100644 --- a/server/pipeline.py +++ b/server/pipeline.py @@ -41,12 +41,21 @@ WALKING_SKELETON_OPENING_LINE = ( ) -def _build_llm_context(): - """Build the LLMContext with the walking-skeleton system prompt.""" +def _build_llm_context(scenario_runtime=None): + """Build the LLMContext with the scenario-driven system prompt (TASK-03-07). + + If a scenario_runtime is provided, uses scenario.setup.system_prompt. + Otherwise falls back to the SLICE-02 walking-skeleton prompt. + """ from pipecat.processors.aggregators.llm_context import LLMContext + if scenario_runtime is not None: + system_prompt = scenario_runtime.system_prompt + else: + system_prompt = WALKING_SKELETON_SYSTEM_PROMPT + messages = [ - {"role": "system", "content": WALKING_SKELETON_SYSTEM_PROMPT}, + {"role": "system", "content": system_prompt}, ] return LLMContext(messages=messages) @@ -131,11 +140,17 @@ def _build_vad_analyzer() -> Any: return SileroVADAnalyzer() -def build_pipeline(webrtc_connection): +def build_pipeline(webrtc_connection, *, scenario_id: str | None = None): """Assemble the full Pipecat pipeline + task + runner for one WebRTC session. - Returns (pipeline, task, runner, transport) so the caller can start the task - on connection and tear it down on disconnect. + Args: + webrtc_connection: a SmallWebRTCConnection with an accepted offer. + scenario_id: if set, load the scenario and use its system prompt + opening + line (TASK-03-07). If None, falls back to the walking-skeleton prompt. + + Returns (pipeline, task, runner, transport, scenario_runtime) so the caller + can start the task on connection, play the opening line, and run the branch + classifier + debrief at session end. """ from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner @@ -144,6 +159,22 @@ def build_pipeline(webrtc_connection): LLMContextAggregator, ) + # Load the scenario runtime (TASK-03-03, TASK-03-07). + scenario_runtime = None + if scenario_id: + try: + from server.scenarios.runtime import build_runtime_from_id + + scenario_runtime = build_runtime_from_id(scenario_id) + logger.info( + f"Loaded scenario {scenario_id!r}: branches={scenario_runtime.scenario.branch_ids()}" + ) + except Exception as exc: + logger.warning( + f"Could not load scenario {scenario_id!r}: {exc}. " + f"Falling back to walking-skeleton prompt." + ) + transport = _build_transport(webrtc_connection) stt = _build_stt() llm = _build_llm() @@ -153,7 +184,7 @@ def build_pipeline(webrtc_connection): latency_observer = LatencyObserver() - context = _build_llm_context() + context = _build_llm_context(scenario_runtime) user_aggregator = LLMContextAggregator(context=context, role="user") assistant_aggregator = LLMContextAggregator(context=context, role="assistant") @@ -182,11 +213,19 @@ def build_pipeline(webrtc_connection): ) runner = PipelineRunner(handle_sigint=False) - return pipeline, task, runner, transport + return pipeline, task, runner, transport, scenario_runtime + + +def build_runtime_from_id(scenario_id: str): + """Re-export of the scenario runtime builder (TASK-03-07).""" + from server.scenarios.runtime import build_runtime_from_id as _br + + return _br(scenario_id) __all__ = [ "build_pipeline", + "build_runtime_from_id", "WALKING_SKELETON_SYSTEM_PROMPT", "WALKING_SKELETON_OPENING_LINE", ] \ No newline at end of file diff --git a/server/scenarios/runtime.py b/server/scenarios/runtime.py new file mode 100644 index 0000000..18d57ff --- /dev/null +++ b/server/scenarios/runtime.py @@ -0,0 +1,103 @@ +"""Scenario runtime — maps a Scenario to a Pipecat Flows state machine (TASK-03-03). + +The v0.1 branch point is a post-hoc outcome classification (G-002): the +conversation is linear, and at session end an LLM-as-judge (TASK-03-06) +classifies the learner's signals into accept_resolution or escalate. Pipecat +Flows is wired so the branch field is part of the data model; Phase 2+ can +activate true in-flight branching without a schema change. + +This module: + - builds the system prompt from scenario.setup.system_prompt + - provides the opening line (scenario.setup.opening_line) as the first TTS utterance + - exposes the branch transition logic (driven by the classifier in TASK-03-06) + +TASK-03-07: the pipeline uses scenario-driven prompts instead of the +SLICE-02 hardcoded walking-skeleton prompt. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from server.scenarios.schema import Branch, Scenario + + +@dataclass +class ScenarioRuntime: + """Runtime state for one scenario session.""" + + scenario: Scenario + branch: Branch | None = None + turn_count: int = 0 + + @property + def system_prompt(self) -> str: + return self.scenario.setup.system_prompt + + @property + def opening_line(self) -> str: + return self.scenario.setup.opening_line + + @property + def branch_id(self) -> str | None: + return self.branch.id if self.branch else None + + @property + def outcome(self) -> str | None: + return self.branch.outcome if self.branch else None + + def set_branch(self, branch_id: str) -> Branch: + """Set the session's branch outcome (from the classifier, TASK-03-06).""" + b = self.scenario.branch_by_id(branch_id) + if b is None: + raise ValueError( + f"Unknown branch id {branch_id!r} for scenario {self.scenario.id!r}; " + f"known: {self.scenario.branch_ids()}" + ) + self.branch = b + return b + + def debrief_focus(self) -> str: + """The debrief focus for the resolved branch (or a default).""" + if self.branch: + return self.branch.debrief_focus + return "General coaching feedback for this session." + + def as_flow_spec(self) -> dict[str, Any]: + """Render the scenario as a Pipecat Flows state-machine spec. + + v0.1: a single 'conversation' state with the system prompt; branch + transitions are post-hoc (G-002). The spec carries the branch metadata + so Phase 2+ can fork in-flight. + """ + return { + "initial_state": "conversation", + "states": { + "conversation": { + "system_prompt": self.system_prompt, + "opening_line": self.opening_line, + "branches": [ + {"id": b.id, "outcome": b.outcome, + "learner_signals": b.trigger.learner_signals} + for b in self.scenario.branches + ], + }, + }, + "transitions": [], # v0.1: no in-flight transitions (G-002) + } + + +def build_runtime(scenario: Scenario) -> ScenarioRuntime: + """Construct a ScenarioRuntime for the given scenario.""" + return ScenarioRuntime(scenario=scenario) + + +def build_runtime_from_id(scenario_id: str) -> ScenarioRuntime: + """Load + build a runtime by scenario id (convenience for the pipeline).""" + from server.scenarios.loader import load + + return build_runtime(load(scenario_id)) + + +__all__ = ["ScenarioRuntime", "build_runtime", "build_runtime_from_id"] \ No newline at end of file diff --git a/tests/test_scenario_runtime.py b/tests/test_scenario_runtime.py new file mode 100644 index 0000000..df5dc32 --- /dev/null +++ b/tests/test_scenario_runtime.py @@ -0,0 +1,68 @@ +"""Unit tests for the scenario runtime + flows spec (TASK-03-03, TASK-03-07).""" + +from __future__ import annotations + +import pytest + +from server.scenarios.runtime import ( + ScenarioRuntime, + build_runtime, + build_runtime_from_id, +) +from server.scenarios.loader import load + + +def test_runtime_uses_scenario_system_prompt(): + s = load("customer_service_refund_ca_v01") + rt = build_runtime(s) + assert "Jordan" in rt.system_prompt + assert "cracked" in rt.opening_line + + +def test_runtime_set_branch_escalate(): + rt = build_runtime_from_id("customer_service_refund_ca_v01") + b = rt.set_branch("escalate") + assert b.outcome == "failure" + assert b.failure_mode == "escalates_unresolved" + assert rt.outcome == "failure" + assert rt.branch_id == "escalate" + + +def test_runtime_set_branch_accept(): + rt = build_runtime_from_id("customer_service_refund_ca_v01") + b = rt.set_branch("accept_resolution") + assert b.outcome == "success" + assert rt.outcome == "success" + + +def test_runtime_set_branch_unknown_raises(): + rt = build_runtime_from_id("customer_service_refund_ca_v01") + with pytest.raises(ValueError, match="Unknown branch id"): + rt.set_branch("nonexistent_branch") + + +def test_runtime_debrief_focus_per_branch(): + rt = build_runtime_from_id("customer_service_refund_ca_v01") + # No branch set → default focus. + assert "General" in rt.debrief_focus() + rt.set_branch("escalate") + assert "escalated" in rt.debrief_focus().lower() + rt.set_branch("accept_resolution") + assert "did well" in rt.debrief_focus().lower() + + +def test_runtime_as_flow_spec_has_branches(): + rt = build_runtime_from_id("customer_service_refund_ca_v01") + spec = rt.as_flow_spec() + assert spec["initial_state"] == "conversation" + assert "system_prompt" in spec["states"]["conversation"] + assert len(spec["states"]["conversation"]["branches"]) == 2 + # v0.1: no in-flight transitions (G-002 — post-hoc classification). + assert spec["transitions"] == [] + + +def test_runtime_default_debrief_model(): + """The scenario's debrief config uses deepseek-v4-flash:cloud no_think (D-020).""" + rt = build_runtime_from_id("customer_service_refund_ca_v01") + assert rt.scenario.debrief.model == "deepseek-v4-flash:cloud" + assert rt.scenario.debrief.mode == "no_think" \ No newline at end of file