393 lines
14 KiB
Python
393 lines
14 KiB
Python
"""
|
|
F14 Intent Signature System (ISS)
|
|
|
|
Deterministic inference of user intent from workflow + runtime signals.
|
|
|
|
Architecture (P2):
|
|
- definitions: Builtin intent definitions (JSON)
|
|
- extractor: Signal extraction from workflow
|
|
- scorer: Intent matching with confidence calculation
|
|
|
|
ISS is explicitly NOT:
|
|
- ML/LLM-based (deterministic rules only)
|
|
- Remote/cloud-dependent (all local)
|
|
- PII-capturing (sanitized signals only)
|
|
"""
|
|
|
|
import logging
|
|
from typing import Dict, Any, List, Optional
|
|
|
|
from ..time_utils import utc_isoformat
|
|
from ..diagnostics.models import (
|
|
IntentSignature,
|
|
IntentMatch,
|
|
SignalEvidence,
|
|
SignalSource,
|
|
)
|
|
from .loader import load_intents # P2: Load from JSON
|
|
|
|
logger = logging.getLogger("comfyui-doctor.intent")
|
|
|
|
|
|
# ============================================================================
|
|
# Intent Definitions (P2)
|
|
# ============================================================================
|
|
|
|
# Schema version for ISS
|
|
ISS_SCHEMA_VERSION = "1.0"
|
|
|
|
|
|
# ============================================================================
|
|
# Signal Extractor (P2)
|
|
# ============================================================================
|
|
|
|
class SignalExtractor:
|
|
"""
|
|
Extracts signals from workflow for intent matching.
|
|
|
|
Signal types:
|
|
- node_type.X: Presence of node type X
|
|
- node_count.X: Count of node type X
|
|
- edge.X_Y: Connection from node type X to Y
|
|
"""
|
|
|
|
def extract(self, workflow: Dict[str, Any]) -> List[SignalEvidence]:
|
|
"""
|
|
Extract signals from workflow.
|
|
|
|
Args:
|
|
workflow: Workflow JSON (graph format or API format)
|
|
|
|
Returns:
|
|
List of extracted signals
|
|
"""
|
|
signals: List[SignalEvidence] = []
|
|
|
|
# P3: Normalize nodes to list of dicts (handle list vs dict format)
|
|
raw_nodes = workflow.get("nodes", [])
|
|
nodes: List[Dict[str, Any]] = []
|
|
|
|
if isinstance(raw_nodes, list):
|
|
nodes = raw_nodes
|
|
elif isinstance(raw_nodes, dict):
|
|
# API format: "nodes": { "1": { ... }, "2": { ... } }
|
|
nodes = list(raw_nodes.values())
|
|
|
|
links = workflow.get("links", [])
|
|
# links format in Comfy: [[id, source_id, source_enc, target_id, target_enc, type], ...]
|
|
|
|
# Build node maps for edge detection
|
|
node_id_to_type = {}
|
|
node_types: Dict[str, List[int]] = {} # type -> [node_ids]
|
|
|
|
for node in nodes:
|
|
if not isinstance(node, dict):
|
|
continue
|
|
|
|
node_type = node.get("type") or node.get("class_type") or ""
|
|
node_id = node.get("id")
|
|
|
|
if node_type and node_id is not None:
|
|
# Store for edge detection
|
|
try:
|
|
# node_id might be string or int in JSON, simpler to map as is
|
|
node_id_to_type[node_id] = node_type
|
|
# Store as int for consistency if possible, else str
|
|
nid_safe = int(node_id) if str(node_id).isdigit() else str(node_id)
|
|
|
|
if node_type not in node_types:
|
|
node_types[node_type] = []
|
|
node_types[node_type].append(nid_safe)
|
|
except ValueError:
|
|
pass
|
|
|
|
# 1. Node Type & Count Signals
|
|
for node_type, node_ids in node_types.items():
|
|
# Presence signal
|
|
signals.append(SignalEvidence(
|
|
signal_id=f"node_type.{node_type}",
|
|
weight=1.0, # Base weight, actual impact defined in intent JSON
|
|
value=True,
|
|
source=SignalSource.WORKFLOW,
|
|
node_ids=node_ids[:20], # cap list size
|
|
explain=_sanitize_evidence_string(f"Node type '{node_type}' present"),
|
|
))
|
|
|
|
# Count signal (if multiple)
|
|
if len(node_ids) > 1:
|
|
signals.append(SignalEvidence(
|
|
signal_id=f"node_count.{node_type}",
|
|
weight=0.5,
|
|
value=len(node_ids),
|
|
source=SignalSource.WORKFLOW,
|
|
node_ids=node_ids[:20],
|
|
explain=_sanitize_evidence_string(f"{len(node_ids)} instances of '{node_type}'"),
|
|
))
|
|
|
|
# 2. Edge Signals
|
|
if isinstance(links, list):
|
|
unique_edges = set()
|
|
|
|
for link in links:
|
|
# defensive check against malformed link arrays
|
|
if not isinstance(link, list) or len(link) < 5:
|
|
continue
|
|
|
|
# link schema: [id, origin_id, origin_slot, target_id, target_slot, type]
|
|
try:
|
|
origin_id = link[1]
|
|
target_id = link[3]
|
|
|
|
origin_type = node_id_to_type.get(origin_id) or node_id_to_type.get(str(origin_id)) or node_id_to_type.get(int(origin_id) if str(origin_id).isdigit() else None)
|
|
target_type = node_id_to_type.get(target_id) or node_id_to_type.get(str(target_id)) or node_id_to_type.get(int(target_id) if str(target_id).isdigit() else None)
|
|
|
|
if origin_type and target_type:
|
|
edge_key = f"edge.{origin_type}_{target_type}"
|
|
if edge_key not in unique_edges:
|
|
unique_edges.add(edge_key)
|
|
signals.append(SignalEvidence(
|
|
signal_id=edge_key,
|
|
weight=0.8,
|
|
value=True,
|
|
source=SignalSource.WORKFLOW,
|
|
node_ids=[origin_id, target_id],
|
|
explain=_sanitize_evidence_string(f"Connection: {origin_type} -> {target_type}"),
|
|
))
|
|
except (IndexError, ValueError):
|
|
continue
|
|
|
|
return signals
|
|
|
|
|
|
# ============================================================================
|
|
# Helpers (P3)
|
|
# ============================================================================
|
|
|
|
def _sanitize_evidence_string(text: str, max_len: int = 120) -> str:
|
|
"""
|
|
Sanitize evidence string to prevent PII leakage and enforce bounds.
|
|
"""
|
|
if not text:
|
|
return ""
|
|
|
|
# Cap length first to avoid regex on huge strings
|
|
if len(text) > max_len:
|
|
text = text[:max_len] + "..."
|
|
|
|
# Redact Windows Paths (e.g. C:\Users\...)
|
|
# Look for drive letter + colon + backslash + chars allowed in paths
|
|
import re
|
|
# Greedy match for path-like characters after C:\
|
|
text = re.sub(r'[a-zA-Z]:\\[\w\s\-\.\\]+', '[REDACTED]', text)
|
|
|
|
# Redact Unix Paths (e.g. /home/user/...)
|
|
# Look for common roots
|
|
text = re.sub(r'/(?:home|Users|var|etc|tmp)/[\w\-\./]+', '[REDACTED]', text)
|
|
|
|
# Redact Bearer Tokens
|
|
text = re.sub(r'Bearer\s+[a-zA-Z0-9\-\._]+', 'Bearer [REDACTED]', text)
|
|
|
|
# Redact common API Key patterns (Conservative)
|
|
# OpenAI (sk-...), Google (AIza...), HuggingFace (hf_...)
|
|
text = re.sub(r'\bsk-[a-zA-Z0-9]{20,}\b', '[REDACTED]', text)
|
|
text = re.sub(r'\bAIza[a-zA-Z0-9\-_]{30,}\b', '[REDACTED]', text)
|
|
text = re.sub(r'\bhf_[a-zA-Z0-9]{20,}\b', '[REDACTED]', text)
|
|
|
|
return text
|
|
|
|
|
|
# ============================================================================
|
|
# Intent Scorer (P2)
|
|
# ============================================================================
|
|
|
|
class IntentScorer:
|
|
"""
|
|
Scores intents based on extracted signals.
|
|
|
|
Scoring algorithm:
|
|
- Required signals must all be present (confidence = 0 if missing)
|
|
- Positive signals add to confidence
|
|
- Negative signals subtract from confidence
|
|
- Final confidence normalized and clamped
|
|
"""
|
|
|
|
def __init__(self, intents: Optional[Dict[str, Dict[str, Any]]] = None):
|
|
"""
|
|
Initialize scorer with intent definitions.
|
|
|
|
Args:
|
|
intents: Intent definitions dict (default: loaded from JSON)
|
|
"""
|
|
self.intents = intents or load_intents()
|
|
self.extractor = SignalExtractor()
|
|
|
|
async def compute(
|
|
self,
|
|
workflow: Dict[str, Any],
|
|
workflow_hash: str,
|
|
top_k: int = 3,
|
|
) -> IntentSignature:
|
|
"""
|
|
Compute intent signature for a workflow.
|
|
|
|
Args:
|
|
workflow: Workflow JSON
|
|
workflow_hash: Pre-computed workflow hash
|
|
top_k: Number of top intents to return
|
|
|
|
Returns:
|
|
IntentSignature with matched intents
|
|
"""
|
|
# Extract signals
|
|
signals = self.extractor.extract(workflow)
|
|
|
|
# Optimization: Map signal_id to signal object for O(1) lookups
|
|
# Also supports partial matches if needed, but strict mapping first
|
|
signal_map = {s.signal_id: s for s in signals}
|
|
|
|
# To support "contains" matching (like node_type.KSampler matching node_type.KSamplerAdvanced if configured),
|
|
# we can build a set of IDs. For P2, exact string matching or prefix matching depends on definitions.
|
|
# Our JSON uses exact keys like "node_type.KSampler", so let's stick to set lookup.
|
|
available_signal_ids = set(signal_map.keys())
|
|
|
|
# Score each intent
|
|
scored_intents: List[tuple[str, float, List[SignalEvidence]]] = []
|
|
|
|
for intent_id, intent_def in self.intents.items():
|
|
required = intent_def.get("required_signals", [])
|
|
positive = intent_def.get("positive_signals", [])
|
|
negative = intent_def.get("negative_signals", [])
|
|
|
|
# 1. Check required signals
|
|
# Allow flexible matching: if req="node_type.KSampler", we look for exactly that signal.
|
|
# Could extended to wildcard later, but P2 scope is deterministic exact/prefix.
|
|
|
|
missing_required = False
|
|
req_evidence: List[SignalEvidence] = []
|
|
|
|
for req in required:
|
|
# Find matching signal
|
|
# For now, simplistic check: is req in available_signal_ids?
|
|
if req in available_signal_ids:
|
|
req_evidence.append(signal_map[req])
|
|
else:
|
|
missing_required = True
|
|
break
|
|
|
|
if missing_required and required:
|
|
continue
|
|
|
|
# 2. Calculate base confidence
|
|
# Heuristic: Start at 0.0
|
|
# Each required signal (met) adds significant confidence
|
|
# Each positive signal adds moderate confidence
|
|
# Each negative signal subtracts confidence
|
|
|
|
confidence = 0.0
|
|
evidence: List[SignalEvidence] = []
|
|
evidence.extend(req_evidence)
|
|
|
|
# Base score from meeting strict requirements
|
|
if required:
|
|
confidence += 0.4
|
|
else:
|
|
# If no requirements (e.g. video), start lower
|
|
confidence += 0.1
|
|
|
|
# Add positives
|
|
for pos in positive:
|
|
if pos in available_signal_ids:
|
|
confidence += 0.2
|
|
evidence.append(signal_map[pos])
|
|
|
|
# Subtract negatives
|
|
for neg in negative:
|
|
if neg in available_signal_ids:
|
|
confidence -= 0.3
|
|
# Negative evidence usually interesting to show why score is low?
|
|
# Actually, usually we show evidence for why it IS matched.
|
|
|
|
# Normalize/Clamp
|
|
confidence = max(0.0, min(1.0, confidence))
|
|
|
|
if confidence > 0.1: # Threshold to be considered a candidate
|
|
scored_intents.append((intent_id, confidence, evidence))
|
|
|
|
# Sort: Primary = Confidence (Desc), Secondary = ID (Asc) for stability
|
|
scored_intents.sort(key=lambda x: (-x[1], x[0]))
|
|
|
|
top_candidates = scored_intents[:top_k]
|
|
|
|
# Build IntentMatches
|
|
intent_matches: List[IntentMatch] = []
|
|
for intent_id, confidence, evidence in top_candidates:
|
|
intent_def = self.intents[intent_id]
|
|
|
|
# Deduplicate evidence
|
|
unique_evidence = []
|
|
seen_sigs = set()
|
|
for ev in evidence:
|
|
if ev.signal_id not in seen_sigs:
|
|
unique_evidence.append(ev)
|
|
seen_sigs.add(ev.signal_id)
|
|
|
|
intent_matches.append(IntentMatch(
|
|
intent_id=intent_id,
|
|
confidence=round(confidence, 2),
|
|
stage=intent_def.get("stage"),
|
|
evidence=unique_evidence[:5], # Cap evidence items per intent
|
|
))
|
|
|
|
# Global signals: High-value signals for debugging (e.g. key nodes found)
|
|
# Sort by weight? For now, just node types.
|
|
global_signals = [s for s in signals if s.signal_id.startswith("node_type.")][:10]
|
|
|
|
return IntentSignature(
|
|
schema_version=ISS_SCHEMA_VERSION,
|
|
timestamp=utc_isoformat(),
|
|
workflow_hash=workflow_hash,
|
|
top_intents=intent_matches,
|
|
global_signals=global_signals,
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# Module Interface
|
|
# ============================================================================
|
|
|
|
_scorer: Optional[IntentScorer] = None
|
|
|
|
|
|
def get_intent_scorer() -> IntentScorer:
|
|
"""Get or create the global IntentScorer instance."""
|
|
global _scorer
|
|
if _scorer is None:
|
|
_scorer = IntentScorer()
|
|
return _scorer
|
|
|
|
|
|
def init_intent_system():
|
|
"""
|
|
Initialize the intent signature system.
|
|
|
|
Called during module initialization to:
|
|
1. Load intent definitions
|
|
2. Register scorer with diagnostics runner
|
|
"""
|
|
from ..diagnostics import get_diagnostics_runner
|
|
|
|
scorer = get_intent_scorer()
|
|
runner = get_diagnostics_runner()
|
|
runner.set_intent_scorer(scorer)
|
|
|
|
logger.info(f"Intent system initialized with P2 extractor")
|
|
|
|
|
|
__all__ = [
|
|
"ISS_SCHEMA_VERSION",
|
|
"SignalExtractor",
|
|
"IntentScorer",
|
|
"get_intent_scorer",
|
|
"init_intent_system",
|
|
]
|