70 lines
2.4 KiB
Python
70 lines
2.4 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
POLICY_CANDIDATE = Path('/home/openclaw/.openclaw/workspace/data/policy_candidate.json')
|
||
|
|
|
||
|
|
|
||
|
|
def load_policy_candidate(path: Path = POLICY_CANDIDATE) -> dict[str, Any]:
|
||
|
|
try:
|
||
|
|
return json.loads(path.read_text(encoding='utf-8'))
|
||
|
|
except Exception:
|
||
|
|
return {'plans': {}, 'families': {}, 'generated_at': ''}
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def plan_prior(policy: dict[str, Any], chosen_plan: str) -> dict[str, Any]:
|
||
|
|
row = ((policy or {}).get('plans') or {}).get(chosen_plan or '', {}) or {}
|
||
|
|
return {
|
||
|
|
'mode': row.get('mode', 'observe'),
|
||
|
|
'beta_mean': float(row.get('beta_mean', 0.5) or 0.5),
|
||
|
|
'count': int(row.get('count', 0) or 0),
|
||
|
|
'alpha': float(row.get('alpha', 1.0) or 1.0),
|
||
|
|
'beta': float(row.get('beta', 1.0) or 1.0),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def family_priors(policy: dict[str, Any], families: list[str]) -> list[dict[str, Any]]:
|
||
|
|
out = []
|
||
|
|
bucket = (policy or {}).get('families') or {}
|
||
|
|
for fam in families or []:
|
||
|
|
row = bucket.get(fam, {}) or {}
|
||
|
|
out.append({
|
||
|
|
'family': fam,
|
||
|
|
'mode': row.get('mode', 'observe'),
|
||
|
|
'beta_mean': float(row.get('beta_mean', 0.5) or 0.5),
|
||
|
|
'count': int(row.get('count', 0) or 0),
|
||
|
|
'alpha': float(row.get('alpha', 1.0) or 1.0),
|
||
|
|
'beta': float(row.get('beta', 1.0) or 1.0),
|
||
|
|
})
|
||
|
|
return out
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def apply_bandit_bias(*, base_decision: str, base_reason: str, chosen_plan: str, families: list[str], uncertainty: dict[str, Any], policy: dict[str, Any]) -> dict[str, Any]:
|
||
|
|
plan = plan_prior(policy, chosen_plan)
|
||
|
|
fams = family_priors(policy, families)
|
||
|
|
decision = base_decision
|
||
|
|
reason = base_reason
|
||
|
|
|
||
|
|
if plan['mode'] == 'prefer' and plan['beta_mean'] >= 0.65:
|
||
|
|
decision = 'run_plan'
|
||
|
|
reason = f'policy_prefers_plan:{chosen_plan}'
|
||
|
|
elif plan['mode'] == 'avoid' and uncertainty.get('level') in {'medium', 'high'}:
|
||
|
|
if 'ambiguous_access' in families:
|
||
|
|
decision = 'ask_clarification'
|
||
|
|
reason = f'policy_avoids_plan:{chosen_plan}'
|
||
|
|
elif base_decision == 'run_plan':
|
||
|
|
decision = 'answer_direct'
|
||
|
|
reason = f'policy_deescalates_plan:{chosen_plan}'
|
||
|
|
|
||
|
|
return {
|
||
|
|
'decision': decision,
|
||
|
|
'reason': reason,
|
||
|
|
'plan_prior': plan,
|
||
|
|
'family_priors': fams,
|
||
|
|
}
|