Spaces:
Sleeping
Sleeping
| """ | |
| 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 | |