Create run_two_phase.py

pipeline/run_two_phase.py

@@ -0,0 +1,93 @@
+"""
+Reference pipeline orchestrator.
+This module shows how to wire the two-phase flow. Replace `call_model` with your LLM runner.
+"""
+
+from pathlib import Path
+from typing import Tuple, Dict, Any
+import json
+
+from ..validators import schema_validator, unit_validator, math_validator, policy_validator
+from ..graders.rule_grader import grade
+from .io_utils import load_json
+
+# -------------------------
+# Replace this with your LLM runner.
+def call_model(system_prompt: str, user_prompt: str, temperature: float = 0.2, top_p: float = 0.9) -> str:
+    """
+    Stub for LLM call. Should return the model's raw text output.
+    In production: plug into your provider (OpenAI, Azure, Anthropic, etc.)
+    """
+    raise NotImplementedError("call_model must be implemented in your environment.")
+# -------------------------
+
+def build_user_prompt(template_text: str, context: str, data_inputs: str, constraints: str) -> str:
+    return (
+        template_text
+        .replace("{CONTEXT}", context)
+        .replace("{DATA_INPUTS}", data_inputs)
+        .replace("{CONSTRAINTS}", constraints)
+    )
+
+def run_clarityops(pack_dir: str) -> Tuple[Dict[str, Any], Any]:
+    pack = Path(pack_dir)
+    root = pack.parents[1]
+
+    system_prompt = (root / "prompts" / "system_two_phase.txt").read_text(encoding="utf-8")
+    user_template = (root / "prompts" / "user_template.txt").read_text(encoding="utf-8")
+
+    inputs = load_json(pack / "inputs.json")
+    constraints = load_json(pack / "constraints.json")
+    schema_cfg = load_json(pack / "schema.json")
+    rubric = load_json(pack / "rubric.json")
+    expected = load_json(pack / "expected.json")
+
+    # Build the human-readable blocks
+    context_block = inputs.get("context", "No context provided.")
+    data_block = json.dumps(inputs.get("data_inputs", {}), ensure_ascii=False, indent=2)
+    constraints_block = json.dumps(constraints, ensure_ascii=False, indent=2)
+
+    # ---- Phase 1: Clarification Questions
+    user_prompt_phase1 = build_user_prompt(user_template, context_block, data_block, constraints_block)
+    # Tell the model explicitly: generate Phase 1 only
+    user_prompt_phase1 += "\n\n[INSTRUCTION TO MODEL] Produce **Phase 1** only. Do not produce Phase 2 yet."
+
+    clarif_raw = call_model(system_prompt, user_prompt_phase1)
+    # Expect clarif_raw to contain either "No clarifications required" or a numbered list of questions.
+
+    # Option A: automated answers for CI (if provided)
+    clarif_answers_path = pack / "clarifications.json"
+    if clarif_answers_path.exists():
+        clarif_answers = load_json(clarif_answers_path)
+    else:
+        # Option B: interactive collection (replace as needed)
+        raise RuntimeError("Clarification answers required. Provide packs/<scenario>/clarifications.json or implement an interactive flow.")
+
+    # Merge clarifications into inputs for Phase 2
+    merged_inputs = inputs.copy()
+    merged_inputs["clarifications"] = clarif_answers
+
+    # ---- Phase 2: Structured Analysis
+    user_prompt_phase2 = build_user_prompt(user_template, context_block, json.dumps(merged_inputs, ensure_ascii=False, indent=2), constraints_block)
+    user_prompt_phase2 += "\n\n[INSTRUCTION TO MODEL] Produce **Phase 2** only (final structured analysis), using clarified inputs."
+
+    final_raw = call_model(system_prompt, user_prompt_phase2)
+
+    # Expect final_raw to be JSON or parseable. If your model returns markdown, strip code fences first.
+    try:
+        # naive parse: assume JSON
+        output = json.loads(final_raw)
+    except Exception as e:
+        raise ValueError(f"Failed to parse model output as JSON. Raw:\n{final_raw}") from e
+
+    # Validators
+    schema_validator.assert_valid(output, str(root / "schemas" / "analysis_output.schema.json"))
+    unit_validator.assert_valid(output, str(root / "core" / "policy_global.json"))
+    math_validator.assert_valid(output)
+    policy_validator.assert_valid(output, str(pack / "constraints.json"))
+
+    # Grading (optional): compare to expected
+    grader_result = grade(output, str(pack / "rubric.json"))
+    output["_grader"] = grader_result
+
+    return output, clarif_raw