Spaces:
Sleeping
Sleeping
File size: 5,938 Bytes
8277290 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 | import json
from datetime import datetime
from app.contracts import EngineRequest, EngineResponse, EngineError
from app.hf_client import HFClient
class OBLEngine:
def __init__(self):
self.client = HFClient()
async def process(self, request: EngineRequest) -> EngineResponse:
try:
action = request.action.lower()
if action == "create_outcome_contract":
return await self._create_outcome_contract(request)
elif action == "decompose_outcome":
return await self._decompose_outcome(request)
elif action == "generate_path":
return await self._generate_path(request)
elif action == "update_readiness":
return await self._update_readiness(request)
else:
raise ValueError(f"Unknown action: {action}")
except Exception as e:
return EngineResponse(
request_id=request.request_id,
ok=False,
status="error",
engine="obl-engine",
action=request.action,
error=EngineError(code="ENGINE_EXECUTION_ERROR", detail=str(e))
)
async def _create_outcome_contract(self, request: EngineRequest) -> EngineResponse:
outcome_text = request.input.text or "Not specified"
prompt = (
f"User Goal: {outcome_text}\n"
"Create a structured 'Outcome Contract' for this goal. It represents a verified result (job skill, exam pass, etc.). "
"Output JSON only with fields: "
"'outcome_statement', 'time_horizon', 'proof_criteria' (list of specific evidences), 'engagement_rules', 'pricing_model_type'."
)
messages = [{"role": "system", "content": prompt}]
response_text = await self.client.generate(messages)
try:
# Simple cleanup for JSON extraction
contract_data = json.loads(response_text.replace("```json", "").replace("```", "").strip())
except:
contract_data = {"raw_output": response_text, "note": "Failed to parse JSON contract."}
return EngineResponse(
request_id=request.request_id,
ok=True,
status="success",
engine="obl-engine",
action="create_outcome_contract",
result={"contract_draft": contract_data}
)
async def _decompose_outcome(self, request: EngineRequest) -> EngineResponse:
contract = request.input.refs.get("contract", {})
prompt = (
f"Analyze this Outcome Contract: {json.dumps(contract)}\n"
"Decompose it into:\n"
"1. Skills (What they must do)\n"
"2. Proofs (How we verify it - exams, projects, simulations)\n"
"Output JSON with keys: 'skills' (list of objects with id, name), 'proofs' (list of objects with id, type, rubric)."
)
messages = [{"role": "system", "content": prompt}]
response_text = await self.client.generate(messages)
try:
decomposition = json.loads(response_text.replace("```json", "").replace("```", "").strip())
except:
decomposition = {"raw_output": response_text}
return EngineResponse(
request_id=request.request_id,
ok=True,
status="success",
engine="obl-engine",
action="decompose_outcome",
result=decomposition
)
async def _generate_path(self, request: EngineRequest) -> EngineResponse:
skills = request.input.refs.get("skills", [])
proofs = request.input.refs.get("proofs", [])
prompt = (
f"Skills: {json.dumps(skills)}\nProofs: {json.dumps(proofs)}\n"
"Generate a 'shortest path' learning plan. Prioritize actions that lead directly to proofs. "
"Output JSON key 'plan' as a list of steps. Each step has: 'step_id', 'action' (Lesson/Practice/Sim/Teachback), 'linked_proof_id'."
)
messages = [{"role": "system", "content": prompt}]
response_text = await self.client.generate(messages)
try:
plan = json.loads(response_text.replace("```json", "").replace("```", "").strip())
except:
plan = {"raw_output": response_text}
return EngineResponse(
request_id=request.request_id,
ok=True,
status="success",
engine="obl-engine",
action="generate_path",
result=plan
)
async def _update_readiness(self, request: EngineRequest) -> EngineResponse:
evidence = request.input.text # e.g., "Passed quiz with 90%"
current_readiness = request.input.refs.get("readiness_score", 0.0)
prompt = (
f"Current Readiness: {current_readiness}\n"
f"New Evidence: {evidence}\n"
"Estimate the new readiness score (0.0 to 1.0) and identify any risk flags. "
"Output JSON: 'new_readiness_score', 'risk_flags' (list)."
)
messages = [{"role": "system", "content": prompt}]
response_text = await self.client.generate(messages)
try:
update_data = json.loads(response_text.replace("```json", "").replace("```", "").strip())
except:
update_data = {"new_readiness_score": current_readiness, "note": "Failed to parse update."}
return EngineResponse(
request_id=request.request_id,
ok=True,
status="success",
engine="obl-engine",
action="update_readiness",
result=update_data
)
engine = OBLEngine()
|