100 lines
2.5 KiB
Python
100 lines
2.5 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from dataclasses import dataclass, asdict
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class ToolNode:
|
||
|
|
name: str
|
||
|
|
kind: str
|
||
|
|
description: str
|
||
|
|
input_schema: str
|
||
|
|
output_schema: str
|
||
|
|
effect_type: str
|
||
|
|
risk: str
|
||
|
|
latency_class: str
|
||
|
|
groundedness: str
|
||
|
|
cost_class: str
|
||
|
|
|
||
|
|
def to_dict(self) -> dict[str, Any]:
|
||
|
|
return asdict(self)
|
||
|
|
|
||
|
|
|
||
|
|
KIND_TO_EFFECT = {
|
||
|
|
'action': 'state_change',
|
||
|
|
'evidence': 'evidence_only',
|
||
|
|
'final': 'answer_only',
|
||
|
|
}
|
||
|
|
|
||
|
|
KIND_TO_RISK = {
|
||
|
|
'action': 'high',
|
||
|
|
'evidence': 'low',
|
||
|
|
'final': 'medium',
|
||
|
|
}
|
||
|
|
|
||
|
|
KIND_TO_OUTPUT = {
|
||
|
|
'action': 'action_result',
|
||
|
|
'evidence': 'evidence_blob',
|
||
|
|
'final': 'final_answer',
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def classify_latency(name: str, kind: str) -> str:
|
||
|
|
if name in {'url_research', 'web_research', 'ops_deep_analyze'}:
|
||
|
|
return 'slow'
|
||
|
|
if kind == 'action':
|
||
|
|
return 'medium'
|
||
|
|
if kind == 'evidence':
|
||
|
|
return 'fast'
|
||
|
|
return 'medium'
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def classify_cost(name: str, kind: str) -> str:
|
||
|
|
if name in {'web_research', 'url_research', 'ops_deep_analyze'}:
|
||
|
|
return 'high'
|
||
|
|
if kind == 'action':
|
||
|
|
return 'medium'
|
||
|
|
return 'low'
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def classify_groundedness(name: str, kind: str) -> str:
|
||
|
|
if name in {'setup_lookup', 'memory_profile', 'web_root_cause', 'user_service_access_diagnose', 'light_status', 'emby_user_provision'}:
|
||
|
|
return 'strong'
|
||
|
|
if kind == 'evidence':
|
||
|
|
return 'strong'
|
||
|
|
if name in {'general_answer', 'personal_assist', 'expert_write', 'expert_strategy'}:
|
||
|
|
return 'weak'
|
||
|
|
return 'medium'
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def infer_input_schema(name: str) -> str:
|
||
|
|
if name == 'memory_profile':
|
||
|
|
return 'user_query+memory_mode'
|
||
|
|
if name == 'setup_lookup':
|
||
|
|
return 'user_query+service_hint'
|
||
|
|
return 'user_query'
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def build_tool_graph(tool_registry: dict[str, dict[str, Any]]) -> dict[str, ToolNode]:
|
||
|
|
graph: dict[str, ToolNode] = {}
|
||
|
|
for name, info in (tool_registry or {}).items():
|
||
|
|
kind = str(info.get('kind') or 'final')
|
||
|
|
graph[name] = ToolNode(
|
||
|
|
name=name,
|
||
|
|
kind=kind,
|
||
|
|
description=str(info.get('description') or ''),
|
||
|
|
input_schema=infer_input_schema(name),
|
||
|
|
output_schema=KIND_TO_OUTPUT.get(kind, 'opaque'),
|
||
|
|
effect_type=KIND_TO_EFFECT.get(kind, 'answer_only'),
|
||
|
|
risk=KIND_TO_RISK.get(kind, 'medium'),
|
||
|
|
latency_class=classify_latency(name, kind),
|
||
|
|
groundedness=classify_groundedness(name, kind),
|
||
|
|
cost_class=classify_cost(name, kind),
|
||
|
|
)
|
||
|
|
return graph
|