62 lines
2.4 KiB
Python
62 lines
2.4 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from tool_graph import build_tool_graph
|
||
|
|
from uncertainty_model import estimate_uncertainty
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def shadow_decision(message: str, analysis: dict[str, Any], family_candidates: list[dict[str, Any]] | None, tool_registry: dict[str, dict[str, Any]]) -> dict[str, Any]:
|
||
|
|
graph = build_tool_graph(tool_registry)
|
||
|
|
uncertainty = estimate_uncertainty(message, analysis, family_candidates)
|
||
|
|
tools = list(analysis.get('tools') or [])
|
||
|
|
families = [str((item or {}).get('family') or '') for item in (family_candidates or []) if (item or {}).get('family')]
|
||
|
|
|
||
|
|
decision = 'answer_direct'
|
||
|
|
reason = 'single_grounded_or_low_uncertainty'
|
||
|
|
suggested_memory_mode = ''
|
||
|
|
|
||
|
|
if uncertainty['level'] == 'high' and 'ambiguous_access' in families:
|
||
|
|
decision = 'ask_clarification'
|
||
|
|
reason = 'ambiguous_service_access'
|
||
|
|
elif analysis.get('needs_memory') and analysis.get('needs_setup_context'):
|
||
|
|
decision = 'run_plan'
|
||
|
|
reason = 'mixed_memory_plus_setup'
|
||
|
|
suggested_memory_mode = 'setup'
|
||
|
|
elif analysis.get('needs_memory'):
|
||
|
|
decision = 'use_memory_mode'
|
||
|
|
reason = 'memory_required'
|
||
|
|
suggested_memory_mode = 'profile' if analysis.get('task_type') == 'memory' else 'preference'
|
||
|
|
elif analysis.get('needs_setup_context') or len(tools) > 1:
|
||
|
|
decision = 'run_plan'
|
||
|
|
reason = 'evidence_required'
|
||
|
|
elif uncertainty['level'] == 'medium' and graph.get(tools[0], None) and graph[tools[0]].groundedness == 'weak':
|
||
|
|
decision = 'run_plan'
|
||
|
|
reason = 'weak_grounding_under_uncertainty'
|
||
|
|
|
||
|
|
return {
|
||
|
|
'ts': datetime.now(timezone.utc).isoformat(),
|
||
|
|
'message': message,
|
||
|
|
'decision': decision,
|
||
|
|
'reason': reason,
|
||
|
|
'suggested_memory_mode': suggested_memory_mode,
|
||
|
|
'suggested_tools': tools,
|
||
|
|
'uncertainty': uncertainty,
|
||
|
|
'family_candidates': families,
|
||
|
|
'normalized_task': f"{analysis.get('role','')}:{analysis.get('task_type','')}",
|
||
|
|
'chosen_plan': str(analysis.get('composition_reason') or 'single_tool'),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def log_shadow_decision(log_path, decision_row: dict[str, Any]) -> None:
|
||
|
|
try:
|
||
|
|
log_path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
with log_path.open('a', encoding='utf-8') as f:
|
||
|
|
f.write(json.dumps(decision_row, ensure_ascii=False) + '\n')
|
||
|
|
except Exception:
|
||
|
|
pass
|