openclaw-intelligence-core-.../syncpatch/agent-orchestrate

3609 lines
230 KiB
Text
Raw Permalink Normal View History

2026-03-21 07:34:09 +00:00
#!/usr/bin/env python3
import sys
sys.path.insert(0, '/home/openclaw/.openclaw/workspace/bin')
sys.path.insert(0, '/home/openclaw/.openclaw/workspace/lib')
try:
from buenzli_wrapper import buenzli_response
USE_BUENZLI = True
except:
USE_BUENZLI = False
def buenzli_response(t): return t
try:
from structured_router import classify_intent
USE_STRUCTURED_ROUTER = True
except:
USE_STRUCTURED_ROUTER = False
def classify_intent(m): return None
USE_BUENZLI = False
def buenzli_response(t): return t
try:
from intent_families import classify_intent_family, collect_intent_families
USE_INTENT_FAMILIES = True
except Exception:
USE_INTENT_FAMILIES = False
def classify_intent_family(_message, service_hints=None): return {'family': ''}
try:
from playbook_store import format_playbook_context
USE_PLAYBOOKS = True
except Exception:
USE_PLAYBOOKS = False
def format_playbook_context(_message, analysis=None, limit=None): return ''
try:
from plan_first import build_plan, render_plan_context
USE_PLAN_FIRST = True
except Exception:
USE_PLAN_FIRST = False
def build_plan(_message, _analysis, memory_ctx='', retrieval_ctx='', playbook_ctx=''): return None
def render_plan_context(_plan): return ''
try:
from intent_composition import compose_family_plan
USE_INTENT_COMPOSITION = True
except Exception:
USE_INTENT_COMPOSITION = False
def compose_family_plan(_message, _analysis, family_candidates=None): return {'composed': False}
try:
from composition_policy import choose_plan
USE_COMPOSITION_POLICY = True
except Exception:
USE_COMPOSITION_POLICY = False
def choose_plan(base_tools, proposed_tools, family_candidates=None):
return {'selected_tools': list(proposed_tools), 'policy': 'fallback_no_policy'}
try:
from result_state import persist_result_state as _persist_result_state
except Exception:
def _persist_result_state(
message,
analysis,
final_text,
evidence_items,
*,
should_skip_heavy_persistence,
update_topic_state,
compact_episode_text,
remember_episode,
record_incident_if_useful,
):
if should_skip_heavy_persistence(message, analysis, evidence_items):
return
update_topic_state(message, analysis, final_text)
remember_episode(compact_episode_text(message, analysis, final_text, evidence_items))
record_incident_if_useful(message, analysis, final_text, evidence_items)
try:
from final_answer_policy import should_skip_verify as _should_skip_verify
except Exception:
def _should_skip_verify(analysis, evidence_items, tool_kinds):
single_final = (
len(evidence_items) == 1
and tool_kinds.get(str(evidence_items[0].get('tool') or '')) == 'final'
and bool(evidence_items[0].get('grounded'))
)
if single_final:
return (
analysis.get('role') in {'ops', 'research'}
or analysis.get('task_type') in {'status', 'summarize', 'memory'}
or str(evidence_items[0].get('tool') or '') in {'memory_profile', 'light_status', 'url_research', 'public_services_health', 'emby_user_provision', 'web_root_cause'}
)
last_ev = evidence_items[-1] if evidence_items else {}
last_tool = str(last_ev.get('tool') or '')
prep_tools_ok = {'final', 'memory', 'evidence'}
composed_final = (
len(evidence_items) >= 2
and bool(analysis.get('force_sequential'))
and bool(analysis.get('composition_reason'))
and tool_kinds.get(last_tool) == 'final'
and bool(last_ev.get('grounded'))
and all(bool(item.get('grounded')) for item in evidence_items[:-1])
and all(tool_kinds.get(str(item.get('tool') or '')) in prep_tools_ok for item in evidence_items[:-1])
)
return composed_final and last_tool in {'personal_assist', 'user_service_access_diagnose', 'web_root_cause', 'url_research', 'general_answer', 'light_status'}
try:
from sequential_runner import run_sequential_tools as _run_sequential_tools
except Exception:
def _run_sequential_tools(
message,
tools,
analysis,
*,
is_complex_task,
run_tool,
evidence_object,
trace_stage,
sequential_tool_budget,
summarize_sequential_context,
):
collected = []
seq_ctx = ''
is_seq = bool(analysis.get('force_sequential')) or is_complex_task(message, analysis)
failed_ev = None
for name in tools:
msg = f"{message}\n\nBisherige Analyseergebnisse:\n{seq_ctx}" if is_seq and seq_ctx else message
prev_budget = os.environ.get('BIBABOT_TOOL_TIMEOUT')
budget = sequential_tool_budget(name, is_seq)
try:
if budget:
os.environ['BIBABOT_TOOL_TIMEOUT'] = str(budget)
rc, out, err = run_tool(name, msg)
finally:
if prev_budget is None:
os.environ.pop('BIBABOT_TOOL_TIMEOUT', None)
else:
os.environ['BIBABOT_TOOL_TIMEOUT'] = prev_budget
ev = evidence_object(name, rc, out, err)
collected.append(ev)
if out and is_seq:
seq_ctx = summarize_sequential_context(collected)
if rc == 124:
trace_stage('sequential_timeout', name, f'budget={budget}')
failed_ev = ev
break
if rc != 0:
trace_stage('sequential_tool_error', name, (err or '')[:200])
failed_ev = ev
break
trace_stage('multi_tool_collected', f'seq={is_seq}', str([(item.get("tool"), len(item.get("output", ""))) for item in collected]))
return {'collected': collected, 'failed_ev': failed_ev, 'is_sequential': is_seq}
try:
from composition_selector import apply_composition_selection as _apply_composition_selection
except Exception:
def _apply_composition_selection(analysis, composition, family_candidates, *, choose_plan, use_policy=True):
if not composition or not composition.get('composed'):
return analysis, None, None
before_comp = dict(analysis)
if use_policy:
policy = choose_plan(
before_comp.get('tools') or [],
composition.get('tools') or [],
family_candidates,
composition_reason=composition.get('reason', ''),
) or {}
else:
policy = {'selected_tools': list(composition.get('tools') or []), 'policy': 'no_policy'}
updated = dict(analysis)
updated.update(
role=composition.get('role', analysis.get('role')),
task_type=composition.get('task_type', analysis.get('task_type')),
tools=policy.get('selected_tools') or composition.get('tools') or analysis.get('tools'),
needs_setup_context=bool(composition.get('needs_setup_context', analysis.get('needs_setup_context'))),
needs_memory=bool(composition.get('needs_memory', analysis.get('needs_memory'))),
force_sequential=bool(composition.get('force_sequential')),
composition_reason=composition.get('reason', ''),
composition_policy=policy.get('policy', ''),
)
return updated, before_comp, {**composition, 'policy': policy.get('policy', '')}
try:
from intent_normalizer import normalize_intent_family as _normalize_intent_family
except Exception:
def _normalize_intent_family(message, analysis, family_info=None, *, classify_intent_family, service_hints):
tools = list(analysis.get('tools') or [])
primary = tools[0] if tools else ''
family = (family_info or classify_intent_family(message, service_hints)).get('family', '')
if family == 'url_research':
analysis.update(role='research', task_type='summarize', tools=['url_research'], needs_setup_context=False, needs_web_research=False, confidence=max(float(analysis.get('confidence', 0.0)), 0.96))
return analysis
if family == 'web_diagnose':
analysis.update(role='ops', task_type='diagnose', tools=['web_root_cause'], needs_setup_context=True, needs_web_research=False, confidence=max(float(analysis.get('confidence', 0.0)), 0.95))
return analysis
if family in {'memory_store', 'memory_lookup'}:
analysis.update(role='personal', task_type='memory', tools=['memory_profile'], needs_memory=True, needs_web_research=False, confidence=max(float(analysis.get('confidence', 0.0)), 0.93))
return analysis
if family == 'live_status':
analysis.update(role='ops', task_type='status', tools=['light_status'], needs_setup_context=True, needs_web_research=False, confidence=max(float(analysis.get('confidence', 0.0)), 0.95))
return analysis
if family == 'emby_provision':
analysis.update(role='ops', task_type='howto', tools=['emby_user_provision'], needs_setup_context=True, needs_web_research=False, confidence=max(float(analysis.get('confidence', 0.0)), 0.95))
return analysis
if family == 'user_access':
analysis.update(role='ops', task_type='diagnose', tools=['user_service_access_diagnose'], needs_setup_context=True, needs_web_research=False, confidence=max(float(analysis.get('confidence', 0.0)), 0.94))
return analysis
if family == 'stable_knowledge' and primary in {'web_research', 'url_research'}:
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_web_research=False, confidence=max(float(analysis.get('confidence', 0.0)), 0.9))
return analysis
if family == 'personal_help' and primary not in {'personal_assist', 'expert_write'}:
analysis.update(role='personal', task_type='assist', tools=['personal_assist'], needs_memory=bool((family_info or {}).get('needs_memory')), needs_web_research=False, confidence=max(float(analysis.get('confidence', 0.0)), 0.9))
return analysis
if family == 'personal_help' and primary == 'personal_assist' and not (family_info or {}).get('needs_memory'):
analysis['needs_memory'] = False
return analysis
try:
from meta_controller import shadow_decision as _shadow_decision, log_shadow_decision as _log_shadow_decision
USE_META_CONTROLLER = True
except Exception:
USE_META_CONTROLLER = False
def _shadow_decision(message, analysis, family_candidates, tool_registry):
return {}
def _log_shadow_decision(log_path, decision_row):
return None
try:
from outcome_logging import (
log_intent_family_shadow as _log_intent_family_shadow,
log_intent_composition as _log_intent_composition,
record_task_outcome as _record_task_outcome,
)
except Exception:
def _log_intent_family_shadow(message, family_info, before_tools, after_tools, *, log_path, collect_intent_families, service_hints):
family = str((family_info or {}).get('family') or '')
if not family:
return
try:
log_path.parent.mkdir(parents=True, exist_ok=True)
try:
candidates = [item.get('family') for item in collect_intent_families(message, service_hints) if item.get('family')]
except Exception:
candidates = [family]
row = {
'ts': datetime.now(timezone.utc).isoformat(),
'message': message,
'family': family,
'family_candidates': candidates,
'before_tool': before_tools[0] if before_tools else '',
'after_tool': after_tools[0] if after_tools else '',
'overridden': (before_tools[:1] != after_tools[:1]),
}
with log_path.open('a', encoding='utf-8') as f:
f.write(json.dumps(row, ensure_ascii=False) + '\n')
except Exception:
pass
def _log_intent_composition(message, family_candidates, analysis_before, analysis_after, composition, *, log_path):
if not composition or not composition.get('composed'):
return
try:
log_path.parent.mkdir(parents=True, exist_ok=True)
row = {
'ts': datetime.now(timezone.utc).isoformat(),
'message': message,
'family_candidates': [item.get('family') for item in family_candidates if item.get('family')],
'before_tools': list(analysis_before.get('tools') or []),
'after_tools': list(analysis_after.get('tools') or []),
'reason': composition.get('reason', ''),
'policy': composition.get('policy', analysis_after.get('composition_policy', '')),
'force_sequential': bool(analysis_after.get('force_sequential')),
}
with log_path.open('a', encoding='utf-8') as f:
f.write(json.dumps(row, ensure_ascii=False) + '\n')
except Exception:
pass
def _record_task_outcome(message, analysis, final_text, evidence_items, *, status='success', error_text='', log_path, classify_intent_family, collect_intent_families, service_hints, refresh_composition_policy_async):
try:
log_path.parent.mkdir(parents=True, exist_ok=True)
grounded_items = [item for item in evidence_items if item.get('grounded')]
try:
candidates = [item.get('family') for item in collect_intent_families(message, service_hints) if item.get('family')]
except Exception:
candidates = []
row = {
'ts': datetime.now(timezone.utc).isoformat(),
'status': status,
'message': message,
'role': analysis.get('role'),
'task_type': analysis.get('task_type'),
'planned_tools': list(analysis.get('tools') or []),
'used_tools': [item.get('tool') for item in evidence_items],
'family': (classify_intent_family(message, service_hints) or {}).get('family', ''),
'family_candidates': candidates,
'grounded_count': len(grounded_items),
'evidence_count': len(evidence_items),
'answer_len': len((final_text or '').strip()),
'needs_memory': bool(analysis.get('needs_memory')),
'needs_setup_context': bool(analysis.get('needs_setup_context')),
'error_text': (error_text or '')[:300],
'composition_reason': str(analysis.get('composition_reason') or ''),
'composition_policy': str(analysis.get('composition_policy') or ''),
}
with log_path.open('a', encoding='utf-8') as f:
f.write(json.dumps(row, ensure_ascii=False) + '\n')
if row.get('composition_reason') or row.get('composition_policy'):
refresh_composition_policy_async()
except Exception:
pass
try:
from response_policy import (
postprocess_final_answer as _policy_postprocess_final_answer,
should_skip_heavy_persistence,
can_fast_exit_single_tool,
should_skip_quality_feedback,
build_lightweight_composed_answer,
sequential_tool_budget,
summarize_sequential_context,
)
except Exception:
def _policy_postprocess_final_answer(_message, _analysis, final_text):
return (final_text or '').strip()
def should_skip_heavy_persistence(_message, _analysis, _evidence_items):
return False
def can_fast_exit_single_tool(_analysis, _evidence_items, memory_ctx='', retrieval_ctx=''):
return False
def should_skip_quality_feedback(_analysis, _evidence_items):
return False
def build_lightweight_composed_answer(_message, _analysis, _evidence_items):
return ''
def sequential_tool_budget(_name, is_sequential, default=12):
return default if is_sequential else 0
def summarize_sequential_context(_evidence_items, max_chars=900):
return ''
import argparse
import json
import os
import re
import subprocess
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path
ROOT = Path('/home/openclaw/.openclaw/workspace')
LLM_MAC = ROOT / 'bin' / 'llm-mac'
LLM_LOCAL = ROOT / 'bin' / 'llm-answer-local'
AGENT_MEMORY = Path(os.environ.get('BIBABOT_AGENT_MEMORY_BIN', str(ROOT / 'bin' / 'agent-memory')))
CONTEXT_PACK = ROOT / 'bin' / 'context-pack'
LOG_PATH = ROOT / 'logs' / 'orchestrator-events.jsonl'
INTENT_FAMILY_LOG = ROOT / 'logs' / 'intent-family-shadow.jsonl'
TASK_OUTCOME_LOG = ROOT / 'logs' / 'task-outcomes.jsonl'
INTENT_COMPOSITION_LOG = ROOT / 'logs' / 'intent-composition.jsonl'
META_CONTROLLER_LOG = ROOT / 'logs' / 'meta-controller-shadow.jsonl'
COMPOSITION_POLICY_REFRESH = ROOT / 'bin' / 'composition-policy-refresh'
TOPIC_STATE = ROOT / 'memory' / 'topic-state.json'
TOPIC_HISTORY = ROOT / 'memory' / 'topic-history.jsonl'
PROFILE_STORE = ROOT / 'memory' / 'profile.jsonl'
def trace_stage(*parts):
if os.environ.get('BIBABOT_TRACE', '0') != '1':
return
try:
print('[trace]', *parts, file=sys.stderr, flush=True)
except Exception:
pass
TOOLS = {
'memory_profile': {'description': 'Persoenliches Gedaechtnis lesen oder speichern. Nutze das fuer Fragen ueber Vorlieben, Stil, Identitaet oder wenn der Nutzer explizit etwas merken will.', 'kind': 'final'},
'briefing': {'description': 'Liefert ein aktuelles Lagebild zu Haus, Homelab, Erinnerungen und Agent-Gesundheit.', 'kind': 'final'},
'setup_lookup': {'description': 'Liest das echte Homelab-Inventar: welche Dienste es gibt, wo sie laufen, wie man an Logs kommt und welche Hosts relevant sind.', 'kind': 'evidence'},
'public_services_health': {'description': 'Prueft oeffentliche Services, Hosts und aktuelle Warnings/Fehler.', 'kind': 'final'},
'log_hotspots': {'description': 'Analysiert die haeufigsten Fehler in immich/nextcloud/traefik Logs.', 'kind': 'final'},
'host_slow_diagnose': {'description': 'Diagnose fuer langsame Hosts oder Services anhand CPU, RAM, Disk, Netzwerk und Logs.', 'kind': 'final'},
'kasm_diagnose': {'description': 'Diagnose fuer Kasm-Zugriff, Kasm-Deployment und Erreichbarkeit.', 'kind': 'final'},
'user_service_access_diagnose': {'description': 'Diagnostiziert warum eine bestimmte Person oder ein Benutzer nicht auf einen bestimmten Dienst zugreifen kann. Erst Dienstgesundheit, dann Auth-/Rechtepfad pruefen.', 'kind': 'final'},
'web_root_cause': {'description': 'Diagnose warum eine gehostete Website nicht erreichbar ist.', 'kind': 'final'},
'sonos_library_diagnose': {'description': 'Diagnose fuer Sonos lokale Musikbibliothek auf Unraid/NAS/SMB. Erst Sonos-Mechanismus verstehen, dann Share/Pfad/Zugriff auf Unraid pruefen.', 'kind': 'final'},
'emby_user_provision': {'description': 'Erklaert und verifiziert, wie Emby-Benutzer in deinem Setup angelegt werden sollen, insbesondere ueber LDAP/Authentik statt lokale Emby-User.', 'kind': 'final'},
'light_status': {'description': 'Zeigt welche Lichter aktuell eingeschaltet sind.', 'kind': 'final'},
'light_usage': {'description': 'Analysiert Lichtnutzung und Automatisierungspotenzial.', 'kind': 'final'},
'light_control': {'description': 'Schaltet Lichter ein oder aus. Das ist eine echte Aktion mit Seiteneffekt.', 'kind': 'action'},
'web_research': {'description': 'Recherchiert im Web mit Quellen. Nutze das fuer Fakten, unbekannte Begriffe, Named Entities oder aktuelle Informationen.', 'kind': 'final'},
'url_research': {'description': 'Analysiert eine konkrete URL oder Webseite.', 'kind': 'final'},
'compare_selfhosted': {'description': 'Vergleicht Self-Hosted-Loesungen mit Blick auf die bestehende Infrastruktur.', 'kind': 'final'},
'security_plan': {'description': 'Erstellt einen priorisierten Sicherheitsplan fuer Homelab/Reverse Proxy/SSH/Secrets/Backups.', 'kind': 'final'},
'media_advice': {'description': 'Hilft bei Medien-Workflows wie Lidarr, Emby, Audiobooks, Audiobookshelf.', 'kind': 'final'},
'personal_assist': {'description': 'Persoenliche Hilfe wie Mails, Nachrichten, Erinnerungen, Einkaufslisten, To-dos.', 'kind': 'final'},
'dev_build': {'description': 'Programmier- und Build-Aufgaben: Tools, Skripte, Apps, Implementierung.', 'kind': 'final'},
'general_answer': {'description': 'Allgemeine hilfreiche Antwort fuer Erklaerungen, How-tos, Rezepte, Listen und alltaegliche Fragen.', 'kind': 'final'},
'agent_introspect': {'description': 'Selbstbeobachtung: Pruefe eigene Konfidenz und markiere Unsicherheiten vor der Ausgabe', 'kind': 'evidence'},
'agent_self_correct': {'description': 'Lerne aus Fehlern und vermeide bekannte Fehlermuster', 'kind': 'evidence'},
'agent_reason': {'description': 'Komplexes Multi-Step Reasoning mit Checkpoints und Backtracking', 'kind': 'evidence'},
'agent_anticipate': {'description': 'Vorhersage von User-Intentions fuer proaktive Unterstuetzung', 'kind': 'evidence'},
'agent_emotion': {'description': 'Emotionale Intelligenz fuer adaptive Kommunikation', 'kind': 'evidence'},
'agent_simulate': {'description': 'Mentale Simulation fuer Konsequenzen-Abschaetzung vor Aktionen', 'kind': 'evidence'},
'agent_synthesize': {'description': 'Cross-Domain Synthese fuer kreative Problemloesung', 'kind': 'evidence'},
'ops_exec': {'description': 'Fuehrt Befehle auf Homelab-Hosts aus und analysiert Ausgabe', 'kind': 'action', 'script': 'ops-exec-action'},
'ops_install': {'description': 'Installiert und konfiguriert Software, testet und verifiziert', 'kind': 'action', 'script': 'ops-install-action'},
'ops_api': {'description': 'Fragt Homelab REST-APIs ab: HA, Docker, Emby, Nextcloud, Authentik', 'kind': 'final', 'script': 'ops-api-query'},
'ops_deep_analyze': {'description': 'Tiefe Analyse: Logs + Befehle + Web-Recherche + Synthese', 'kind': 'final', 'script': 'ops-deep-analyze'},
'qdrant_search': {'description': 'Semantische Suche in der lokalen Wissensdatenbank (KB-Dateien, Homelab-Doku, Setup-Infos).', 'kind': 'evidence'},
'expert_write': {'description': 'Experte fuer kreatives Schreiben, Journalismus, Artikel, Emails, Storytelling, technische Doku.', 'kind': 'final'},
'expert_code': {'description': 'Senior Software Architect: Code-Architektur, Review, Debugging, Performance, Security.', 'kind': 'final'},
'expert_research': {'description': 'Tiefer Research-Analyst: Multi-Quellen-Synthese, Fact-Checking, investigative Analyse.', 'kind': 'final'},
'expert_edit': {'description': 'Professionelles Lektorat: Grammatik, Stil, Struktur, Klarheit — Text sofort verwendbar.', 'kind': 'final'},
'expert_strategy': {'description': 'Strategy Consultant: Business-Planung, OKRs, Roadmaps, Priorisierung, Management-Frameworks.', 'kind': 'final'},
'goals_manage': {'description': 'Verwaltet Ziele, Projekte und Wochen-Prioritaeten. Hinzufuegen, anzeigen, aktualisieren.', 'kind': 'final'},
'contacts_manage': {'description': 'Verwaltet Kontakte, Follow-ups und Geburtstage. Hinzufuegen, anzeigen, suchen.', 'kind': 'final'},
}
def env_models(name: str, default: list[str]) -> list[str]:
raw = os.environ.get(name, '').strip()
if not raw:
return list(default)
return [item.strip() for item in raw.split(',') if item.strip()] or list(default)
ANALYSIS_MODELS = env_models('BIBABOT_ANALYSIS_MODELS', ['qwen2.5:3b', 'qwen2.5:1.5b', 'llama3.1:8b'])
VERIFY_MODELS = env_models('BIBABOT_VERIFY_MODELS', ['qwen2.5:3b', 'gemma3:27b-it-qat', 'gemma3:12b'])
FOLLOWUP_MODELS = env_models('BIBABOT_FOLLOWUP_MODELS', ['gemma3:12b', 'qwen2.5:3b', 'llama3.1:8b'])
GENERAL_CHAIN_DEFAULT = 'qwen2.5:3b,qwen2.5:1.5b,llama3.1:8b,mistral-small3.1:latest'
FINAL_ROLE_MAP = {'ops': 'ops', 'research': 'research', 'personal': 'personal', 'dev': 'dev', 'general': 'general'}
ROLE_HINTS = {'ops', 'research', 'personal', 'dev', 'general'}
SERVICE_HINTS = {'kasm', 'emby', 'sonos', 'unraid', 'traefik', 'nextcloud', 'immich', 'lidarr', 'radarr', 'sonarr', 'authentik', 'ldap', 'home assistant', 'homeassistant', 'tailscale', 'emby', 'wiki', 'share', 'smb'}
SERVICE_HINTS_PATTERN = re.compile(r'\b(' + '|'.join(re.escape(h) for h in SERVICE_HINTS if ' ' not in h) + r')\b')
FOLLOWUP_STOP = {'welche', 'welcher', 'welches', 'wie', 'was', 'warum', 'wieso', 'und', 'dafuer', 'dafür', 'dazu', 'damit', 'relevant', 'genau', 'nochmal', 'bitte'}
GROUNDED_FASTPATH_TOOLS = {'briefing', 'setup_lookup', 'public_services_health', 'log_hotspots', 'host_slow_diagnose', 'kasm_diagnose', 'user_service_access_diagnose', 'web_root_cause', 'sonos_library_diagnose', 'emby_user_provision', 'light_status', 'light_usage', 'compare_selfhosted', 'security_plan', 'url_research', 'personal_assist', 'goals_manage', 'contacts_manage'}
INCIDENT_MEMORY_TOOLS = {'public_services_health', 'log_hotspots', 'host_slow_diagnose', 'kasm_diagnose', 'user_service_access_diagnose', 'web_root_cause', 'sonos_library_diagnose', 'emby_user_provision'}
EVIDENCE_KIND_BY_TOOL = {
'memory_profile': 'memory',
'briefing': 'status',
'setup_lookup': 'setup',
'public_services_health': 'diagnostic',
'log_hotspots': 'diagnostic',
'host_slow_diagnose': 'diagnostic',
'kasm_diagnose': 'diagnostic',
'user_service_access_diagnose': 'diagnostic',
'web_root_cause': 'diagnostic',
'sonos_library_diagnose': 'diagnostic',
'emby_user_provision': 'diagnostic',
'light_status': 'status',
'light_usage': 'diagnostic',
'light_control': 'action',
'web_research': 'web',
'url_research': 'web',
'compare_selfhosted': 'web',
'security_plan': 'web',
'media_advice': 'diagnostic',
'personal_assist': 'general',
'dev_build': 'general',
'general_answer': 'general',
'agent_introspect': 'introspection',
'agent_self_correct': 'learning',
'agent_reason': 'reasoning',
'agent_anticipate': 'prediction',
'agent_emotion': 'emotion',
'agent_simulate': 'simulation',
'agent_synthesize': 'synthesis',
'expert_write': 'general',
'expert_code': 'general',
'expert_research': 'web',
'expert_edit': 'general',
'expert_strategy': 'general',
}
def norm(text: str) -> str:
text = (text or '').lower()
text = text.replace('ä', 'ae').replace('ö', 'oe').replace('ü', 'ue').replace('ß', 'ss')
return re.sub(r'\s+', ' ', text).strip()
def normalize_inbound_message(text: str) -> str:
clean = (text or '').replace('\r', '').strip()
clean = re.sub(r'^\[[^\]\n]{8,80}\]\s*', '', clean)
clean = re.sub(r'^\s*Current time:\s.*$', '', clean, flags=re.I | re.M)
return clean.strip()
def is_complex_task(message: str, analysis: dict) -> bool:
"""True when a task needs sequential steps where each step uses the previous result."""
low = norm(message)
if analysis.get('role') not in {'ops', 'dev'}:
return False
if analysis.get('task_type') not in {'action', 'diagnose', 'build'}:
return False
# Explicit multi-step language: diagnose → then fix
if re.search(r'(analysier|prüf|check|diagnose).{1,40}(und|dann|anschliessend).{1,40}(fix|reparier|beheb|restart|install|konfigur|deploy)', low):
return True
# Fix → then verify
if re.search(r'(fix|install|deploy|reparier).{1,40}(und|dann).{1,40}(test|verif|prüf|zeig|sicher)', low):
return True
# Explicit deep/thorough analysis
if re.search(r'(schritt.fuer.schritt|vollstaendig.*analysier|tiefgreifend|root.cause|ursache.*finden|warum.*kaputt|warum.*fehler|alles.*beheben|komplett.*analysier)', low):
return True
# Multiple action verbs = implied multi-step
verbs = sum(1 for v in ['analysier', 'diagnose', 'fixe', 'reparier', 'installier', 'restart', 'konfigur', 'deploy', 'verif', 'beheb'] if v in low)
return verbs >= 2
def run(cmd, timeout=120, env=None):
try:
if isinstance(cmd, (list, tuple)):
cmd = [c.replace('\x00', '') if isinstance(c, str) else c for c in cmd]
budget = (os.environ.get('BIBABOT_TOOL_TIMEOUT') or '').strip()
if budget:
try:
timeout = min(int(timeout), max(1, int(budget)))
except Exception:
pass
run_env = None
if env is not None:
run_env = os.environ.copy()
run_env.update(env)
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, env=run_env)
return proc.returncode, (proc.stdout or '').strip(), (proc.stderr or '').strip()
except subprocess.TimeoutExpired:
return 124, '', 'timeout'
def call_llm(model: str, system: str, user: str, timeout: int = 180):
return run([str(LLM_MAC), 'chat', model, system, user], timeout=timeout)
def call_llm_schema(model: str, schema: dict, system: str, user: str, timeout: int = 180):
with tempfile.NamedTemporaryFile('w', suffix='.schema.json', delete=False, encoding='utf-8') as tmp:
json.dump(schema, tmp, ensure_ascii=False)
schema_path = tmp.name
try:
return run([str(LLM_MAC), 'chat-schema', model, schema_path, system, user], timeout=timeout)
finally:
try:
os.unlink(schema_path)
except OSError:
pass
def model_exists(model: str) -> bool:
rc, _out, _err = run([str(LLM_MAC), 'has', model], timeout=20)
return rc == 0
ANALYSIS_SCHEMA = {
"type": "object",
"properties": {
"role": {"type": "string", "enum": ["ops", "research", "personal", "dev", "general"]},
"task_type": {"type": "string", "enum": ["diagnose", "explain", "compare", "howto", "write", "action", "status", "build", "memory"]},
"needs_setup_context": {"type": "boolean"},
"needs_web_research": {"type": "boolean"},
"needs_memory": {"type": "boolean"},
"dangerous_action": {"type": "boolean"},
"tools": {"type": "array", "items": {"type": "string"}},
"reasoning_focus": {"type": "array", "items": {"type": "string"}},
"confidence": {"type": "number"}
},
"required": ["role", "task_type", "needs_setup_context", "needs_web_research", "needs_memory", "dangerous_action", "tools", "reasoning_focus", "confidence"],
"additionalProperties": False
}
FOLLOWUP_SCHEMA = {
"type": "object",
"properties": {
"is_followup": {"type": "boolean"},
"root_message": {"type": "string"},
"reply_mode": {"type": "string", "enum": ["direct", "compare", "steps", "single_choice", "diagnose"]},
"focus": {"type": "string"}
},
"required": ["is_followup", "root_message", "reply_mode", "focus"],
"additionalProperties": False
}
VERIFY_SCHEMA = {
"type": "object",
"properties": {
"verdict": {"type": "string", "enum": ["keep", "revise", "retry"]},
"grounded": {"type": "boolean"},
"addresses_request": {"type": "boolean"},
"wrong_entity_jump": {"type": "boolean"},
"retry_tool": {"type": "string"},
"missing_points": {"type": "array", "items": {"type": "string"}},
"rationale": {"type": "string"},
"revised_answer": {"type": "string"},
},
"required": ["verdict", "grounded", "addresses_request", "wrong_entity_jump", "retry_tool", "missing_points", "rationale", "revised_answer"],
"additionalProperties": False
}
def extract_json(text: str):
if not text:
return None
start = text.find('{')
while start != -1:
depth = 0
in_str = False
esc = False
for idx in range(start, len(text)):
ch = text[idx]
if in_str:
if esc:
esc = False
elif ch == '\\':
esc = True
elif ch == '"':
in_str = False
continue
if ch == '"':
in_str = True
elif ch == '{':
depth += 1
elif ch == '}':
depth -= 1
if depth == 0:
snippet = text[start:idx + 1]
try:
return json.loads(snippet)
except Exception:
break
start = text.find('{', start + 1)
return None
def tool_catalog_text() -> str:
return '\n'.join(f'- {name}: {meta["description"]}' for name, meta in TOOLS.items())
def topic_terms(text: str):
terms = []
for tok in re.findall(r'[a-z0-9._-]{3,}', norm(text)):
if tok in FOLLOWUP_STOP:
continue
terms.append(tok)
return list(dict.fromkeys(terms))[:12]
def load_topic_state():
if not TOPIC_STATE.exists():
return None
try:
data = json.loads(TOPIC_STATE.read_text(encoding='utf-8'))
except Exception:
return None
return data if isinstance(data, dict) else None
def load_topic_history(limit: int = 12):
if not TOPIC_HISTORY.exists():
return []
items = []
try:
lines = TOPIC_HISTORY.read_text(encoding='utf-8', errors='ignore').splitlines()
except Exception:
return []
for line in lines[-limit:]:
line = line.strip()
if not line:
continue
try:
data = json.loads(line)
except Exception:
continue
if isinstance(data, dict):
items.append(data)
return items
def append_topic_history(payload: dict):
try:
TOPIC_HISTORY.parent.mkdir(parents=True, exist_ok=True)
with TOPIC_HISTORY.open('a', encoding='utf-8') as f:
f.write(json.dumps(payload, ensure_ascii=False) + '\n')
except Exception:
pass
def topic_state_match_score(message: str, state: dict | None):
if not state:
return 0.0
score = 0.0
qterms = topic_terms(message)
combined = ' '.join([str(state.get('message') or ''), ' '.join(state.get('keywords') or []), str(state.get('summary') or '')])
hay = norm(combined)
for term in qterms:
if term in hay:
score += 2.0
if is_followup_query(message):
score += 1.4
if state.get('analysis', {}).get('role') in {'ops', 'dev'} and any(s in hay for s in SERVICE_HINTS):
score += 0.4
return score
def recent_topic_event(message: str):
candidates = []
state = load_topic_state()
if state:
candidates.append(state)
candidates.extend(load_topic_history())
best = None
best_score = 0.0
seen = set()
for item in reversed(candidates):
key = (str(item.get('ts') or ''), str(item.get('message') or ''), str(item.get('summary') or ''))
if key in seen:
continue
seen.add(key)
score = topic_state_match_score(message, item)
if score > best_score:
best = item
best_score = score
if best_score >= 1.4:
return best
return None
def resolve_followup(message: str, state: dict | None):
if not state or not is_followup_query(message):
return None
system = (
'Du loest Anschlussfragen fuer J.A.R.V.I.S. auf. Nutze nur den gegebenen Themenkontext. '
'Gib strikt JSON gemaess Schema aus.'
)
summary = str(state.get('summary') or '').strip()
root = str(state.get('root_message') or state.get('message') or '').strip()
user = (
f'Aktuelles Thema:\n{root}\n\n'
f'Bisheriger Kerninhalt:\n{summary}\n\n'
f'Neue Nutzerfrage:\n{message}\n\n'
'Bestimme, ob die neue Frage eine Anschlussfrage zum Thema ist, welches Ursprungsproblem gemeint ist und wie knapp geantwortet werden soll.'
)
for model in FOLLOWUP_MODELS:
if not model_exists(model):
continue
rc, out, _err = call_llm_schema(model, FOLLOWUP_SCHEMA, system, user, timeout=45)
if rc != 0 or not out:
continue
try:
data = json.loads(out)
except Exception:
continue
if isinstance(data, dict):
return data
return None
def update_topic_state(message: str, analysis: dict, final_text: str):
if not isinstance(analysis, dict):
return
tools = [t for t in analysis.get('tools', []) if t in TOOLS and TOOLS[t]['kind'] != 'action']
if not tools:
return
previous = load_topic_state() or {}
summary_lines = []
summary_cap = 4
if analysis.get('role') == 'general' and analysis.get('task_type') in {'howto', 'compare', 'explain'}:
summary_cap = 14
elif analysis.get('task_type') in {'compare', 'build'}:
summary_cap = 10
for line in (final_text or '').splitlines():
line = line.strip()
if not line:
continue
summary_lines.append(line)
if len(summary_lines) >= summary_cap:
break
root_message = message
prev_match = topic_state_match_score(message, previous) if previous else 0.0
if prev_match >= 1.4 and previous.get('root_message'):
root_message = str(previous.get('root_message')).strip() or message
payload = {
'ts': datetime.now(timezone.utc).isoformat(),
'root_message': root_message,
'message': message,
'keywords': topic_terms((root_message or message) + ' ' + message + ' ' + ' '.join(tools)),
'summary': '\n'.join(summary_lines),
'analysis': {
'role': analysis.get('role'),
'task_type': analysis.get('task_type'),
'needs_setup_context': bool(analysis.get('needs_setup_context')),
'needs_web_research': bool(analysis.get('needs_web_research')),
'needs_memory': bool(analysis.get('needs_memory')),
'tools': tools[:3],
},
}
try:
TOPIC_STATE.parent.mkdir(parents=True, exist_ok=True)
TOPIC_STATE.write_text(json.dumps(payload, ensure_ascii=False), encoding='utf-8')
append_topic_history(payload)
except Exception:
pass
def current_topic_context(message: str):
state = recent_topic_event(message)
if not state:
return ''
tools = ', '.join(state.get('analysis', {}).get('tools', [])[:3])
summary = str(state.get('summary') or '').strip()
out = [f"Aktives Thema: {str(state.get('root_message') or state.get('message') or '').strip()} | tools={tools}"]
if summary:
out.append(summary)
return '\n'.join(out)
def is_followup_query(message: str) -> bool:
low = norm(message)
if re.search(r'https?://', message):
return False
if any(h in low for h in SERVICE_HINTS):
return False
pronouns = ['dafuer', 'dafuer', 'dazu', 'damit', 'darauf', 'darin', 'darueber', 'davon']
if any(p in low for p in pronouns):
return True
# 'diese/dieses/dieser' as standalone pronouns → followup, but NOT as articles before nouns/temporal words
TEMPORAL_NOUNS = r'(monat|jahr|tag|abend|morgen|sommer|winter|herbst|fruehjahr|fruhling|woche|quartal|semester|moment|mal|abschnitt)'
if re.search(r'\b(dies|dieses|diese[rn]?)\b', low) and not re.search(r'\bdiesen?\s+' + TEMPORAL_NOUNS, low) and not re.search(r'\bdieses\s+' + TEMPORAL_NOUNS, low) and not re.search(r'\bdieser?\s+' + TEMPORAL_NOUNS, low):
return True
if low.startswith('und '):
return True
if re.search(r'^welche[rns]? davon\b', low):
return True
return False
def recent_dialogue_event(message: str):
if not LOG_PATH.exists():
return None
try:
lines = LOG_PATH.read_text(encoding='utf-8', errors='ignore').splitlines()[-12:]
except Exception:
return None
query_terms = topic_terms(message)
ranked = []
for idx, line in enumerate(reversed(lines), start=1):
line = line.strip()
if not line:
continue
try:
data = json.loads(line)
except Exception:
continue
if not isinstance(data, dict) or not data.get('analysis', {}).get('tools'):
continue
if norm(str(data.get('message') or '')) == norm(message):
continue
tools = [t for t in data.get('analysis', {}).get('tools', []) if t in TOOLS and TOOLS[t]['kind'] != 'action']
if not tools:
continue
data['analysis']['tools'] = tools
hay = norm(str(data.get('message') or '') + ' ' + ' '.join(TOOLS[t]['description'] for t in tools))
score = 0.0
for term in query_terms:
if term in hay:
score += 1.8
score += max(0.0, 0.35 - idx * 0.03)
ranked.append((score, data))
if not ranked:
return None
ranked.sort(key=lambda x: x[0], reverse=True)
if ranked[0][0] <= 0:
return None
return ranked[0][1]
def recent_dialogue_context():
if not LOG_PATH.exists():
return ''
try:
lines = LOG_PATH.read_text(encoding='utf-8', errors='ignore').splitlines()[-4:]
except Exception:
return ''
parts = []
for line in lines:
try:
data = json.loads(line)
except Exception:
continue
msg = str(data.get('message') or '').strip()
tools = ', '.join(data.get('analysis', {}).get('tools', [])[:3])
if msg:
parts.append(f'- Letzte Anfrage: {msg} | tools={tools}')
return '\n'.join(parts[:3])
def analyze_with_llm(message: str, role_hint: str | None, memory_ctx: str = '', retrieval_ctx: str = ''):
system = (
'Du bist der Planungs-Kern von J.A.R.V.I.S. Verstehe zuerst den Mechanismus der Anfrage und waehle dann passende Tools. '
'Verlasse dich nicht auf Keyword-Aehnlichkeit. Waehle nur Tools, die wirklich zum Problem passen. '
'Beispiele: Sonos + Unraid Bibliothek -> sonos_library_diagnose. "Was ist Sitterdorf?" -> web_research. '
'Bei mehrstufigen Aufgaben: waehle 2 Tools in der richtigen Reihenfolge (erst Info, dann Analyse/Aktion). '
'Analyse+Fix: [ops_api, ops_deep_analyze]. Installation+Test: [ops_install, ops_exec]. '
'Einfache Statusfragen brauchen nur 1 Tool. '
'Gib nur JSON aus.'
)
hint_text = f'Bevorzugte Rolle: {role_hint}. Nutze sie als weichen Hinweis, aber nicht wenn sie fachlich offensichtlich falsch waere.\n' if role_hint else ''
context_parts = []
if memory_ctx:
context_parts.append('Relevantes Nutzer-/Korrektur-Gedaechtnis:\n' + memory_ctx)
if retrieval_ctx:
context_parts.append('Relevantes lokales Setup-/Wissens-Retrieval:\n' + retrieval_ctx)
topic_ctx = current_topic_context(message)
if topic_ctx:
context_parts.append('Aktiver Themenkontext:\n' + topic_ctx)
context_block = ('\n\n'.join(context_parts) + '\n\n') if context_parts else ''
user = (
f'Anfrage:\n{message}\n\n'
f'{hint_text}'
f'{context_block}'
'Verfuegbare Tools:\n'
f'{tool_catalog_text()}\n\n'
'JSON-Schema:\n'
'{"role":"ops|research|personal|dev|general",'
'"task_type":"diagnose|explain|compare|howto|write|action|status|build|memory",'
'"needs_setup_context":true,'
'"needs_web_research":false,'
'"needs_memory":false,'
'"dangerous_action":false,'
'"tools":["tool_name"],'
'"reasoning_focus":["kurzer Punkt"],'
'"confidence":0.0}\n\n'
'Regeln:\n'
'- Waehle maximal 3 Tools.\n'
'- Wenn die Anfrage zuerst Sachverstaendnis braucht, waehle das passende Sach-/Diagnosetool vor lokalen Checks.\n'
'- Fuer alltaegliche Erklaerungen/Rezepte/How-tos ohne aktuelle Fakten reicht general_answer.\n'
'- Fuer Named-Entity-, Orts- oder Faktenfragen mit Halluzinationsrisiko nimm web_research.\n'
'- Fuer Vergleiche von lokal/self-hosted/selbst gehosteten Loesungen nimm compare_selfhosted.\n'
'- Fuer Coding-/Build-Aufgaben nimm dev_build.\n'
'- Fuer Mail/Nachricht/Reminder nimm personal_assist.\n'
'- Fuer Fragen ueber Benutzerprofil/Vorlieben oder "merke dir" nimm memory_profile.\n'
'- Fuer Lichter nur bei echter Anweisung light_control, sonst light_status oder light_usage.\n'
'- Nimm setup_lookup nur wenn echtes Homelab-/Host-/Service-Wissen gebraucht wird.\n'
'- Wenn Anfrage sowohl Diagnose als auch Fix will: [ops_api, ops_deep_analyze].\n'
'- Wenn Anfrage Installation + Verifikation will: [ops_install, ops_exec].\n'
'- Bei "was soll ich tun", "wie kann ich", "fix das" nach einer Fehlerdiagnose: ops_deep_analyze benutzen.\n'
)
for model in ANALYSIS_MODELS:
if not model_exists(model):
continue
rc, out, _err = call_llm_schema(model, ANALYSIS_SCHEMA, system, user, timeout=90)
if rc == 0 and out:
try:
parsed = json.loads(out)
except Exception:
parsed = extract_json(out)
if isinstance(parsed, dict):
return parsed
return None
def heuristic_analysis(message: str, role_hint: str | None = None):
low = norm(message)
# === EXPERT-EDIT: Highest priority — lectorat/proofread patterns before anything else ===
# These are unambiguous writing-help patterns that must not be caught by service/ops rules
if re.search(r'(korrigier.{0,20}(text|folgenden|meinen|diesen|den)|'
r'lektorier|lektorat|proofreading|proofread|'
r'verbessere.{0,20}(text|meinen text|folgenden text|diesen text|schreibstil)|'
r'ueberarbeite.{0,20}(text|meinen|diesen|folgenden)|'
r'überarbeite.{0,20}(text|meinen|diesen|folgenden)|'
r'rechtschreibung.*pruefen|grammatik.*pruefen|pruef.*grammatik|pruef.*rechtschreibung|'
r'stilistisch.*verbessern|stilverbesserung|besser.*formulier|umformulier|'
r'klingt.*besser|klingt.*natuerlicher|klingt.*professioneller|'
r'feedback.*text|feedback.*meinem text|text.*feedback)', low) and not any(s in low for s in ('docker', 'code', 'skript')):
analysis = {
'role': 'general', 'task_type': 'write', 'needs_setup_context': False,
'needs_web_research': False, 'needs_memory': False, 'dangerous_action': False,
'tools': ['expert_edit'], 'reasoning_focus': [], 'confidence': 0.95,
}
return analysis
# === FIX: CREATIVE-WRITE EARLY INTERCEPT (höchste Priorität) ===
# Creative writing -> personal_assist (tweet, gedicht, witz, bio, instagram post, etc.)
# MUSS VOR goals_manage und expert_write kommen!
if re.search(r'(\btweet\b|erstelle.*tweet|tweet.*erstelle|schreib.*tweet|twitter.*post)', low):
return {'role': 'personal', 'task_type': 'write', 'tools': ['personal_assist'], 'needs_memory': True, 'needs_setup_context': False, 'needs_web_research': False, 'dangerous_action': False, 'confidence': 0.94}
if re.search(r'(\bgedicht\b|schreib.*gedicht|erstelle.*gedicht)', low):
return {'role': 'personal', 'task_type': 'write', 'tools': ['personal_assist'], 'needs_memory': True, 'needs_setup_context': False, 'needs_web_research': False, 'dangerous_action': False, 'confidence': 0.94}
if re.search(r'(\bwitz\b|schreib.*witz|erstelle.*witz|witz.*schreib)', low):
return {'role': 'personal', 'task_type': 'write', 'tools': ['personal_assist'], 'needs_memory': True, 'needs_setup_context': False, 'needs_web_research': False, 'dangerous_action': False, 'confidence': 0.94}
if re.search(r'(instagram.*post|facebook.*post|kurze.*bio|\bbio\b.*github)', low):
return {'role': 'personal', 'task_type': 'write', 'tools': ['personal_assist'], 'needs_memory': True, 'needs_setup_context': False, 'needs_web_research': False, 'dangerous_action': False, 'confidence': 0.94}
# GOALS-MANAGE: Ziel/Projekt-Verwaltung (explizite Verwaltungsbefehle — nicht Strategie-Anfragen)
# FIX: Ausschluss von "tweet" und creative writing bereits oben abgehandelt
if re.search(r'(fuege.*ziel|ziel.*hinzufuegen|neues ziel|setze.*ziel|wochen.*prio|prioritaet.*setze|constraint.*hinzufueg|ziel.*abgehakt|ziel.*erledigt|zeig.*ziele|was sind.*ziele|aktuelle.*ziele|zeig.*projekte)', low) or (re.search(r'(projekt.*erstell|neues.*projekt|projekt.*hinzufuegen)', low) and re.search(r'(jarvis|agent|merke|speicher|trag.*ein)', low)):
return {'role': 'personal', 'task_type': 'memory', 'tools': ['goals_manage'], 'needs_memory': False, 'needs_setup_context': False, 'needs_web_research': False, 'dangerous_action': False, 'confidence': 0.93}
# CONTACTS-MANAGE: Kontakt-Verwaltung
if re.search(r'(neuer kontakt|kontakt.*hinzufuegen|merke.*kontakt|wer hat.*geburtstag|geburtstage.*naechste|follow.?up.*faellig|follow.?up.*hinzufueg|letzter kontakt.*mit|zeig.*kontakte|habe.*kontaktiert|schuldige.*follow|offene.*follow)', low):
return {'role': 'personal', 'task_type': 'memory', 'tools': ['contacts_manage'], 'needs_memory': False, 'needs_setup_context': False, 'needs_web_research': False, 'dangerous_action': False, 'confidence': 0.93}
# === FIX: Pre-Check Fast-Path für Rezepte (höchste Priorität) ===
# Diese Patterns müssen VOR allen anderen Checks kommen, um ungrounded general_answer zu vermeiden
# Fix RECIPE-ROUTING-FINAL: Vereinfachte, robuste Patterns ohne \b am Ende
# ── PRO-WRITE EARLY INTERCEPT ─────────────────────────────────────────────
# Professionelle Kommunikation -> expert_write (vor personal_assist)
# FIX: "bewerbung" nur wenn formell/professionell (anschreiben, motivationsschreiben, etc.)
# "schreibe mir eine bewerbung" -> personal_assist (persönliche Hilfe, nicht formelle Dokumentation)
_pro_write_types = r'(anschreiben|motivationsschreiben|cover.?letter|kundigungsschreiben|kuendigungsschreiben|offizielle.*anfrage|formeller.*brief|formelle.*email|gehaltsverhandlung.*email|gehalts.*email|email.*(chef|vorgesetzt|manager|leitung|boss).*(gehal|bonus|kuendigung|versetz|urlaub)|email.*wegen.*(gehal|kuendigung|bonus|versetz)|kundigung.*verfass|kündigung.*verfass|offizielle.*stellungnahme|pressemitteilung|pitch.*deck|investor.*email|business.*proposal|vertrags.*entwurf|meeting.*protokoll|protokoll.*erstell)'
_pro_trigger = r'(kannst du|hilf mir|erstell|formulier|schreib|verfass|mach mir|ich brauche|bitte.*erstell|entwirf)'
if re.search(_pro_write_types, low) and re.search(_pro_trigger, low):
return {'role': 'research', 'task_type': 'write', 'tools': ['expert_write'], 'needs_memory': False, 'needs_setup_context': False, 'needs_web_research': False, 'dangerous_action': False, 'confidence': 0.91}
# FIX BEWERBUNG: "schreibe mir eine bewerbung" -> personal_assist (persönliche Hilfe)
# "bewerbung als X" oder "anschreiben fuer X" -> expert_write (formell)
if re.search(r'(schreib.*mir.*bewerbung|bewerbung.*schreib|hilf.*bewerbung|bewerbung.*hilf)', low) and not re.search(r'(anschreiben|motivationsschreiben|formell|professionell)', low):
return {'role': 'personal', 'task_type': 'write', 'tools': ['personal_assist'], 'needs_memory': True, 'needs_setup_context': False, 'needs_web_research': False, 'dangerous_action': False, 'confidence': 0.92}
# INTERNAL-TOOL-IDEAS: brainstorming for internal tools should stay general/dev, not recipe/personal
if re.search(r'(nenne|gib|zeig).{0,20}(ideen|vorschlaege|vorschläge|optionen)', low) and re.search(r'(internes? tool|internes? werkzeug|zugriffspruefung|zugriffsprüfung|access tool)', low):
return {'role': 'general', 'task_type': 'explain', 'tools': ['general_answer'], 'needs_memory': False, 'needs_setup_context': False, 'needs_web_research': False, 'dangerous_action': False, 'confidence': 0.94}
# WICHTIG: Patterns enden mit \s oder .* damit sie auch mit Leerzeichen/Folgetext matchen
RECIPE_FASTPATH = (
# Basis: "Gib mir X Rezept/Ideen/Vorschlaege..."
r'^(gib mir|gib|zeig mir|zeig|bringe mir|bringe|schick mir|schick)\s+(ein|eine|einen|\d+)\s+(rezept|ideen|vorschlaege|vorschläge)',
r'^(gib mir|gib|zeig mir|zeig)\s+(rezept|ideen|vorschlaege|vorschläge)',
# "Rezept fuer...", "Ideen fuer..."
r'(rezept|rezepte|ideen|vorschlaege|vorschläge)\s+(fuer|für)',
# "schnelles Abendessen", "was kochen"
r'(schnelles|einfaches|gutes)\s+(abendessen|mittagessen|fruehstueck|frühstück|essen|gericht)',
r'(abendessen|mittagessen|fruehstueck|frühstück)\s+(ideen|vorschlaege|vorschläge|rezepte)',
r'(was|was soll ich)\s+(kochen|essen|zum essen|zum abendessen|zum mittagessen)',
# Spezifische Gerichte
r'(omelett|omelette|omeletten|pfannkuchen|spaghetti|pasta|risotto|curry|suppe|salat|bowl|wraps)',
)
if any(re.search(p, low) for p in RECIPE_FASTPATH) and not any(h in low for h in ('homelab', 'server', 'docker', 'container', 'menu', 'code', 'script', 'programm')):
return {'role': 'personal', 'task_type': 'howto', 'tools': ['personal_assist'], 'needs_memory': True, 'needs_setup_context': False, 'needs_web_research': False, 'dangerous_action': False, 'confidence': 0.95}
# Pattern-Definitionen am Anfang der Funktion (Fix: RECIPE_PATTERNS muss vor Verwendung definiert sein)
RECIPE_PATTERNS = (
r'(gib mir.*rezept|rezept fuer|rezept f.r|wie mache ich|wie bereite ich)',
r'(ideen.*abendessen|schnelles abendessen|was koche ich|was esse ich|abendessen.*ideen|abendessen.*vorschlag)',
r'(omelett|omelette|omeletten|pasta|spaghetti|pfannkuchen|wraps|shakshuka|risotto|curry|suppe|salat|bowl)',
r'(zutaten|kochen|koch|backen|essen fuer|was koche|was esse|schnell.*essen|einfach.*kochen)',
r'(hunger|hungrig|mahlzeit|gericht|gerichte|speise|speisen)',
# Fix RECIPE-ROUTING-11: "Gib mir X fuer Y" - allgemeines Pattern für Rezept-Anfragen
r'^(gib mir|gib)\s+\w+\s+(fuer|für)\s+\w+',
# Fix RECIPE-ROUTING-12: "Gib mir X Ideen fuer Y" - erlaube beliebigen Text nach "fuer"
r'^(gib mir|gib)\s+\d+\s+ideen\s+(fuer|für)\b',
)
# (WRITE-CREATIVE Patterns jetzt oben unter CREATIVE-WRITE EARLY INTERCEPT)
SCIENCE_KB_PATTERNS = (
r'(himmel.*blau|blauer himmel|warum.*himmel)',
r'(sonne.*rot|sonne.*orange|sonnenuntergang.*rot|sonnenuntergang.*orange|untergehende sonne)',
r'(mondfinsternis.*rot|roter mond|blutmond|mond.*rot)',
r'(rayleigh.*streuung|lichtstreuung|streuung.*licht)',
r'(mond.*grösser|mond.*größer|mondtäuschung|mondtaeuschung)',
)
GENERAL_KB_PATTERNS = (
r'(was ist docker|was ist ein docker|was.*erklaer.*docker|docker vs.*vergleich|unterschied.*docker)',
r'(was ist ein reverse proxy|was ist reverse proxy|reverse proxy.*was)',
r'(was ist tailscale|tailscale.*was|vpn.*tailscale)',
r'(unraid.*truenass|truenass.*unraid|unraid vs|truenass vs)',
r'(unterschied.*wetter.*klima|wetter.*klima.*unterschied|was ist.*wetter|was ist.*klima)',
r'(photosynthese|photo.*synthese|pflanzen.*licht|licht.*pflanzen)',
r'(lichtjahr|licht jahr|was ist.*lichtjahr)',
r'(fleisch.*auftauen|auftauen.*fleisch|fleisch.*tiefkühl|fleisch.*gefrier)',
r'(waschmittel.*pulver|waschmittel.*flüssig|waschmittel.*pods|pulver.*flüssig)',
r'(kühlschrank.*temperatur|kühlschrank.*grad|gefrierfach.*temperatur)',
r'(schnittwunde.*erste hilfe|erste hilfe.*schnitt|schnitt.*wunde.*was tun)',
r'(wasser.*pro tag|wieviel wasser|wie viel wasser.*trinken)',
r'(schlaf.*dauer|schlaf.*stunden|wie lange.*schlafen|schlafdauer)',
r'(3-säulen.*system|drei säulen|ahv.*pensionskasse|säule 3a|säule 3b)',
r'(was ist ein etf|etf.*was|etf.*fonds|exchange traded)',
r'(du.*sie.*deutsch|sie.*du.*deutsch|wann du.*wann sie)',
# Neue Patterns für erweiterte KB_GENERAL (Fix K1-K12)
r'(berechne.*prozent|trinkgeld|prozent.*berechnen|prozentrechnung)',
r'(tage bis|wie viele tage|tage.*weihnachten|tage.*geburtstag)',
r'(stunden.*jahr|zeitrechnung|minuten.*jahr|stunden pro)',
r'(vitamin d|vitamin-d|mangel.*vitamin)',
r'(kopfschmerzen|kopf.*tut weh|migräne|migrane|schmerzen.*kopf)',
r'(gestresst|stress.*was tun|entspannung|entspannen)',
r'(schweiz.*gegründet|bundesstaat.*schweiz|gründung.*schweiz|schweiz.*geschichte)',
r'(http.*status|status code|503|502|404|500.*error|http.*fehler)',
r'(passwort.*ändern|passwoerter.*aendern|warum.*passwort)',
r'(python.*error|typeerror|python.*fehler|was bedeutet.*error)',
r'(übersetze|uebersetze|auf englisch|auf italienisch|auf deutsch|wie sagt man)',
r'(buch.*empfehl|buch.*schlafen|gutes buch|lesen.*schlafen)',
# Neue Patterns für erweiterte KB_GENERAL (Fix K13-K24)
r'(was ist.*ip.*adresse|ip.*adresse.*was|ip-adresse|meine ip)',
r'(was ist.*dns|dns.*was|domain name system)',
r'(was ist.*vpn|vpn.*was|virtual private)',
r'(router.*switch|unterschied.*router|access point.*unterschied)',
r'(makronaehrstoff|makro.*nährstoff|kohlenhydrate.*protein|protein.*fett)',
r'(glykaemisch|glykämisch|gi.*wert|glykaemischer index)',
r'(keto.*diaet|keto.*diät|ketose|ketogene)',
r'(was ist.*inflation|inflation.*was|verbraucherpreisindex)',
r'(zinseszins|zins.*zins|zinsrechnung)',
r'(aktien.*etf|etf.*aktien|einzelaktie.*etf)',
r'(waschmaschine.*funktioniert|wie funktioniert.*waschmaschine)',
r'(kuehlschrank.*funktioniert|kühlschrank.*funktioniert|wie funktioniert.*kühlschrank)',
r'(backpulver.*natron|natron.*hefe|backpulver.*hefe|unterschied.*backpulver)',
r'(jahreszeiten.*warum|warum.*jahreszeiten|warum gibt es.*jahreszeiten)',
r'(biologisch.*oekologisch|biologisch.*ökologisch|unterschied.*bio)',
r'(was ist.*oekosystem|oekosystem.*was|was ist.*ökossystem|ökossystem.*was)',
# Neue Patterns für erweiterte KB_GENERAL (Fix K25-K48) - Sport, Technik, Kochen, Psychologie
r'\b(hiit|high intensity interval|interval training)\b',
r'\bkrafttraining\b|\bkraft.*training\b|\bmuskeltraining\b',
r'(ausdauer.*kraft|kraft.*ausdauer|unterschied.*ausdauer)',
r'(oled.*led|led.*oled|qled.*oled|oled.*qled|unterschied.*fernseher)',
r'(energie.*klasse|energieeffizienz|stromverbrauch.*gerät|was bedeuten.*klassen)',
r'(usb.*a|usb.*c|lightning|unterschied.*usb|usb.*unterschied)',
r'(sous.*vide|sousvide|unterschied.*sous|vakuum.*garen)',
r'(salzen.*vor|salzen.*nach|fleisch.*salzen|wann.*salzen)',
r'(maillard.*reaktion|maillard|braten.*kruste|röstaromen)',
r'(sarkasmus.*ironie|ironie.*sarkasmus|unterschied.*sarkasmus)',
r'(gaslighting|gas.*lighting|manipulation.*wahrnehmung)',
r'(dunning.*kruger|dunningkruger|kompetenz.*überschätz|selbsteinschätzung)',
r'(confirmation.*bias|bestätigungsfehler|bestaetigungsfehler|bias.*bestätigung)',
r'(himmel.*blau.*warum|warum.*himmel.*blau|rayleigh)',
r'(meteor.*meteorit|meteorit.*asteroid|unterschied.*meteor)',
r'(eis.*schwimmt|schwimmt.*eis|eis.*wasser|warum.*eis)',
r'(freizügigkeitskonto|freizuegigkeitskonto|säule.*2.*konto|pensionskasse.*konto)',
r'(ahv.*lücke|ahv.*luecke|beitragslücke|beitragsluecke|rente.*lücke)',
r'(schaltjahr|schalt.*jahr|warum.*schaltjahr|29.*februar)',
r'(gmt.*utc|utc.*gmt|unterschied.*gmt|coordinated.*universal)',
# Neue Patterns für erweiterte KB_GENERAL (Fix K49-K64) - Auto, Reisen, Büro
r'(benzin.*diesel|diesel.*benzin|unterschied.*benzin|unterschied.*diesel|was ist.*diesel)',
r'(ampel.*farbe|ampel.*bedeutung|rot.*gelb.*grün|ampel.*regel)',
r'(rechts.*vor.*links|vorfahrt.*rechts|wer.*vorfahrt)',
r'(tempolimit|geschwindigkeit.*auto|wie schnell.*fahren|autobahn.*tempo)',
r'(handgepäck|aufgegabengepäck|flüssigkeiten.*flug|was.*handgepäck)',
r'(jetlag|jet.*lag|zeitverschiebung|umstellung.*zeit)',
r'(reisepass|personalausweis|unterschied.*pass|pass.*ausweis)',
r'(visum|visa|einreise.*erlaubnis|touristenvisum|schengen.*visum)',
r'(pomodoro.*technik|pomodoro|25.*minuten.*arbeit|produktiv.*technik)',
r'(pdf.*word|word.*pdf|unterschied.*pdf|dokument.*pdf)',
r'(cloud.*speicher|cloud.*storage|google.*drive|dropbox|icloud)(?!.*next)',
r'(e.*mail.*etikette|email.*etikette|betreff.*mail|anrede.*mail)',
)
# Fix KB-SCIENCE-NEW: Neue Patterns für KB_GENERAL_SCIENCE (Physik, Chemie, Astronomie)
SCIENCE_NEW_PATTERNS = (
r'(newton.*gesetz|newtonsches.*gesetz|trägheit|actio.*reactio|f.*=.*m.*a)',
r'(energie.*formen|kinetische.*energie|potentielle.*energie|thermische.*energie)',
r'(lichtgeschwindigkeit|schallgeschwindigkeit|wie schnell.*licht|wie schnell.*schall)',
r'(elektromagnetisch.*spektrum|radio.*welle|mikrowelle|infrarot|ultraviolett|röntgen|gamma)',
r'(aggregatzustand|fest.*flüssig.*gas|plasma.*zustand|zustand.*materie)',
)
# Fix KB-HEALTH: Gesundheit & Wohlbefinden Patterns für KB_HEALTH_WELLNESS
HEALTH_KB_PATTERNS = (
r'(schlaf.*dauer|schlaf.*stunden|wie lange.*schlafen|optimale.*schlaf|schlafhygiene)',
r'(power.*nap|powernap|nickerchen|mittagsschlaf|kurz.*schlaf)',
r'(wasser.*trinken|wieviel.*wasser|wie viel.*wasser|wasser.*pro.*tag|hydratation)',
r'(vitamin.*d|vitamin-d|magnesium|eisen.*mangel|mineralien|vitamine)',
r'(herzfrequenz|puls.*zone|training.*zone|fitness.*zone|ausdauer.*zone)',
r'(schritte.*tag|schritte.*pro|10\.000.*schritte|schrittzahl)',
r'(cortisol|stress.*hormon|stress.*reduktion|entspannung.*technik)',
r'(burnout|burn.*out|ueberlastung|überlastung|erschöpfung)',
r'(ballaststoffe|zucker.*konsum|makronaehrstoffe|makro.*nährstoff)',
r'(erste.*hilfe|schnittwunde|verbrennung|bewusstlos|bewusstlosigkeit)',
r'(schlaf.*qualitaet|besser.*schlafen|einschlaf.*tip|durchschlaf)',
r'(dehydration|dehydriert|durst|flüssigkeits.*mangel|wasser.*bedarf)',
r'(ph.*wert|ph.*skala|sauer.*basisch|säure.*lauge|zitronensaft.*ph)',
r'(periodensystem|chemisches.*element|wasserstoff|kohlenstoff|sauerstoff|eisen|gold)',
r'(zelle.*biologie|zellmembran|zytoplasma|zellkern|mitochondrien|dna.*erklärung)',
r'(photosynthese|pflanzen.*licht|co2.*wasser.*licht|chloroplasten)',
r'(dns.*basen|adenin.*thymin|guanin.*cytosin|doppelhelix|genetik.*einfach)',
r'(sonnensystem|planet.*reihenfolge|merkur.*venus.*erde|jupiter.*saturn|lichtjahr)',
r'(erde.*durchmesser|mond.*entfernung|astronomische.*einheit|ae.*entfernung)',
)
# Fix KB-EVERYDAY: Neue Patterns für KB_EVERYDAY_KNOWLEDGE (Kochen, Haushalt, Gesundheit)
EVERYDAY_KB_PATTERNS = (
r'(garzeit|kochzeit|nudeln.*kochen|reis.*kochen|kartoffeln.*kochen|eier.*kochen)',
r'(fleisch.*temperatur|kern.*temperatur|braten.*grad|steak.*medium)',
r'(haltbarkeit|wie lange.*haltbar|kühlschrank.*tage|aufbewahren)',
r'(flecken.*entfernen|rotwein.*fleck|kaffee.*fleck|fett.*entfernen|blut.*entfernen)',
r'(putzen.*tip|reinigung.*tip|hausputz|sauber.*machen|mikrofaser)',
r'(notruf|erste.*hilfe|bewusstlos|blutung|verbrennung|erste.*hilfe.*maßnahme)',
r'(vitamin.*funktion|vitamin.*quelle|vitamin.*a|vitamin.*c|vitamin.*d|vitamin.*b)',
r'(steuerklasse|steuerklasse.*i|steuerklasse.*iii|brutto.*netto|abkürzung.*kv)',
r'(passwort.*sicher|sicheres.*passwort|2fa|zwei.*faktor|passwort.*manager)',
r'(datei.*größe|kb.*mb.*gb|kilobyte|megabyte|gigabyte|terabyte)',
r'(internet.*geschwindigkeit|dsl.*kabel|glasfaser|5g.*geschwindigkeit)',
r'(müll.*trennung|gelber.*sack|biomüll|papier.*tonne|glas.*container)',
r'(energie.*sparen|strom.*sparen|led.*lampe|kühlschrank.*sparen)',
)
# Fix KB-AUTO: Neue Patterns für KB_AUTOMOBILITY (Autos, Fahrzeuge, Mobilität)
AUTO_KB_PATTERNS = (
r'(benzin.*diesel|diesel.*benzin|unterschied.*benzin|unterschied.*diesel|was ist.*diesel|e10.*benzin|super.*e10)',
r'(ölwechsel|öl.*wechsel|oelwechsel|oel.*wechsel|ölstand|oelstand|öl.*prüfen|oel.*pruefen)',
r'(reifen.*druck|reifendruck|reifen.*wechsel|winterreifen|sommerreifen|reifen.*profil|profiltiefe)',
r'(tanken|tankstelle|benzin.*preis|diesel.*preis|tank.*kosten|sprit.*kosten)',
r'(elektroauto|e-auto|eauto|ladestation|schnellladen|reichweite|akku.*auto|batterie.*auto)',
r'(führerschein|fuehrerschein|fahrerlaubnis|führerschein.*kategorie|fuehrerschein.*kategorie)',
r'(tempolimit|geschwindigkeit.*auto|wie schnell.*fahren|autobahn.*tempo|tempo.*limit)',
r'(auto.*kaufen|auto.*verkaufen|gebrauchtwagen|neuwagen|wertverlust|auto.*kosten)',
r'(versicherung.*auto|auto.*versicherung|kasko|vollkasko|haftpflicht|kfz.*versicherung)',
r'(promillegrenze|promille.*auto|alkohol.*fahren|trunkenheit|blutalkohol)',
r'(wartung.*auto|auto.*wartung|inspektion.*auto|auto.*inspektion|service.*auto)',
r'(ampel.*farbe|ampel.*bedeutung|rot.*gelb.*grün|ampel.*regel|vorfahrt.*regel)',
r'(rechts.*vor.*links|vorfahrt.*rechts|wer.*vorfahrt|vorfahrtsregel)',
r'(unfall.*auto|auto.*unfall|pannen|abschleppen|kfz.*schaden)',
r'(mietwagen|car.*sharing|carsharing|mobility|auto.*mieten)',
r'(motor.*funktioniert|wie.*funktioniert.*motor|ottomotor|dieselmotor|hybrid.*auto)',
r'(antrieb.*auto|frontantrieb|heckantrieb|allrad|quattro|xDrive|4matic)',
r'(verbrauch.*auto|auto.*verbrauch|liter.*100km|kraftstoff.*verbrauch)',
)
analysis = {
'role': role_hint if role_hint in ROLE_HINTS else 'general',
'task_type': 'explain',
'needs_setup_context': False,
'needs_web_research': False,
'needs_memory': False,
'dangerous_action': False,
'tools': [],
'reasoning_focus': ['Verstehe zuerst das eigentliche Objekt und den Mechanismus der Anfrage.'],
'confidence': 0.35,
}
if re.search(r'https?://', message):
analysis.update(role='research', task_type='explain', tools=['url_research'], needs_web_research=True, confidence=0.82)
return analysis
if is_followup_query(message):
prev = recent_topic_event(message) or recent_dialogue_event(message)
if prev:
prev_analysis = prev.get('analysis', {})
prev_tools = prev_analysis.get('tools', [])
# Fix 9: don't inherit url_research for ops/setup queries
OPS_SIGNALS = ('hoste ich', 'mein setup', 'meine infrastruktur', 'welche services', 'welche hosts', 'wo laeuft', 'oeffentlich', 'hosts laufen', 'laufen sie', 'gerade fehler')
if prev_tools == ['url_research'] and any(s in low for s in OPS_SIGNALS):
analysis.update(role='ops', task_type='status', tools=['setup_lookup'], needs_setup_context=True, confidence=0.87)
return analysis
resolved = resolve_followup(message, prev) if prev.get('summary') else None
focus = str((resolved or {}).get('focus') or str(prev.get('message') or '')).strip()
analysis.update(
role=prev_analysis.get('role', role_hint if role_hint in ROLE_HINTS else 'general'),
task_type=prev_analysis.get('task_type', 'explain'),
tools=prev_tools[:3],
needs_setup_context=bool(prev_analysis.get('needs_setup_context')),
needs_web_research=bool(prev_analysis.get('needs_web_research')),
needs_memory=bool(prev_analysis.get('needs_memory')),
confidence=0.92 if resolved and resolved.get('is_followup') else 0.9,
reasoning_focus=[f'Kurze Follow-up-Anfrage; nutze den zuletzt verifizierten Kontext: {focus}'],
)
return analysis
if re.search(r'^(bitte\s+)?(merke dir|merk dir|denk daran|behalte im kopf)\b', low) or re.search(r'was weisst du (alles |so )?(ueber|über) mich', low) or re.search(r'was weisst du über mich', message.lower()) or re.search(r'(mein profil|mein gespeichertes profil|profil anzeigen|profil zeigen|gespeichertes profil)', low) or re.search(r'\b(was ist mir .* wichtig|wie sollst du (mir )?antworten|welche vorlieben|welchen stil|welche art hilfe wuensche ich mir|welche art hilfe wünsche ich mir|welche praeferenzen hast du von mir gespeichert|welche präferenzen hast du von mir gespeichert)\b', low) or (re.search(r'was weisst du', low) and re.search(r'vorlieben', low)) or (re.search(r'wie sollst du mir antworten', low) and re.search(r'kennst', low)):
analysis.update(role='personal', task_type='memory', tools=['memory_profile'], needs_memory=True, confidence=0.92)
return analysis
if '?' not in message and re.search(r'\b(ich will|ich bevorzuge|mir ist wichtig|achte darauf|das ist falsch|es sollte|er sollte|wenn es um .* geht|ich .* wichtig finde)\b', low) and not re.search(r'\b(installieren|einrichten|deployen|konfigurieren|aufsetzen|einbinden|migrieren|upgraden)\b', low) and not re.search(r'(homelab|server|services).{0,50}(sicher|security|absichern|haerten)', low) and not re.search(r'(sicher genug|ausreichend sicher|homelab.*sicher|server.*sicher)', low) and not re.search(r'\b(wechseln|migrieren|umsteigen|umziehen|umstellen)\b', low) and not re.search(r'\bvon\b.{1,30}\b(auf|zu)\b.{1,30}\b(wechseln|migrieren|umsteigen|umziehen|umstellen)\b', low) and not re.search(r'(abnehmen|ernaehrungsplan|ernaehrung.*plan|trainingsplan|fitness.*plan|sport.*plan|diaet.*plan|kalorien.*plan|mahlzeiten.*plan|essensplan|gewicht.*verlieren|gewicht.*reduzier)', low):
analysis.update(role='personal', task_type='memory', tools=['memory_profile'], needs_memory=True, confidence=0.94)
return analysis
# Fix MEMORY-ROUTING-1: Persönliche Präferenz-Fragen ("Welche Art X mag ich?", "Was bevorzuge ich?")
# WICHTIG: Ausschluss für Gesundheitsbeschwerden ("ich habe kopfschmerzen") und Planung ("was habe ich geplant")
if re.search(r'\b(welche|was)\b.*\b(ich|mein|meine|mir)\b', low) and re.search(r'\b(mag|bevorzuge|nehme|trinke|esse|kaufe|nutze|verwende|sammle|interessiere)\b', low) and not re.search(r'\b(habe|hatte|bekomme|fuehle|fühle|spüre|spuere)\b.*\b(kopfschmerzen|schmerzen|weh|krank|beschwerden|geplant|vor|termin|woche)\b', low) and not re.search(r'(homelab|server|docker|container|service|host|setup|install|konfig)', low):
analysis.update(role='personal', task_type='memory', tools=['memory_profile'], needs_memory=True, confidence=0.93)
return analysis
if 'briefing' in low or 'lagebild' in low or 'was ist gerade wichtig' in low:
analysis.update(role='ops', task_type='status', tools=['briefing'], needs_setup_context=True, confidence=0.92)
return analysis
if re.search(r'\b(dnssec|markdown|html|backup|snapshot|ram|ssd)\b', low) and not re.search(r'(aktuell|heute|neueste|latest)', low):
if re.search(r'(was ist|erklaer|erkläre|in einfachen worten|in drei saetzen|in drei sätzen|unterschied|vergleiche? kurz)', low):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
if re.search(r'\b(sonarr|radarr)\b', low) and re.search(r'(unterschied|vergleiche?|vergleich)', low):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix PROMETHEUS-INSTALL: Service-Installation + Verifikation -> ops_install + ops_exec (Multi-step)
# Fix INSTALL-VERIFY: "installieren und danach verifizieren" -> Multi-step Pattern
if re.search(r'(prometheus|grafana|node.*exporter|cadvisor|loki|promtail|influxdb|telegraf|zabbix|nagios)', low) and re.search(r'(installieren|install|aufsetzen|deployen|einrichten)', low) and re.search(r'(verifizieren|verifizier|testen|prüfen|pruefen|checken|sicherstellen|danach|anschliessend|anschließend)', low):
analysis.update(role='ops', task_type='build', tools=['ops_install', 'ops_exec'], needs_setup_context=True, dangerous_action=True, confidence=0.96)
return analysis
# Fix PROBLEM-FOLLOWUP: "Was war das Problem" nach Diagnose -> ops_deep_analyze
if re.search(r'^(was war das problem|was war das|welches problem|welches war das problem|das problem)', low) and re.search(r'(gefunden|diagnose|analyse|vorher|eben|gerade|du|hast)', low):
analysis.update(role='ops', task_type='diagnose', tools=['ops_deep_analyze'], needs_setup_context=True, confidence=0.94)
return analysis
if any(hint in low for hint in SERVICE_HINTS if hint not in {"share", "smb"}) and "sonos" not in low and (re.search(r'\b(?:kann|can)\s+(?!mein\b|meine\b|my\b|der\b|die\b|das\b)[a-z0-9_.@-]{2,40}\s+nicht\b', low) or re.search(r'\b[a-z0-9_.@-]{2,40}\b.*(kann sich nicht|kommt nicht rein|hat keinen zugriff|hat kein zugriff|kann nicht zugreifen|kann sich nicht anmelden|sieht .* nicht|findet .* nicht|entdeckt .* nicht|hat keine berechtigung|permission|forbidden|denied)', low) or re.search(r'\bwarum\s+kann\s+[a-z0-9_.@-]{2,40}\s+sich\s+nicht\b', low) or re.search(r'\bwarum\s+kann\s+[a-z0-9_.@-]{2,40}\s+nicht\b', low)) and re.search(r'(zugreifen|zugriff|anmelden|login|einloggen|rein|access|sieht .* nicht|findet .* nicht|entdeckt .* nicht|berechtigung|permission|forbidden|denied|mfa|duo|2fa)', low):
analysis.update(role='ops', task_type='diagnose', tools=['user_service_access_diagnose'], needs_setup_context=True, confidence=0.95)
return analysis
if any(hint in low for hint in SERVICE_HINTS if hint not in {'share', 'smb'}) and re.search(r'(sieht .* nicht|findet .* nicht|entdeckt .* nicht|fehlt|fehlen|bibliothek|library|\bserie\b|\bstaffel\b|\bfolge\b|\bepisod|\bfilm\b|ordner|ressource)', low) and 'sonos' not in low and not re.search(r'\b(sonarr|radarr|lidarr)\b', low) and not re.search(r'\b(soll ich|oder|vs|versus|vergleich|einsetzen|nehmen|waehlen)\b', low):
analysis.update(role='ops', task_type='diagnose', tools=['user_service_access_diagnose'], needs_setup_context=True, confidence=0.93)
return analysis
if 'sonos' in low and re.search(r'(bibliothek|library|musik|music|share|freigabe|smb|nas|unraid|zugreifen|zugriff|findet|erreicht|scan)', low):
analysis.update(role='ops', task_type='diagnose', tools=['sonos_library_diagnose'], needs_setup_context=True, confidence=0.97)
return analysis
# CAS-1: mass user failure after config change → diagnose (not provision)
if re.search(r'(alle benutzer|alle user|niemand kann|keiner kann|kein benutzer|alle mitglieder).*(anmelden|einloggen|zugreifen|verbinden|nutzen)', low) or re.search(r'(anmelden|einloggen).*(alle|niemand|keiner|kein benutzer)', low) or re.search(r'(ldap|authentik|sso).*(aenderung|update|neuer|geaendert|provider|einstellung).*(anmelden|einloggen|zugriff|nutzen)', low) or re.search(r'(neuen? ldap|ldap.*provider|authentik.*ldap).*(benutzer|user|anmelden|einloggen)', low):
analysis.update(role='ops', task_type='diagnose', tools=['user_service_access_diagnose'],
needs_setup_context=True, confidence=0.92)
return analysis
if 'emby' in low and re.search(r'(einrichten|anlegen|neu(?:en|e|er)?.*benutzer|neuen.*account|ldap|authentik|wie bisherige|wie die bisherigen|wie die bestehenden|bestehenden benutzer|standard)', low):
analysis.update(role='ops', task_type='howto', tools=['emby_user_provision'], needs_setup_context=True, confidence=0.96)
return analysis
# Fix EMBY-GENERAL: Emby + Problem/Integration/Streaming/Abbruch -> setup_lookup (not general_answer)
# Ausschluss: Vergleiche (vs, oder, besser, empfehl) -> compare_selfhosted
if 'emby' in low and re.search(r'(bricht|ab|problem|fehler|streaming|stream|integration|integriere|media player|player|transcoding|hw.*accel|hardware.*accel|absturz|crashed|hängt|haengt|nicht mehr|kein bild|kein ton)', low) and not re.search(r'(vs|versus|oder.*besser|besser.*oder|empfiehl|soll ich.*oder)', low):
analysis.update(role='ops', task_type='diagnose', tools=['setup_lookup'], needs_setup_context=True, confidence=0.95)
return analysis
if re.search(r'(schalte|mach|ausschalten|einschalten|deaktivier|aktivier|turn off|turn on|dimm|dimmen|heller|dunkler|helligkeit)', low) and re.search(r'(licht|lichter|lights|wohnzimmer|buero|badezimmer|kueche|esszimmer|bedroom|kitchen|living|manuel|katja|alle)', low) and not re.search(r'(welche|welches|status|gerade|aktuell|eingeschaltet|eingeschalten|an sind|an ist)', low):
analysis.update(role='ops', task_type='action', tools=['light_control'], dangerous_action=True, confidence=0.97)
return analysis
if re.search(r'(welche lichter|welches licht|welche lampen|welche beleuchtung|welche lichtbereiche|lichtbereiche|gerade.*licht|aktuell.*licht|aktuelle.*beleuchtung|zimmerlampen|status.*licht|licht.*status|wie.*licht.*status|lights.*status|status.*lights)', low):
analysis.update(role='ops', task_type='status', tools=['light_status'], needs_setup_context=True, confidence=0.98)
return analysis
if 'licht' in low and re.search(r'(daten|automatis|wann|wie lange)', low):
analysis.update(role='ops', task_type='diagnose', tools=['light_usage'], needs_setup_context=True, confidence=0.88)
return analysis
# Fix W1: Erweiterte Web-Diagnose Patterns für interne Seiten/Doku
if (re.search(r'[a-z0-9.-]+\.[a-z]{2,}', low) and re.search(r'(nicht oeffnen|nicht öffnen|nicht laden|webseite|website|url|zugriff|erreichbar|ausfällt|ausfaellt)', low)) or (re.search(r'(subdomain|sub-domain|hostname|hostnamen|host|domain|wiki|wissensseite|wissens-url|doku-seite|dokuseite|dokumentationsseite|dokumentation|documentation|docs?|interne.*seite|intern.*seite|wissens.*seite)', low) and re.search(r'(nicht laden|nicht oeffnen|nicht öffnen|nicht aufzugehen|nicht auf|aufgehen|von aussen|von außen|von ausserhalb|ausserhalb|extern|erreichbar|oeffnen|ausfällt|ausfaellt|welche zwei infos|welche zwei angaben|brauchst du zuerst|was brauchst du zuerst|welche.*infos|welche.*angaben|brauchst du|was brauchst)', low)):
analysis.update(role='ops', task_type='diagnose', tools=['web_root_cause'], needs_setup_context=True, confidence=0.92)
return analysis
# LOG-COMPLEX: "häufigsten fehler", "fehler-patterns", "wochenbericht"
if re.search(r'(haeufigsten?\s*fehler|fehler.{0,20}haeufig|welche fehler|fehler.?patterns|fehler.?muster)', low) or re.search(r'(fehler.{0,30}(letzten|24|stunden|tagen|woche|monat)|wochenbericht.*fehler|log.*auswertung|log.*analyse|logs.*auswerter)', low) or re.search(r'(fehler.*tauchen.*logs|logs.*fehler|logs.*letzten|zeig.*fehler.*logs|fehler.*logs.*letzten)', low):
analysis.update(role='ops', task_type='diagnose', tools=['log_hotspots'],
needs_setup_context=True, confidence=0.90)
return analysis
# Fix PIHOLE-REPORT: PiHole DNS-Report/Statistik-Anfragen -> setup_lookup (not general_answer)
# Fix PIHOLE-CASE: Case-insensitive matching für Pihole/PiHole/pihole
# Fix PIHOLE-BOUNDARY: Match auch "Pihole-Report" (Bindestrich ist keine Wortgrenze)
if re.search(r'\bpihole[-_]?|\bpi[-_]?hole|\bpi-hole\b', low) and re.search(r'(report|statistik|statistiken|anfragen|blockiert|geblockt|dns.*heute|heute.*dns|wie viele|anzahl)', low):
analysis.update(role='ops', task_type='status', tools=['setup_lookup'], needs_setup_context=True, confidence=0.94)
return analysis
if re.search(r'\b(logs?)\b', low) and re.search(r'(immich|nextcloud|traefik)', low):
analysis.update(role='ops', task_type='diagnose', tools=['log_hotspots'], needs_setup_context=True, confidence=0.92)
return analysis
if re.search(r'(langsam|langsamer|performance|cpu|ram|disk|netzwerk)', low) and 'host' in low:
analysis.update(role='ops', task_type='diagnose', tools=['host_slow_diagnose'], needs_setup_context=True, confidence=0.9)
return analysis
if 'kasm' in low and re.search(r'(warum|wieso|zugreifen|kann|problem|fehler)', low):
analysis.update(role='ops', task_type='diagnose', tools=['kasm_diagnose'], needs_setup_context=True, confidence=0.9)
return analysis
if re.search(r'(homelab|reverse proxy|reverse-proxy|ssh|secrets|backups?|sicherer|sicherheit|security)', low) and re.search(r'(massnahmenplan|massnahmenplan|\bplan\b|priorisiert|haerten|haerten|recherchiere)', low) and not re.search(r'(urlaub|reise|trip|ferien).{0,30}planen', low):
analysis.update(role='research', task_type='howto', tools=['security_plan'], needs_setup_context=True, needs_web_research=True, confidence=0.95)
return analysis
if re.search(r'(hoerbuch|hörbuch|audiobook|audiobooks?)', low) and re.search(r'(server|system|tool|dienst|plattform|plattformen|option|optionen|vergleich|vergleiche|gegeneinander|abwaegen|abwägen|loesung|lösung|empfehl|waehlen|wählen|priorisieren|aufsetzen|homelab|setup|stack)', low):
analysis.update(role='research', task_type='compare', tools=['compare_selfhosted'], needs_web_research=True, confidence=0.95)
return analysis
if any(t in low for t in ('pihole', 'adguard', 'vaultwarden', 'bitwarden', 'jellyfin', 'plex', 'navidrome', 'paperless', 'gitea', 'forgejo', 'uptime-kuma', 'portainer', 'heimdall', 'dashdot', 'overseerr', 'bazarr', 'prowlarr', 'readarr', 'kavita')) and re.search(r'(besser|oder|vs|vergleich|vergleiche|unterschied|welches|empfiehl|nehmen|waehlen|wählen|statt|anstatt)', low):
analysis.update(role='research', task_type='compare', tools=['compare_selfhosted'], needs_web_research=True, confidence=0.87)
return analysis
# MIG-VS: "X vs Y" / "soll ich X oder Y" service comparison
_cs = r'(nextcloud|seafile|jellyfin|emby|plex|vaultwarden|bitwarden|portainer|traefik|nginx|proxmox|unraid|truenas|synology|docker|podman|gitea|forgejo|immich|photoprism|audiobookshelf|navidrome|lidarr|sonarr|radarr)'
if re.search(_cs + r'\s*(vs\.?|versus)\s*' + _cs, low) or re.search(r'(soll ich|lohnt sich|empfiehlst du).{0,60}' + _cs + r'.{0,40}(oder|vs|anstatt|statt)', low) or re.search(r'(soll ich|lohnt sich|empfiehlst du).{0,60}' + _cs + r'.{0,60}' + _cs, low) or re.search(r'(was ist besser|welches ist besser|was empfiehlst).{0,80}' + _cs, low) and not re.search(r'\bfuer homelab\b', low) or re.search(_cs + r'.{0,40}(oder|vs|versus).{0,40}' + _cs + r'.{0,50}(einsetzen|verwenden|nehmen|waehlen|nutzen)', low):
analysis.update(role='ops', task_type='compare', tools=['compare_selfhosted'],
needs_setup_context=False, confidence=0.91)
return analysis
# MIG: migration / platform switch / "von X auf Y wechseln"
_mig_svcs = r'(unraid|proxmox|docker|truenas|synology|qnap|ubuntu|debian|jellyfin|emby|nextcloud|vaultwarden|bitwarden|portainer|seafile|plex|gitea|forgejo)'
if re.search(r'\bvon\b.{1,40}\b(auf|zu|nach)\b.{1,40}\b(wechseln|migrieren|umsteigen|umziehen|umstellen)\b', low) or re.search(r'\bvon\b\s+' + _mig_svcs + r'.{0,20}\b(auf|zu)\b', low) or re.search(_mig_svcs + r'.{1,30}\b(wechseln|migrieren|umsteigen|umziehen|umstellen)\b', low) or re.search(r'(was muss ich beachten.*(wechseln|migrieren|umsteigen)|was verliere ich dabei|was aendert sich beim wechsel)', low):
analysis.update(role='ops', task_type='compare', tools=['compare_selfhosted'],
needs_setup_context=True, confidence=0.90)
return analysis
# EXPERT-RESEARCH early intercept: deep/comprehensive research queries → expert_research before compare_selfhosted
if re.search(r'(recherchiere.{0,30}(tiefgr|tiefsinnig|vollst|komplett|umfassend|detailliert)|'
r'marktanalyse|wettbewerbsanalyse|kompetitor.*analyse|competitive.*analysis|'
r'investigier|investigative|vollstaendige.*analyse|comprehensive.*analysis|'
r'fact.?check|fakten.*pruefen|quellen.*pruefen|verifizier.*fakten|'
r'recherche.*bericht|research.*report|due.*diligence)', low):
analysis.update(role='research', task_type='explain', tools=['expert_research'], needs_web_research=True, confidence=0.93)
return analysis
if re.search(r'(selbst zu hosten|selbst hosten|selber hosten|self-hosted|self hosted)', low):
analysis.update(role='research', task_type='compare', tools=['compare_selfhosted'], needs_web_research=True, confidence=0.84)
return analysis
# Fix PROTO-SYNTH: Protokoll/Technologie-Erklaerung + was nutzt [Service] konkret
# MUSS VOR allgemeinem 'vergleich' Pattern kommen!
# Catches: "Unterschied OAuth2/OIDC/SAML + was nutzt Authentik?"
if any(h in low for h in SERVICE_HINTS) and re.search(
r'(oauth|oidc|saml|jwt|openid|ldap|'
r'protokoll.*unterschied|unterschied.*protokoll|auth.*protokoll|'
r'was ist.*unterschied|unterschied.*zwischen|erklaer.*protokoll|erklär.*protokoll|'
r'was ist.*oauth|was ist.*oidc|was ist.*saml|was ist.*ldap)', low) and re.search(
r'(nutzt|verwendet|benutzt|einsetzt|konkret|im einsatz|wie setzt.*ein|wie verwendet|was.*nutzt|was.*verwendet)', low):
analysis.update(role='ops', task_type='explain', tools=['ops_deep_analyze'],
needs_setup_context=True, confidence=0.91)
return analysis
# Fix ARCH-REASONING: Architektur/Abhängigkeits/Kaskaden-Fragen -> ops_deep_analyze
# Catches: 'was passiert wenn X ausfällt', 'wie hängen X und Y zusammen', 'Abhängigkeitskette'
if any(h in low for h in SERVICE_HINTS) and re.search(
r'(abhaengigkeits?kette|abhaengigkeiten|zusammenspiel|wie haengen|wie hängen|was passiert wenn|'
r'wenn .{0,30} ausfaellt|wenn .{0,30} ausfällt|wenn .{0,30} down|cascad|kaskad|'
r'zusammenhaeng|zusammenhäng|wie.*funktioniert.*zusammen|login.*flow|login-flow|auth.*flow|'
r'welche.*abhaengig|welche.*abhängig|ich plane .{0,40} zu deployen)', low):
analysis.update(role='ops', task_type='explain', tools=['ops_deep_analyze'],
needs_setup_context=True, confidence=0.93)
return analysis
if re.search(r'(recherchiere|vergleich|vergleiche)', low):
wants_selfhosted = bool(re.search(r'(hosten|hosting|lokal|on-prem|onprem|selbst|self)', low))
analysis.update(role='research', task_type='compare', tools=['compare_selfhosted'] if wants_selfhosted else ['web_research'], needs_web_research=True, confidence=0.82)
return analysis
if re.search(r'(welche services|welche dienste|welche apps|welche anwendungen|welche app|welche anwendung|was hoste ich|hoste ich aktuell|oeffentlich|öffentlich|extern erreichbar|oeffentlich erreichbar|öffentlich erreichbar|gesund|oeffentliche app|öffentliche app|oeffentliche anwendung|öffentliche anwendung)', low) and re.search(r'(fehler|warnings?|warnungen|probleme|kaputt|status|zustand|gesund|gestoert|gestört|stoerung|störung|instabil|instabilsten|unreachable|down)', low):
analysis.update(role='ops', task_type='status', tools=['public_services_health'], needs_setup_context=True, confidence=0.95)
return analysis
if re.search(r'(hoste ich|mein setup|meine infrastruktur|wo laeuft|welche hosts|welche services|oeffentlich)', low):
analysis.update(role='ops', task_type='status', tools=['setup_lookup'], needs_setup_context=True, confidence=0.86)
return analysis
# Fix K13: Homelab service + HTTP error code -> setup_lookup (before GENERAL_KB_PATTERNS)
if any(s in low for s in ('traefik', 'nextcloud', 'immich', 'emby', 'authentik', 'nginx')) and re.search(r'(502|503|404|500|401|403|error|fehler|zurück|zurueck|gibt|return)', low):
analysis.update(role='ops', task_type='diagnose', tools=['setup_lookup'], needs_setup_context=True, confidence=0.94)
return analysis
# Fix STATUS-REPORT: "Statusbericht", "Bericht", "Übersicht" + Homelab/Setup -> setup_lookup
if re.search(r'(statusbericht|status.*bericht|bericht|übersicht|uebersicht|zustand|gesamtzustand)', low) and re.search(r'(homelab|setup|infrastruktur|services|dienste|server|host|system)', low):
analysis.update(role='ops', task_type='status', tools=['setup_lookup'], needs_setup_context=True, confidence=0.95)
return analysis
# Fix NC-OPT: "kannst du mir [SERVICE] optimieren" -> setup_lookup (not web_research)
if re.search(r'(kannst du mir|kannst du|hilf mir|unterstütze mich|unterstuetze mich).{0,30}(optimieren|verbessern|tunen|konfigurieren|einrichten|aufräumen|aufrauemen)', low) and any(s in low for s in ('nextcloud', 'immich', 'traefik', 'authentik', 'emby', 'unraid', 'proxmox', 'docker', 'nginx', 'tailscale', 'home assistant', 'homeassistant', 'kasm', 'pihole', 'adguard', 'wireguard', 'sonarr', 'radarr', 'lidarr', 'audiobookshelf')):
analysis.update(role='ops', task_type='setup', tools=['setup_lookup'], needs_setup_context=True, confidence=0.92)
return analysis
# Fix POWER-OUTAGE: Stromausfall + Homelab/Server -> setup_lookup (Diagnose nötig)
if re.search(r'(stromausfall|strom.*aus|blackout|strom.*weg|strom.*unterbrechung|strom.*wieder|nach.*strom)', low) and re.search(r'(homelab|server|dienste|services|nicht mehr|komplett|funktioniert|erreichbar|neustart|restart)', low):
analysis.update(role='ops', task_type='diagnose', tools=['setup_lookup'], needs_setup_context=True, confidence=0.96)
return analysis
# Fix 13: Recipe/cooking questions covered by KB_RECIPES.md -> direct general_answer
# Fix R1: Erweiterte Rezept-Patterns für bessere Abdeckung
# Fix STAT4: past-tense food/activity queries → memory_profile (not recipe)
if re.search(r'(was habe ich|was hatte ich|was.*ich.*(gemacht|getan|gelernt|gelesen|getrunken|gegessen|geschlafen|trainiert))', low) and re.search(r'(letzte woche|gestern|letzten|vorige woche|letztem monat|letztes jahr|heute morgen|heute frueh)', low):
analysis.update(role='personal', task_type='memory', tools=['memory_profile'], needs_memory=True, confidence=0.89)
return analysis
RECIPE_PATTERNS_EXTENDED = RECIPE_PATTERNS + (
r'(gib mir.*rezept|gib.*rezept|rezept.*bitte|rezept fuer|rezept f.r)',
r'(omeletten|omelette|omelett|eier.*kochen|eier.*braten)',
r'(schnell.*kochen|einfach.*kochen|was.*kochen|was.*essen)',
)
if any(re.search(p, low) for p in RECIPE_PATTERNS_EXTENDED) and not any(h in low for h in ('homelab', 'server', 'docker', 'container', 'menu', 'code', 'script', 'programm')) and not re.search(r'(normalerweise|gewoehnlich|meistens|morgens|abends|taeglich|stets)', low):
analysis.update(role='personal', task_type='howto', tools=['personal_assist'], needs_memory=True, confidence=0.93)
return analysis
# Also catch broader food queries (Fix 7 restored)
if re.search(r'(ideen|abendessen|mittagessen|fruehstueck|rezept|zutaten|kochen|backen|was essen|was koche|was esse|zum essen|schnelles essen)', low) and not any(h in low for h in ('homelab', 'server', 'docker', 'container', 'menu')) and not re.search(r'(normalerweise|gewoehnlich|meistens|oft|immer|morgens|abends|taeglich|stets|ueblicherweise|typisch)', low):
analysis.update(role='personal', task_type='howto', tools=['personal_assist'], needs_memory=True, confidence=0.90)
return analysis
# Fix WEB-ENTITY: Definition/Entity questions that general-local-assist rejects -> web_research FIRST
# These patterns must come BEFORE general_answer patterns to avoid ungrounded results
# BUT: Skip if query matches KB patterns (KB_GENERAL or KB_SCIENCE) to avoid double-routing
# Fix SETUP-PRE: Homelab/Setup-Queries müssen VOR WEB_ENTITY_PATTERN geprüft werden
SETUP_PRE_PATTERN = r'(hoste ich|mein setup|meine infrastruktur|welche services|welche hosts|wo laeuft|welche hosts|welche services|oeffentlich|öffentlich|extern erreichbar|oeffentlich erreichbar|öffentlich erreichbar)'
if re.search(SETUP_PRE_PATTERN, low) and re.search(r'(ich|mein|meine|mir)', low):
analysis.update(role='ops', task_type='status', tools=['setup_lookup'], needs_setup_context=True, confidence=0.92)
return analysis
WEB_ENTITY_PATTERN = r'^(was ist|wer ist|wo ist|wo liegt|wo befindet sich|wie leben|wie lebt|wo leben|wo lebt|was essen|was frisst|wie gross ist|wie groß ist|wie alt wird|wie funktioniert)\b'
if re.search(WEB_ENTITY_PATTERN, low):
# Check if this is covered by KB patterns - if so, let KB patterns handle it
kb_covered = False
# Check against GENERAL_KB_PATTERNS (need to check before they're defined, so inline check)
# Fix KB-COVERAGE: Erweiterte Liste deckt alle GENERAL_KB_PATTERNS ab
kb_general_terms = ['docker', 'reverse proxy', 'tailscale', 'truenass', 'wetter', 'klima',
'photosynthese', 'lichtjahr', 'fleisch.*auftauen', 'waschmittel',
'kühlschrank', 'schnittwunde', 'wasser.*trinken', 'schlaf.*dauer',
'säulen.*system', 'etf', 'du.*sie.*deutsch', 'prozent', 'trinkgeld',
'kaffee.*kochen', 'tee.*zubereiten', 'eier.*kochen', 'pasta.*kochen',
'bohnen.*einweichen', 'reis.*kochen', 'avocado.*reif', 'brot.*aufbewahren',
'zitronen.*haltbar', 'butter.*haltbar', 'öl.*haltbar', 'honig.*haltbar',
'käse.*haltbar', 'zwiebeln.*haltbar', 'knoblauch.*haltbar',
# Neue KB-General Begriffe (K1-K48)
'tage bis', 'stunden.*jahr', 'vitamin d', 'kopfschmerzen', 'gestresst',
'schweiz.*gegründet', 'http.*status', 'passwort.*ändern', 'python.*error',
'übersetze', 'buch.*empfehl', 'ip.*adresse', 'dns', 'vpn',
'router.*switch', 'makronaehrstoff', 'glykaemisch', 'keto',
'inflation', 'zinseszins', 'aktien.*etf', 'waschmaschine',
'kuehlschrank.*funktioniert', 'backpulver', 'jahreszeiten',
'biologisch.*oekologisch', 'oekosystem', 'hiit', 'krafttraining',
'ausdauer', 'oled', 'energie.*klasse', 'usb', 'sous.*vide',
'salzen', 'maillard', 'sarkasmus', 'gaslighting', 'dunning',
'confirmation.*bias', 'meteor', 'eis.*schwimmt', 'freizügigkeitskonto',
'ahv.*lücke', 'schaltjahr', 'gmt.*utc']
for term in kb_general_terms:
if re.search(term.replace('.*', '.*'), low):
kb_covered = True
break
# Check against SCIENCE_KB_PATTERNS
if not kb_covered:
science_terms = ['rayleigh', 'lichtstreuung', 'himmel.*blau', 'mondfinsternis',
'blutmond', 'mondtäuschung', 'mond.*grösser', 'mond.*größer']
for term in science_terms:
if re.search(term, low):
kb_covered = True
break
if not kb_covered:
# Tech architecture topics -> expert_code (not web_research)
_tech_entity_topics = r'(microservice|event.driven|cqrs|saga.pattern|message.queue|rabbitmq|kafka|kubernetes|k8s|terraform|ansible|design.pattern|dependency.injection|solid.prinzip|clean.architektur|hexagonal|rest.*graphql|graphql.*rest|websocket|grpc|jwt.*flow|oauth.*flow|openid.*flow|ci.cd|gitflow|feature.flag|blue.green|canary.deploy|serverless|nosql.*sql|sharding|datenbank.*indexing|orm.*raw.*sql|async.*await|threading.*async|race.condition|deadlock|idempotent|eventual.consistency|cap.theorem|domain.driven|ddd|bounded.context)'
if re.search(_tech_entity_topics, low):
analysis.update(role='dev', task_type='explain', tools=['qdrant_search', 'expert_code'], confidence=0.89)
return analysis
analysis.update(role='research', task_type='explain', tools=['web_research'], needs_web_research=True, confidence=0.85)
return analysis
# Fix 10: Science questions covered by KB_SCIENCE_COMMON.md -> direct general_answer
# Fix S1: Erweiterte Science-Patterns für Mondfinsternis, Sonnenfarbe
SCIENCE_KB_PATTERNS_EXTENDED = SCIENCE_KB_PATTERNS + (
r'(mondfinsternis.*rot|mondfinsternis.*kupfer|roter mond|kupferfarbener mond|mond.*rot|mond.*kupfer)',
r'(sonne.*rot|sonne.*orange|sonnenuntergang.*rot|sonnenuntergang.*orange|untergehende sonne|sonne.*untergang.*farbe)',
r'(mond.*grösser|mond.*größer|mondtäuschung|mondtaeuschung|mond.*groß|mond.*gross)',
r'(himmel.*blau|blauer himmel|warum.*himmel|licht.*himmel)',
)
if any(re.search(p, low) for p in SCIENCE_KB_PATTERNS_EXTENDED):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix KB-SCIENCE-NEW: Neue Science-Patterns für erweiterte KB_GENERAL_SCIENCE
if any(re.search(p, low) for p in SCIENCE_NEW_PATTERNS):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix 15: General knowledge questions covered by KB_GENERAL.md -> direct general_answer
if any(re.search(p, low) for p in GENERAL_KB_PATTERNS):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix KB-EVERYDAY: Alltagswissen-Fragen für KB_EVERYDAY_KNOWLEDGE
if any(re.search(p, low) for p in EVERYDAY_KB_PATTERNS):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix KB-HEALTH: Gesundheit & Wohlbefinden Fragen für KB_HEALTH_WELLNESS
if any(re.search(p, low) for p in HEALTH_KB_PATTERNS):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix KB-LANGUAGES: Sprach- & Übersetzungsfragen für KB_LANGUAGES
LANGUAGE_KB_PATTERNS = (
r'(übersetze|uebersetze|auf englisch|auf deutsch|auf italienisch|auf französisch|wie sagt man|wie heisst|wie heißt)',
r'(du.*sie.*deutsch|sie.*du.*deutsch|wann du.*wann sie|anrede.*deutsch|höflich.*anreden)',
r'(grammatik|zeitformen|konjugation|deklinieren|tenses|present|past|future|perfect)',
r'(false friends|falsche freunde|übersetzungsfehler|übersetzung.*fehler|false.*friend)',
r'(sprache.*lernen|lernen.*sprache|vokabeln|aussprache|sprachniveau|cefr|a1|b1|c1.*sprache)',
r'(email.*etikette|email.*betreff|schlussformel|grussformel|anrede.*email)',
r'(smalltalk|konversation|kommunikation.*tip|gespräch.*führen|small.*talk)',
r'(ciao.*bedeutung|buongiorno|bonjour.*bedeutung|grüße.*italienisch|grüße.*französisch)',
)
if any(re.search(p, low) for p in LANGUAGE_KB_PATTERNS):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix KB-FINANCE: Finanz- & Geldanlagefragen für KB_FINANCE
FINANCE_KB_PATTERNS = (
r'\b(etf|etfs|aktienfond|indexfond|msci world|msci|sp 500|dax|nasdaq)\b',
r'(wie.*anlegen|geld.*anlegen|anlage.*tip|wohin.*geld|sparen.*tip|budget.*tip)',
r'(tagesgeld|festgeld|sparbuch|zinsen|zinssatz|konto.*gebühr|bank.*gebühr)',
r'(steuerklasse|abgeltungssteuer|freistellungsauftrag|steuer.*freibetrag|kapitalertrag)',
r'(versicherung.*tip|haftpflicht|hausrat|berufsunfähigkeit|rechtschutz)',
r'(notgroschen|notfall.*geld|3 monatsgehälter|6 monatsgehälter)',
r'(bitcoin|ethereum|krypto|cryptowährung|wallet|blockchain|coin)',
r'(depot.*eröffnen|broker.*vergleich|neo.*broker|trade republic|scalable)',
r'(50/30/20|fünfzig dreißig zwanzig|budget.*regel|sparrate)',
)
if any(re.search(p, low) for p in FINANCE_KB_PATTERNS):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix KB-LEGAL: Recht & Verbraucherfragen für KB_LEGAL_CONSUMER
LEGAL_KB_PATTERNS = (
r'(kündigungsfrist|kündigen|vertrag.*kündigen|kündigung)',
r'(widerrufsrecht|rückgabe|umtausch|14 tage|14 tage.*recht)',
r'(gewährleistung|garantie|reklamation|mängel|mangel)',
r'(mietrecht|mieterhöhung|kaution|miete.*recht)',
r'(arbeitsrecht|urlaub.*recht|kündigungsschutz|arbeitsvertrag)',
r'(bußgeld|bussgeld|punkte.*führerschein|führerschein.*punkte|verkehrsrecht)',
r'(dsgvo|datenschutz|auskunft.*daten|recht.*vergessen)',
r'(erbrecht|testament|pflichtteil|erbe.*recht)',
r'(verbraucherrecht|mängelrüge|schadensersatz|verbraucher)',
)
if any(re.search(p, low) for p in LEGAL_KB_PATTERNS):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix KB-TRAVEL: Reise & Transport Fragen für KB_TRAVEL_TRANSPORT.md
TRAVEL_KB_PATTERNS = (
r'(bahn|zug|db|sparbpreis|bahncard|quer-durchs-land|deutschlandticket|verspätung|anschluss|zug.*verpasst)',
r'(handgepäck|handgepaeck|flüssigkeit|fluessigkeit|packen|koffer|reiseapotheke|reise.*packen)',
r'(flug|fluggastrecht|eu261|überbuchung|ueberbuchung|annullierung|flug.*annulliert|flug.*verspätet)',
r'(mietwagen|auto.*mieten|vignette|umweltzone|crit.air|critair|grüne.*plakette|gruene.*plakette)',
r'(flixbus|fernbus|bus.*fahren|bus.*reise)',
r'(reise.*buchen|urlaub.*buchen|beste.*buchungszeit|wann.*buchen|früh.*buchen|frueh.*buchen)',
r'(reisedokument|reisepass|personalausweis|visum|einreise|impfung|reise.*versicherung)',
)
if any(re.search(p, low) for p in TRAVEL_KB_PATTERNS):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix KB-COOKING: Kochtechniken & Küchenwissen für KB_COOKING_TECHNIQUES
COOKING_KB_PATTERNS = (
r'(braten|schmoren|dünsten|duensten|blanchieren|sous.*vide|garen|garzeit|gar.*zeit)',
r'(julienne|brunoise|würfeln|wuerfeln|schneiden|hacken|messer.*schärfen|messer.*schaerfen)',
r'(mehlschwitze|roux|binden.*sauce|reduzieren.*sauce|montieren.*butter)',
r'(kerntemperatur|kern.*temperatur|fleisch.*temperatur|braten.*grad|steak.*medium)',
r'(marinieren|marinade|fleisch.*marinieren|fleisch.*würzen|fleisch.*wuerzen)',
r'(hefeteig|rührteig|biskuitteig|mürbeteig|blätterteig|gären|gaeren|backen)',
r'(mise en place|zubereiten|kochtechnik|küchenwissen|kuechenwissen)',
r'(sauce.*machen|sosse.*machen|fond.*machen|braten.*anbraten)',
r'(gemüse.*kochen|gemuese.*kochen|gemüse.*garen|gemuese.*garen|blanchieren)',
r'(fleisch.*ruhen|fleisch.*ruhen.*lassen|braten.*ruhen)',
)
if any(re.search(p, low) for p in COOKING_KB_PATTERNS):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix KB-PETS: Haustiere & Tierhaltung Fragen für KB_PETS_ANIMALS.md
PETS_KB_PATTERNS = (
r'\b(hund|hunde|welpe|welpen|rasse|hunderasse|hunderassen|hundeerziehung|hundetraining)\b',
r'\b(katze|katzen|kitten|kastration|katzenklo|kratzbaum|freigänger|freigaenger)\b',
r'\b(haustier|haustiere|tierhaltung|tierarzt|tiergesundheit|impfung.*tier|entwurmung)\b',
r'(kaninchen|meerschweinchen|hamster|maus|maeuse|vogel|voegel|wellensittich|aquarium|fisch|fische)',
r'(hundefutter|katzenfutter|tierfutter|barf|trockenfutter|nassfutter|futter)',
r'(hund.*temperament|katze.*charakter|welpe.*erziehen|hund.*trainieren)',
r'(giftig.*hund|giftig.*katze|schokolade.*hund|schokolade.*katze|lilien.*katze)',
r'(zecken|zeckenschutz|floh|flöhe|flöhe|entwurmung|impfung.*hund|impfung.*katze)',
r'(hund.*auslauf|katze.*auslauf|gassi|spazieren|hund.*bewegung)',
r'(aquarium.*einrichten|aquarium.*wasserwerte|fische.*haltung|aquaristik)',
r'(tierverhalten|hundesprache|katzensprache|hund.*kommunikation|katze.*kommunikation)',
r'(hundesteuer|hundepflichtversicherung|tier.*versicherung|chip.*tier)',
)
if any(re.search(p, low) for p in PETS_KB_PATTERNS):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix KB-AUTO: Auto & Mobilität Fragen für KB_AUTOMOBILITY.md
if any(re.search(p, low) for p in AUTO_KB_PATTERNS):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix KB-HOUSE-GARDEN: Haus & Garten Wissen für KB_HOUSE_GARDEN.md
HOUSE_GARDEN_PATTERNS = (
r'\b(monstera|sansevieria|ficus|calathea|zamioculcas|pflanze|pflanzen|zimmerpflanze|zimmerpflanzen)\b',
r'\b(sukkulente|sukkulenten|kakteen|kaktus)\b',
r'\b(giessen|giessen|giess|gieß|gieße|wasser.*pflanze|pflanze.*wasser)\b',
r'\b(umtopfen|neuer.*topf|topf.*wechseln|pflanze.*umtopfen)\b',
r'\b(garten|beet|rasen|mähen|maehen|rasen.*mähen|rasen.*maehen)\b',
r'\b(tomaten|gemüse|gemuese|kräuter|kraeuter|hochbeet|aussaat|ernte)\b',
r'\b(unkraut|schädling|schaedling|blattlaus|schnecke|mehltau|krankheit.*pflanze)\b',
r'\b(kompost|kompostieren|komposthaufen|komposterde)\b',
r'\b(gartengerät|gartengeraet|spaten|schere|schaufel|rechen|hacke)\b',
r'\b(frühling|fruehling|sommer|herbst|winter|saison|jahreszeit.*garten)\b',
r'\b(streichen|wand.*streichen|farbe.*wand|anstrich|renovieren)\b',
r'\b(löcher.*wand|loch.*wand|wand.*loch|reparieren.*wand|dübel|duebel)\b',
r'\b(wasserhahn|hahn.*tropft|tropfender.*hahn|armatur.*reparieren)\b',
r'\b(drainage|topf.*drainage|pflanze.*drainage|wurzelfäule|wurzelfaeule)\b',
r'\b(lichtbedarf|sonne.*pflanze|pflanze.*sonne|halbschatten|schattenpflanze)\b',
r'\b(garzeit|kochzeit|nudeln.*kochen|reis.*kochen|kartoffeln.*kochen|eier.*kochen)\b',
r'\b(fleisch.*temperatur|kern.*temperatur|braten.*grad|steak.*medium)\b',
r'\b(haltbarkeit|wie lange.*haltbar|kühlschrank.*tage|aufbewahren)\b',
r'\b(flecken.*entfernen|rotwein.*fleck|kaffee.*fleck|fett.*entfernen|blut.*entfernen)\b',
r'\b(putzen.*tip|reinigung.*tip|hausputz|sauber.*machen|mikrofaser)\b',
)
if any(re.search(p, low) for p in HOUSE_GARDEN_PATTERNS) and not any(s in low for s in ('homelab', 'server', 'docker', 'container', 'service', 'host', 'setup', 'infrastruktur', 'nextcloud', 'traefik', 'proxmox', 'unraid')):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix KB-SPORT-FITNESS: Sport, Fitness & Bewegung für KB_SPORT_FITNESS.md
# WICHTIG: Patterns müssen spezifisch sein - keine Planungs- oder Homelab-Queries matchen
SPORT_FITNESS_PATTERNS = (
# Spezifische Sportarten/Disziplinen (nicht "sport" allein)
r'\b(krafttraining|cardio|ausdauertraining)\b',
r'\b(laufen\b|joggen\b|radfahren\b|schwimmen\b|marathon\b|triathlon\b|ironman\b)(?!\s*sollten|\s*soll|\s*werden|\s*können|\s*koennen)',
r'\b(yoga\b|pilates\b|crossfit\b|calisthenics\b|bodybuilding\b|powerlifting\b)',
r'\b(kampfsport\b|boxen\b|kickboxen\b|muay thai\b|bjj\b|judo\b|mma\b|wrestling\b)',
r'\b(teamsport\b|fussball\b|basketball\b|volleyball\b|handball\b|eishockey\b)',
# Spezifische Übungen/Equipment
r'\b(kniebeuge\b|bankdrücken\b|kreuzheben\b|deadlift\b|squat\b|bench press\b)',
r'\b(klimmzug\b|liegestütz\b|liegestuetz\b|pushup\b|pullup\b|muscle up\b|plank\b)',
r'\b(hantel\b|hanteln\b|kettlebell\b|widerstandsband\b|resistance band\b)',
r'\b(laufschuhe|sport.*schuhe|schuhe.*laufen)',
# Trainingskonzepte (nicht "training" allein)
r'\b(workout\b|trainingsplan\b|trainingsplanung\b|progressive überlastung)',
r'\b(wiederholungen\b|sätze\b|satz\b|satz.*zahl|wiederholungszahl)',
r'\b(herzfrequenz\b|puls\b|puls.*zone|zone 2|zone 3|hiit\b|liss\b)',
r'\b(aufwaermen|aufwärmen|mobilität|dehnübung|dehnuebung)',
# Ernährung spezifisch für Sport
r'\b(protein.*muskel|muskel.*protein|kreatin\b|supplement\b)',
r'\b(ernährung.*sport|sport.*ernährung|fitness.*ernährung)',
# Verletzungen/Regeneration
r'\b(verletzung.*sport|sport.*verletzung|muskelkater\b|regeneration\b)',
r'\b(schlaf.*muskel|muskel.*schlaf|muskel.*wachstum)',
)
# Nur matchen wenn KEINE Planungs- oder Homelab-Keywords vorhanden
if any(re.search(p, low) for p in SPORT_FITNESS_PATTERNS):
if not any(s in low for s in ('homelab', 'server', 'docker', 'container', 'service', 'host', 'setup', 'infrastruktur', 'plane', 'planen', 'woche', 'monat', 'januar', 'februar', 'märz', 'april', 'mai', 'juni', 'juli', 'august', 'september', 'oktober', 'november', 'dezember')):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix DNS-PROBLEM: DNS-Auflösungsprobleme im Homelab → setup_lookup (not general_answer)
if re.search(r'(dns|dns.*aufloesung|dns.*auflösung|domain.*nicht|namensauflösung)', low) and re.search(r'(funktioniert.*nicht|problem|fehler|nicht.*erreichbar|timeout)', low):
analysis.update(role='ops', task_type='diagnose', tools=['setup_lookup'], needs_setup_context=True, confidence=0.94)
return analysis
# Fix KB-EVERYDAY-TECH: Alltägliche Technik & Gadgets für KB_EVERYDAY_TECH.md
EVERYDAY_TECH_PATTERNS = (
r'\b(iphone|android|smartphone|handys?|tablets?|ipad)\b',
r'\b(speicher voll|akku|batterie|laden|ladekabel|face id|touch id|fingerabdruck)\b',
r'\b(screenshot|bildschirmfoto|drucken|tastenkürzel|shortcut)\b',
r'\b(passwort|passwort-manager|2fa|zwei-faktor|sicher)\b',
r'\b(phishing|spam|betrug|virus|malware|schutz)\b',
r'\b(wlan|wifi|internet|router|verbindung|netzwerk)\b',
r'\b(bluetooth|kopfhörer|airpods|lautsprecher|boxen)\b',
r'\b(streaming|netflix|spotify|qualität|daten|auflösung)\b',
r'\b(drucker|scanner|drucken|tinte|laser)\b',
r'\b(e-mail|email|anhang|mail|posteingang)\b',
r'\b(whatsapp|signal|telegram|messenger|chat)\b',
r'\b(update|software|app|programm|installieren|deinstallieren)\b',
r'\b(cloud|icloud|google drive|dropbox|backup|speichern)\b',
r'\b(usb|kabel|ladegerät|adapter|hdmi|displayport)\b',
r'\b(airplay|chromecast|bildschirm.*spiegeln|screen mirroring)\b',
r'\b(alexa|google assistant|siri|smart home|smarte steckdose)\b',
r'\b(thermostat|heizung.*steuern|smart.*heizung)\b',
r'\b(fotos?|bilder|videos?|galerie|kamera)\b',
r'\b(datei.*größe|kb|mb|gb|kilobyte|megabyte|gigabyte)\b',
r'\b(energie.*klasse|stromverbrauch|verbrauch.*gerät)\b',
r'\b(oled|qled|led|fernseher|tv|bildschirm|monitor)\b',
r'\b(pdf|word|dokument|datei.*konvertieren|konvertieren)\b',
r'\b(pomodoro|produktiv|zeitmanagement|zeit.*managen)\b',
)
if any(re.search(p, low) for p in EVERYDAY_TECH_PATTERNS) and not any(s in low for s in ('homelab', 'server', 'docker', 'container', 'service', 'host', 'setup', 'infrastruktur', 'nextcloud', 'traefik', 'proxmox', 'unraid')) and not re.search(r'(skalier|von.*auf.*user|performance.*app|bottleneck|load.balanc)', low) and not re.search(r'(email.*(chef|vorgesetzt|manager|leitung|gehal|bonus)|wegen.*gehal)', low):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix KB-EVERYDAY-PROBLEMS: Alltags-Probleme & Life Hacks für KB_EVERYDAY_PROBLEMS.md
EVERYDAY_PROBLEMS_PATTERNS = (
r'(smartphone.*lädt|handy.*lädt|lädt.*nicht|ladeproblem|ladekabel|akku.*problem)',
r'(laptop.*heiss|laptop.*laut|laptop.*heiß|lüfter.*laut|cpu.*100)',
r'(wlan.*langsam|internet.*langsam|wifi.*langsam|verbindung.*schlecht)',
r'(bluetooth.*koppelt|koppelt.*nicht|bluetooth.*problem)',
r'(fleck.*entfernen|flecken.*entfernen|kaffeefleck|rotweinfleck|fettfleck)',
r'(abfluss.*verstopft|verstopft.*abfluss|rohr.*verstopft|abfluss.*reinigen)',
r'(mikrowelle.*reinigen|mikrowelle.*sauber|mikrowelle.*putzen)',
r'(kühlschrank.*stinkt|kühlschrank.*geruch|kühlschrank.*muffig)',
r'(kleidung.*trocknen|schnell.*trocknen|nass.*trocknen)',
r'(schuhe.*trocknen|schuhe.*nass|schuhe.*feucht)',
r'(reißverschluss.*klemmt|reißverschluss.*hakt|zipper.*klemmt)',
r'(messer.*schärfen|messer.*scharf|stumpf.*messer)',
r'(wein.*öffnen|korkenzieher|flasche.*öffnen|weinflasche)',
r'(eier.*schälen|eier.*pellen|schälen.*eier)',
r'(knoblauch.*hände|knoblauchgeruch|geruch.*hände)',
r'(avocado.*reif|avocado.*reifen|avocado.*schneller)',
r'(einschlafen|nicht.*müde|schlafen.*nicht|schlaf.*problem)',
r'(sonnenbrand|verbrannt|haut.*rot|sonne.*verbrannt)',
r'(schnupfen|erkältet|erkältung|krank|husten|kopfschmerzen)',
r'(stich|insekt|mückenstich|bienenstich|juckt|jucken)',
r'(batterie.*tot|auto.*springt|starten.*nicht|starthilfe)',
r'(koffer.*packen|packen.*koffer|effizient.*packen)',
r'(jetlag|zeitverschiebung|müde.*reise|schlaf.*reise)',
r'(papierstau|drucker.*stau|drucker.*problem)',
r'(passwort.*vergessen|einloggen.*nicht|zugang.*vergessen)',
r'(life.*hack|lifehack|trick|tipp|hilfe|funktioniert.*nicht|geht.*nicht|kaputt)',
)
if any(re.search(p, low) for p in EVERYDAY_PROBLEMS_PATTERNS) and not any(s in low for s in ('homelab', 'server', 'docker', 'container', 'service', 'host', 'setup', 'infrastruktur', 'nextcloud', 'traefik', 'proxmox', 'unraid')):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix KB-HOWTO-EXPLAIN: Alltags-Erklärungen & How-Tos für KB_HOWTO_EXPLAIN.md
HOWTO_EXPLAIN_PATTERNS = (
r'(wie sage ich|wie lehne ich|wie gebe ich.*feedback|wie sage ich.*zu spät)',
r'(wie lehne ich.*ab|höflich.*ablehnen|ablehnen.*einladung)',
r'(konstruktives feedback|feedback.*geben|kritik.*äußern)',
r'(wie putze ich|badezimmer.*putzen|putzen.*badezimmer)',
r'(wie wasche ich.*bettwäsche|bettwäsche.*waschen)',
r'(wie entferne ich.*kalk|kalk.*entfernen|kalk.*armatur)',
r'(wie lagere ich.*gemüse|gemüse.*lagern|gemüse.*kühlschrank)',
r'(wie erstelle ich.*passwort|sicheres passwort|passwort.*erstellen)',
r'(wie sichere ich.*daten|daten.*sichern|backup.*regel)',
r'(wie erkenne ich.*phishing|phishing.*erkennen|phishing.*mail)',
r'(wie gehe ich.*schlafstörung|schlafstörung.*umgehen|schlafhygiene)',
r'(wie trinke ich.*mehr wasser|mehr wasser.*trinken)',
r'(wie mache ich.*pause|effektive pause|pomodoro)',
r'(wie erstelle ich.*haushaltsplan|haushaltsplan.*erstellen|50/30/20)',
r'(wie sortiere ich.*unterlagen|unterlagen.*sortieren|wes-system)',
r'(schlafhygiene|einschlafen|schlaf.*tip|besser.*schlafen)',
r'(trinken.*wasser|wasser.*trinken|mehr.*wasser)',
r'(haushaltsplan|budget.*plan|finanz.*plan)',
r'(unterlagen.*sortieren|dokumente.*sortieren|papier.*sortieren)',
)
if any(re.search(p, low) for p in HOWTO_EXPLAIN_PATTERNS) and not any(s in low for s in ('homelab', 'server', 'docker', 'container', 'service', 'host', 'setup', 'infrastruktur', 'nextcloud', 'traefik', 'proxmox', 'unraid')):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix STAT: Personal tracking/stats queries → memory_profile
# Fix STAT-BOOKS: "wie viele buecher habe ich..." Pattern
if re.search(r'(wie viel.*(km|kilometer).*ich|\bkm\b.*gelaufen|laufstatistik|wie oft.*gelaufen|'
r'schlafzeiten|schlafdauer|wann.*geschlafen|letzte.*schlaf|wie lange.*geschlafen|'
r'trainingsstatistik|zeig.*meine.*statistik|meine.*trainings|wie oft.*trainiert|'
r'wie viele.*buecher.*(ich|habe|gelesen)|buecher.*(ich|habe).*(gelesen|dieses jahr|letztes jahr)|gelesen.*dieses jahr|'
r'meine.*lekture|wie viele.*(ich|habe).*(gelesen|geschrieben|gemacht|getan).*(jahr|monat))', low):
analysis.update(role='personal', task_type='memory', tools=['memory_profile'], needs_memory=True, confidence=0.89)
return analysis
# Fix STAT-PASTTENSE: what did I do/eat last week → memory_profile (not recipe)
if re.search(r'(was habe ich.*(letzte woche|gestern|letzten|vorige woche|letztem monat).*(gegessen|getrunken|gemacht|getan|gelernt)|'
r'was.*ich.*(letzten|letzte woche|gestern).*(gegessen|getrunken|gemacht|getan))', low):
analysis.update(role='personal', task_type='memory', tools=['memory_profile'], needs_memory=True, confidence=0.89)
return analysis
# Fix 1: Personal preference questions -> memory_profile
# Fix PREF-BOOKS: "was ist mein lieblingsbuch" Pattern
if re.search(r'(was esse ich|was trinke ich|was hore ich|was lese ich|was schaue ich)', low) and re.search(r'(normalerweise|gewoehnlich|meistens|oft|immer|morgens|abends|taeglich|stets|ueblicherweise|typisch)', low):
analysis.update(role='personal', task_type='memory', tools=['memory_profile'], needs_memory=True, confidence=0.91)
return analysis
if re.search(r'\b(welche art|welche|was fuer|was fur|wie viel|wieviel|welchen|am liebsten)\b', low) and re.search(r'\b(ich|mein|meine|mir)\b', low) and re.search(r'\b(mag|bevorzuge|trinke|esse|hore|hoere|lese|moechte|mochte|vorliebe|gewohnheit|kaffee|tee|musik|buch|buecher|filme|sport|essen|gericht|liebsten|lieber)\b', low):
analysis.update(role='personal', task_type='memory', tools=['memory_profile'], needs_memory=True, confidence=0.9)
return analysis
# Fix 17: Personal location + known places → memory_profile
if re.search(r"(sitterdorf|thurgau.*erlen|erlen.*thurgau)", low) and not re.search(r"(komme ich|fahrt|strecke|route|bahn|zug|bus|wie.*nach|von.*nach|entfernung|navigation)", low):
analysis.update(role='personal', task_type='memory', tools=['memory_profile'], needs_memory=True, confidence=0.93)
return analysis
if re.search(r'(wo wohne ich|wo lebe ich|meine adresse|mein wohnort|wo wohn|wo leb)', low) and re.search(r'\b(ich|mir|mein|wir)\b', low):
analysis.update(role='personal', task_type='memory', tools=['memory_profile'], needs_memory=True, confidence=0.92)
return analysis
# ── TECH-EXPLAIN INTERCEPT ────────────────────────────────────────────────
# Tech-Themen bei erklaer/wie funktioniert -> expert_code (nicht general_answer)
_tech_explain_topics = r'(microservice|event.driven|cqrs|saga.pattern|message.queue|rabbitmq|kafka|kubernetes|k8s|terraform|ansible|grafana|prometheus|design.pattern|dependency.injection|solid.prinzip|clean.architektur|hexagonal|rest.*graphql|graphql.*rest|websocket|grpc|jwt.*flow|oauth.*flow|openid.*flow|ci.cd.*pipeline|gitflow|feature.flag|a.b.testing|blue.green|canary.deploy|serverless|lambda.funktion|nosql.*sql|sharding|indexing.*datenbank|orm.*raw.*sql|async.*await|promise.*callback|threading.*async|race.condition|deadlock|idempotent|eventual.consistency|cap.theorem)'
if re.search(r'^(erklaer|erklaere|wie funktioniert|wie funk|was ist der unterschied|unterschied zwischen|was sind die vor|vor.*und nachteil)', low) and re.search(_tech_explain_topics, low):
analysis.update(role='dev', task_type='explain', tools=['qdrant_search', 'expert_code'], confidence=0.89)
return analysis
# SKALIER INTERCEPT: Skalierungs- und App-Design-Fragen -> expert_strategy+code
if re.search(r'(wie skalier|wie.*von.*auf.*user|wie.*app.*skalier|horizontal.*skalier|vertical.*skalier|load.*balanc|wie.*performance.*verbessern.*app|wie.*app.*schneller|bottleneck.*app|database.*skalier)', low) and not any(s in low for s in ('homelab', 'nextcloud', 'emby', 'traefik')):
analysis.update(role='dev', task_type='explain', tools=['expert_strategy'], confidence=0.88)
return analysis
if re.search(r'^(warum|weshalb|wieso|wie entsteht|wie funktioniert|erklaer|erklaere)\b', low) and not any(h in low for h in ('server', 'host', 'docker', 'kasm', 'ssl', 'cert', 'dns', 'traefik', 'nginx', 'container', 'homelab', 'setup', 'infrastruktur', 'authentik', 'tailscale', 'proxmox', 'unraid', 'linux', 'nextcloud', 'immich', 'emby', 'lidarr', 'radarr', 'sonarr', 'audiobookshelf')):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.78)
return analysis
# Fix DOCKER-MAINT-2: docker maintenance/cleanup → setup_lookup (MUST come before JARVIS patterns)
if re.search(r'(docker|container)', low) and re.search(r'(aufraeumen|aufraumen|prune|bereinigen|cleanup|clean.*up)', low):
analysis.update(role='ops', task_type='setup', tools=['setup_lookup'], needs_setup_context=True, confidence=0.94)
return analysis
# CONV-EXEC: Container action commands (restart/stop/start) → ops_exec
if re.search(r'(neu starten|neustart|restart|stoppen|starten|kill|beenden|stop|start).{0,30}(container|docker|service)', low) or re.search(r'(container|docker|service).{0,30}(neu starten|neustart|restart|stoppen|starten)', low):
analysis.update(role='ops', task_type='action', tools=['ops_exec'], needs_setup_context=False, confidence=0.91)
return analysis
# Fix DOCKER-ROUTING-2: Docker/Container Inventar-Queries → setup_lookup (not ops_api)
# "zeig mir alle docker container", "docker list", "welche container laufen"
if re.search(r'(zeig mir|zeige mir|zeig|zeige|liste|list|alle|welche).{0,15}(docker|container)', low) and re.search(r'(docker|container|laufen|vorhanden|installiert|hoste)', low):
analysis.update(role='ops', task_type='status', tools=['setup_lookup'], needs_setup_context=True, confidence=0.94)
return analysis
# Fix DOCKER-ROUTING-3: Docker compose/maintenance → setup_lookup
if re.search(r'(docker|compose).{0,20}(update|upgrade|aktualisieren|durchfuehren|ausfuehren)', low):
analysis.update(role='ops', task_type='setup', tools=['setup_lookup'], needs_setup_context=True, confidence=0.93)
return analysis
# Fix NC-ROUTING: "kannst du [SERVICE] prüfen" → setup_lookup (Inventar/Config, nicht Live-API)
# WICHTIG: prufen (mit u) und pruefen (mit ue) beide abdecken
if re.search(r'(kannst du|koenntest du).{0,30}(prufen|prüfen|pruefen|checken|testen|ueberpruefen|ueberprüfen|uberprufen|überprüfen)', low) and any(s in low for s in ('nextcloud', 'immich', 'traefik', 'authentik', 'emby', 'unraid', 'proxmox', 'docker', 'nginx', 'tailscale', 'home assistant', 'homeassistant', 'kasm', 'pihole', 'adguard', 'wireguard', 'sonarr', 'radarr', 'lidarr', 'audiobookshelf')):
analysis.update(role='ops', task_type='setup', tools=['setup_lookup'], needs_setup_context=True, confidence=0.92)
return analysis
# CONV-OPS: Conversational queries with embedded ops commands → ops_api
# "kannst du mal schauen ob container laufen" / "was läuft auf dem server"
if re.search(r'(kannst du|koenntest du|magst du|bitte|koennte man|schau mal|prüf mal|pruef mal|check mal|zeig mir|zeige mir|schau ob|pruef ob|check ob)', low) and re.search(r'(container|docker|server|services?|homelab|nextcloud|sonarr|radarr|emby|vaultwarden|pihole|ollama|traefik)', low):
analysis.update(role='ops', task_type='status', tools=['ops_api'], needs_setup_context=False, confidence=0.90)
return analysis
# CONV-RUNNING: "was läuft (gerade|auf)" → ops_api live status
if re.search(r'(was laueft|was läuft|was laeuft|wer laueft|wer läuft).{0,30}(server|docker|container|homelab|gerade|tower|ubuntu)', low) or re.search(r'(server|homelab|docker).{0,20}(was laueft|was läuft|was laeuft|aktiv|online|running|up|stehen)', low):
analysis.update(role='ops', task_type='status', tools=['ops_api'], needs_setup_context=False, confidence=0.90)
return analysis
# CONV-HOWIS: "wie geht es dem/der [service]" → ops_api live check
if re.search(r'(wie geht es|wie steht es|wie ist es um).{0,20}(container|docker|nextcloud|sonarr|radarr|emby|vaultwarden|pihole|ollama|traefik|postgresql|service|server)', low):
analysis.update(role='ops', task_type='status', tools=['ops_api'], needs_setup_context=False, confidence=0.89)
return analysis
# CONV-STATUS: "wie läuft/steht es mit" specific services → ops_api
if re.search(r'(wie laueft|wie läuft|wie steht).{0,30}(nextcloud|sonarr|radarr|emby|vaultwarden|pihole|ollama|traefik|docker|container)', low):
analysis.update(role='ops', task_type='status', tools=['ops_api'], needs_setup_context=False, confidence=0.89)
return analysis
# JARVIS: complex diagnose+fix tasks → ops_deep_analyze directly (it already does infra checks)
if re.search(r'(unhealthy|kaputt|defekt|fehler|exited|problem).{0,50}(container|service|docker)', low) and \
re.search(r'(wie|was|fix|reparier|beheb|loesung|was soll|wie kann|wie loes|was tun|schritte|vorgehen)', low):
analysis.update(role='ops', task_type='diagnose', tools=['ops_deep_analyze'], needs_setup_context=False, confidence=0.94)
return analysis
# JARVIS: explicit deep / root-cause analysis → ops_deep_analyze
if re.search(r'(tiefe.*analyse|deep.*analyse|vollstaendig.*analysier|root.*cause|ursache.*finden|warum.*kaputt|warum.*fehler|detailliert.*analysier|grundursache)', low):
analysis.update(role='ops', task_type='diagnose', tools=['ops_deep_analyze'], needs_setup_context=False, confidence=0.93)
return analysis
# JARVIS: unhealthy container diagnose → ops_deep_analyze for full log analysis
if re.search(r'(unhealthy|health.*container|container.*health|kaputt.*container|container.*kaputt|defekt.*container)', low):
analysis.update(role='ops', task_type='diagnose', tools=['ops_deep_analyze'], needs_setup_context=False, confidence=0.95)
return analysis
# JARVIS: docker RAM/memory/volume/disk analysis → ops_api
if re.search(r'(container.*ram|container.*memory|container.*speicher|ram.*container|docker.*volume|docker.*disk|docker.*platz|volume.*groesse|volume.*size|docker.*speicher)', low):
analysis.update(role='ops', task_type='diagnose', tools=['ops_api'], needs_setup_context=False, confidence=0.93)
return analysis
# JARVIS: TLS/cert check → ops_api
if re.search(r'(tls.*zertifikat|zertifikat.*ablauf|cert.*ablauf|traefik.*tls|traefik.*cert|letsencrypt.*ablauf|acme.*check|https.*cert)', low):
analysis.update(role='ops', task_type='diagnose', tools=['ops_api'], needs_setup_context=False, confidence=0.93)
return analysis
# JARVIS: security audit docker → ops_api
if re.search(r'(security.*audit|docker.*sicherheit|docker.*audit|container.*sicherheit|privileged.*container|container.*root|sicherheitsaudit)', low):
analysis.update(role='ops', task_type='diagnose', tools=['ops_api'], needs_setup_context=False, confidence=0.94)
return analysis
# JARVIS: pihole / DNS stats → ops_api
if re.search(r'(pihole|pi-hole|pi hole|dns.*blockiert|blockierte.*domains|dns.*report|dns.*statistik)', low):
analysis.update(role='ops', task_type='status', tools=['ops_api'], needs_setup_context=False, confidence=0.95)
return analysis
# JARVIS: postgresql / database → ops_api
if re.search(r'(postgresql|postgres.*datenbank|datenbank.*groesse|db.*groesse|postgresql.*check|postgres.*query|datenbank.*report)', low):
analysis.update(role='ops', task_type='status', tools=['ops_api'], needs_setup_context=False, confidence=0.94)
return analysis
# JARVIS: full homelab status report → public_services_health
if re.search(r'(vollstaendiger.*statusbericht|homelab.*statusbericht|statusbericht.*homelab|homelab.*report|vollstaendiger.*bericht|alle.*services.*status)', low):
analysis.update(role='ops', task_type='status', tools=['public_services_health'], needs_setup_context=True, confidence=0.93)
return analysis
# Fix DOCKER-INVENTORY: docker/container INVENTORY queries → setup_lookup (not ops_api)
# These are "what do I have" questions, not "what's the current live status"
if re.search(r'(docker|container)', low) and re.search(r'(zeig mir alle|alle docker|wie viele|wie viel|count|anzahl|liste aller|list all)', low):
analysis.update(role='ops', task_type='status', tools=['setup_lookup'], needs_setup_context=True, confidence=0.92)
return analysis
# Fix 18 (upgraded): docker/container queries with live data → ops_api
if re.search(r'(docker|container|pod|stack|compose)', low) and re.search(r'(zeig|status|liste|list|alle|laufen|running|starten|stoppen|restart|logs|pruef|check|unhealthy|health|ram|memory|speicher|volume|disk|platz|security|sicherheit|audit|privileged|zertifikat|cert|tls)', low):
analysis.update(role='ops', task_type='status', tools=['ops_api'], needs_setup_context=False, confidence=0.91)
return analysis
# Fix T1: IP-Adresse / Netzwerk-Info Abfragen → setup_lookup
if re.search(r'(ip adresse|ip-adresse|tailscale ip|tailscale-ip|meine ip|welche ip|zeig.*ip|lautet.*ip|ip.*tailscale)', low):
analysis.update(role='ops', task_type='status', tools=['setup_lookup'], needs_setup_context=True, confidence=0.87)
return analysis
# Fix OPS-PROC: server process/monitoring queries -> setup_lookup
if re.search(r'(ubuntu|server|host|proxmox|unraid|docker)', low) and re.search(r'(prozesse|laufende services|ps aux|top.*cpu|systemctl list|running.*services|logs.*anzeigen|zeige.*logs|logs.*zeigen)', low):
analysis.update(role='ops', task_type='status', tools=['setup_lookup'], needs_setup_context=True, confidence=0.86)
return analysis
# AMB-SERVER-SLOW: personal + server slow → ops signal dominates
if re.search(r'(server|proxmox|homelab|vm|container|service).*(langsam|traege|haengt|keine reaktion|hohe last|ueberlastet|cpu last|ram last|speicher last)', low) or re.search(r'(langsam|traege).*(server|proxmox|homelab|vm|container|service)', low):
analysis.update(role='ops', task_type='diagnose', tools=['setup_lookup'],
needs_setup_context=True, confidence=0.88)
return analysis
# Fix OPS-SETUP: homelab config/setup/install tasks → setup_lookup
if any(s in low for s in ('proxmox', 'traefik', 'nextcloud', 'authentik', 'nginx', 'unraid', 'docker', 'ubuntu', 'tailscale', 'immich', 'homelab', 'sonarr', 'radarr', 'lidarr', 'audiobookshelf', 'emby', 'home assistant', 'homeassistant', 'kasm', 'wireguard', 'fail2ban', 'pihole', 'adguard')) and re.search(r'(einrichten|installieren|konfigurieren|aufsetzen|erstelle.*vm|neue.*vm|reverse proxy.*host|synchronisiere|deploye|setup|upgrade|migrieren|einbinden|integrieren|verbinden|kalibrieren|anpassen|profil.*erstellen|snapshot|auflisten|aufliste|sichern|restore|wiederherstellen|backup.*job|clonen|klonen)', low):
analysis.update(role='ops', task_type='howto', tools=['setup_lookup'], needs_setup_context=True, confidence=0.87)
return analysis
# AMB-SECURITY-DOMINATES: homelab security signal dominates personal request
if re.search(r'(homelab|server|services).*(sicher|security|absichern|geschuetzt|firewall|angriff)', low) or re.search(r'(sicher genug|ausreichend sicher|gut geschuetzt|sicherheit.*homelab|homelab.*sicherheit)', low):
analysis.update(role='ops', task_type='setup', tools=['setup_lookup'],
needs_setup_context=True, confidence=0.88)
return analysis
# SEC-INCIDENT: suspicious requests, brute force, intrusion detection
if re.search(r'(komische|verdaechtige|seltsame|unbekannte|fremde).*(requests?|anfragen|ips?|zugriffe|verbindungen)', low) or re.search(r'(chinesische|russische|unbekannte).*(ips?|adresse|anfragen)|ip.*(china|russland|unknown)', low) or re.search(r'(\d+)\s*mal.*(erfolglos|fehlgeschlagen|versucht|failed).*(einlog|einzulog|anmeld|zugriff|login|loggen)', low) or re.search(r'(brute.?force|einbruchsversuch|hack.{0,20}versuch|angriff|intruder|eindringling)', low) or re.search(r'(moeglicher angriff|möglicher angriff|unter angriff|wird angegriffen|jemand.*(versucht|probiert).*(einlog|hack|zugriff))', low) or re.search(r'(viele.*(401|403|4\d\d).*(unbekannt|fremd|verdaechtig)|log.*verdaechtig|logs.*(angriff|hack))', low):
analysis.update(role='ops', task_type='diagnose', tools=['setup_lookup'],
needs_setup_context=True, confidence=0.91)
return analysis
# Fix SEC-HARDEN: security hardening queries → setup_lookup
if re.search(r'(server|ssh|linux|ubuntu|proxmox|host|system|container|homelab).*(absichern|sichern|haerten|hartung|haertung|hardening|absicherung|security)', low) or re.search(r'(ssh|system|server|linux).*(hartung|haertung|haerten)\b', low) or re.search(r'\b(ssh|fail2ban|firewall|ufw|iptables).*(einrichten|konfigurieren?|absichern|aufsetzen)', low) or re.search(r'\b(konfiguriere?|einrichte?|richte.{0,10}ein).{0,20}\b(fail2ban|ufw|iptables|firewall|ssh)\b', low) or re.search(r'\b(fail2ban|ufw|iptables|firewall)\b.{0,80}\b(sperren|blockieren|bannen|gesperrt|so dass|damit)', low):
analysis.update(role='ops', task_type='setup', tools=['setup_lookup'], needs_setup_context=True, confidence=0.87)
return analysis
# SET-INTEGRATION: "integriere X mit Y", n8n automation, webhooks
if re.search(r'(integriere?|verbinde?|koppel?|synchro).{0,20}(home.?assistant|emby|nextcloud|sonarr|radarr|n8n|telegram|slack|webhook)', low) or re.search(r'(home.?assistant|emby|nextcloud|n8n).{0,30}(integriere?|verbinde?|koppel?)', low) or re.search(r'(n8n.{0,30}(webhook|trigger|workflow|automatisier|notification|benachrichtig))', low) or re.search(r'(automation.{0,30}(homelab|services)|homelab.{0,30}automation)', low):
analysis.update(role='ops', task_type='setup', tools=['setup_lookup'],
needs_setup_context=True, confidence=0.88)
return analysis
# Fix DOCK-MAINT: docker maintenance tasks → setup_lookup
if re.search(r'\bdocker\b', low) and re.search(r'(volumes?.*aufraeumen|aufraeumen.*volumes?|prune|bereinigen|compose.*update|compose.*upgrade|alle.*container.*aktualisier|container.*log|logs.*container|images.*aufraeumen)', low):
analysis.update(role='ops', task_type='setup', tools=['setup_lookup'], needs_setup_context=True, confidence=0.87)
return analysis
# CAS-STROMAUSFALL: nach stromausfall/neustart homelab nicht hochgekommen
if re.search(r'(stromausfall|stromunterbrechung|nach dem neustart|nach neustart|nach.*(reboot|restart)).*(homelab|server|services|vms?|container)', low) or re.search(r'(homelab|server|services).*(nicht.*(hochgekommen|gestartet|oben|erreichbar)|down|ausgefallen).*(stromausfall|neustart|reboot)', low) or re.search(r'(stromausfall.*(homelab|server)|homelab.*(stromausfall|ausgefallen|nicht.*hoch))', low):
analysis.update(role='ops', task_type='diagnose', tools=['setup_lookup'],
needs_setup_context=True, confidence=0.90)
return analysis
# Fix ERR: Raw error message input → setup_lookup
if re.search(r'(econnrefused|connection refused|\[emerg\]|fatal.*failed|permission denied.*publickey|port.*allocated|address.*already.*in.*use|bind\(\).*failed|error response from daemon)', low):
analysis.update(role='ops', task_type='diagnose', tools=['setup_lookup'], needs_setup_context=True, confidence=0.87)
return analysis
# Fix OLLAMA: ollama/LLM server queries → setup_lookup
if re.search(r'\bollama\b', low) or (re.search(r'\bllm\b|\bmodell\b', low) and re.search(r'(mac.?studio|mac.?mini|server|lauft|startet|laeuft|welches)', low)):
analysis.update(role='ops', task_type='setup', tools=['setup_lookup'], needs_setup_context=True, confidence=0.87)
return analysis
# Fix OPS-CRASH: generic homelab/server crash → setup_lookup
if re.search(r'(homelab|server|host|alle.*services|gesamte.*infrastruktur)', low) and re.search(r'(abgestuerzt|abgestuerzt|nicht mehr erreichbar|komplett down|alles down|crash|ausgefallen|ausgefallen)', low):
analysis.update(role='ops', task_type='diagnose', tools=['setup_lookup'], needs_setup_context=True, confidence=0.89)
return analysis
# Fix MED-1: sonarr/radarr/lidarr content/indexer issues → setup_lookup
if re.search(r'\b(sonarr|radarr|lidarr|audiobookshelf)\b', low) and re.search(r'(findet|steckt|haengt|fehler|problem|indexer|episod|serie|film|qualitaet|profil|download|hinzufuegen|nicht|konfigur)', low):
analysis.update(role='ops', task_type='diagnose', tools=['setup_lookup'], needs_setup_context=True, confidence=0.87)
return analysis
# Fix 19: service status queries → setup_lookup
if re.search(r'(status meiner|status der|meine.*status|services.*status|dienste.*status|welche.*laufen|alle.*services)', low) and not re.search(r'(licht|lampe|beleuchtung)', low):
analysis.update(role='ops', task_type='status', tools=['setup_lookup'], needs_setup_context=True, confidence=0.87)
return analysis
# Fix NET: network diagnostic queries → setup_lookup
if re.search(r'(\bport\b.*(nicht erreichbar|geschlossen|blockiert|offen|listen|filtered)|'
r'\bping\b.*(fehl|timeout|keine antwort|schlaegt|schlagt)|'
r'\btraceroute\b|\bdns\b.*(funktioniert nicht|aufloesung|resolution|fehl|problem)|'
r'\bnslookup\b|\bdig\b.*(domain|dns)|'
r'netzwerk.*problem|netzwerkfehler|verbindung.*timeout|connection.*timeout)', low):
analysis.update(role='ops', task_type='diagnose', tools=['setup_lookup'], needs_setup_context=True, confidence=0.87)
return analysis
# Fix SSL: SSL/TLS/certbot issues → setup_lookup
if re.search(r'(ssl.*(zertifikat|cert|ablaufen|abgelaufen|fehler|ungueltig|invalid|handshake)|'
r'zertifikat.*(abgelaufen|ablaufen|ungueltig|fehler|erneuern)|'
r'\bcertbot\b|lets.?encrypt|letsencrypt|\bacme\b)', low):
analysis.update(role='ops', task_type='diagnose', tools=['setup_lookup'], needs_setup_context=True, confidence=0.87)
return analysis
# Fix 20: homelab service fault/check → setup_lookup
if any(s in low for s in ('nextcloud', 'immich', 'traefik', 'authentik', 'emby', 'lidarr', 'radarr', 'sonarr', 'audiobookshelf', 'unraid', 'proxmox', 'docker', 'nginx', 'tailscale', 'home assistant', 'homeassistant', 'kasm', 'pihole', 'adguard', 'wireguard')) and re.search(r'(nicht|problem|fehler|kaputt|down|error|pruef|prufen|check|zeig|status|warum|erreichbar|langsam|performance|ram|cpu|speicher|bricht|abbricht|haengt|haengt sich|crash|absturz|abgestuerzt|friert ein|not working|broken|offline|stuck|failed|issue|sensor.*offline|offline.*sensor|schlaegt fehl|schlagt fehl|funktioniert nicht|sync.*fehl|fehl.*sync|ldap.*fehl|fehl.*ldap)', low):
analysis.update(role='ops', task_type='diagnose', tools=['setup_lookup'], needs_setup_context=True, confidence=0.88)
return analysis
# Fix HA: Home Assistant queries → setup_lookup (entitaten, dashboard, automatisierung)
if re.search(r'\bhome.?assistant\b|homeassistant', low) and not re.search(r'(\blicht\b|\blampe\b|beleuchtung|schalte.*licht|licht.*ein|licht.*aus)', low):
analysis.update(role='ops', task_type='setup', tools=['setup_lookup'], needs_setup_context=True, confidence=0.87)
return analysis
# PERS-PLAN: personal plans (nutrition, fitness, vacation) → personal_assist
if re.search(r'(ernaehrungsplan|ernaehrung.*plan|diaet.*plan|kalorien.*plan|mahlzeiten.*plan|essensplan)', low) or re.search(r'(fitness.*plan|trainingsplan|sport.*plan|lauf.*plan|workout.*plan)', low) or re.search(r'(4.?wochen.*(ernaehrung|abnehmen|sport|training)|wochen.*plan.*(essen|ernaehrung|sport))', low) or re.search(r'(ich will.{0,30}abnehmen|abnehm.*plan|gewicht.*verlieren|gewicht.*reduzier)', low):
analysis.update(role='personal', task_type='assist', tools=['personal_assist'],
needs_memory=True, confidence=0.90)
return analysis
# PERS-WRITE: career writing (Bewerbung, Lebenslauf) → personal_assist
if re.search(r'(bewerbung|lebenslauf|anschreiben|motivationsschreiben|cover.?letter|\bcv\b|resume)', low) or re.search(r'(schreib|erstell).{0,30}(bewerbung|lebenslauf|anschreiben)', low):
analysis.update(role='personal', task_type='assist', tools=['personal_assist'],
needs_memory=True, confidence=0.90)
return analysis
# PERS-TRAVEL: travel planning with specific destination → web_research
if re.search(r'(reiseplan|reiseroute|reisefuehrer|reise.*planen).{0,60}(japan|usa|thailand|europa|italien|spanien|frankreich|portugal|griechenland|tuerkei|bali|dubai|kanada|australien|neuseeland|england|irland|norwegen|schweden|island)', low) or re.search(r'(reise|urlaub|trip).{0,30}(2|3|4|5|6|7|8|9|10|14|21)\s*(wochen|tage|nächte|naechte).{0,60}(japan|usa|thailand|europa)', low) or re.search(r'(hotels?.{0,30}sehenswuerdigkeiten|sehenswuerdigkeiten.{0,30}hotels?).{0,60}(japan|usa|thailand|europa|italien|spanien)', low):
analysis.update(role='personal', task_type='research', tools=['web_research'],
needs_memory=False, confidence=0.90)
return analysis
if re.search(r'(schreibe|erstelle|baue|implementiere|programmiere|deploy)', low) and re.search(r'(tool|app|skript|script|api|programm|code|python|node|bash|kasm)', low):
analysis.update(role='dev', task_type='build', tools=['dev_build'], confidence=0.92)
return analysis
# ARCH: homelab architecture planning, backup-konzept, monitoring-stack, HA
# ARCH-STRATEGY: 'wie gehe ich vor + homelab projekt' -> expert_strategy
if re.search(r'(wie gehe ich vor|was ist der beste ansatz|wie am besten vorgehen|wie.*projekt.*planen|wie.*planen.*projekt)', low) and re.search(r'(homelab.?projekt|neues.*projekt|projekt.*aufbauen|projekt.*strukturier|projekt.*priorisier)', low):
analysis.update(role='research', task_type='strategy', tools=['expert_strategy'], confidence=0.89)
return analysis
if re.search(r'(backup.?(konzept|strategie|plan|lösung|loesung)|backup.*erstell|vollstaendiges backup|komplettes backup)', low) or re.search(r'(monitoring.?stack|prometheus.*grafana|grafana.*prometheus|alertmanager|nagios.*homelab|zabbix.*homelab)', low) or re.search(r'(hochverfuegbarkeit|high.?availab|cluster.*(proxmox|nextcloud|homelab)|proxmox.*cluster|failover.*homelab)', low) or re.search(r'(zweiten?.*(proxmox|node|server).*hinzufueg|neuen?.*(proxmox|node).*hinzufueg|proxmox.*node.*migr)', low) or re.search(r'(homelab.*(aufbauen|planen|konzept|architektur|design|struktur|sicherer|absichern)|wie.*(homelab|server).*(aufbauen|planen))', low) or re.search(r'(backup.*konzept|sicherungskonzept|datensicherung.*konzept|disaster.*recovery)', low):
analysis.update(role='ops', task_type='setup', tools=['setup_lookup'],
needs_setup_context=True, confidence=0.88)
return analysis
# Fix PLAN: day planning and calendar actions → personal_assist
if re.search(r'(tagesplan|tages.*plan|plan.*fuer.*heute|was.*heute.*plane|mein.*tag.*organisier)', low):
analysis.update(role='personal', task_type='howto', tools=['personal_assist'], needs_memory=True, confidence=0.88)
return analysis
if re.search(r'(fuge.*kalender|kalender.*eintrag|termin.*kalender|kalender.*termin|kalender.*hinzufuegen|trage.*ein|fuege.*termin)', low):
analysis.update(role='personal', task_type='write', tools=['personal_assist'], needs_memory=True, confidence=0.88)
return analysis
if re.search(r'(woche|wochenplanung|ueberblick|überblick|dranbleib|struktur)', low) and re.search(r'(ich|meine|mein|plane|planen|organisier|verlier|ueberblick|überblick)', low):
analysis.update(role='personal', task_type='howto', tools=['personal_assist'], needs_memory=True, confidence=0.93)
return analysis
# Fix PROD-1: Produktivität/Wochenplanung mit "wie organisiere ich" Pattern
if re.search(r'(wie|was).{0,10}(organisier|plane).{0,20}(woche|wochen|meine woche|meinen tag)', low) and re.search(r'(pragmatisch|ueberblick|überblick|struktur|dranbleib|verlier|fokus|produktiv)', low):
analysis.update(role='personal', task_type='howto', tools=['personal_assist'], needs_memory=True, confidence=0.94)
return analysis
if re.search(r'(mail|email|e-mail|\bnachricht\b|erinnere|einkauf|einkaufsliste|todo|to-do|packliste|verfasse|verfass|dankesbrief|entschuldig|geburtstagsbrief)', low):
analysis.update(role='personal', task_type='write', tools=['personal_assist'], needs_memory=True, confidence=0.9)
return analysis
# PERSONAL-MESSAGE: short everyday drafts should not escalate to expert_write
if re.search(r'(kurze[nr]?|kurzen|kurze|kleine[nr]?|kleine|knappe[nr]?|knappe|kurz)\s+(entwurf|absage|antwort|nachricht)', low) or \
(re.search(r'(absage|terminverschieb|treffen absagen|termin absagen)', low) and re.search(r'(schreib|entwurf|kurz|kurzen|kurze)', low)):
analysis.update(role='personal', task_type='write', tools=['personal_assist'], needs_memory=True, confidence=0.93)
return analysis
# TOOL-IDEAS: brainstorming for internal tools is not personal assistance
if re.search(r'(idee[n]?|vorschlaege|vorschläge|optionen)', low) and re.search(r'(internes? tool|zugriffspruefung|zugriffsprüfung|internes? werkzeug)', low):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.92)
return analysis
# Fix NAV2: Navigation / Reisefragen -> web_research
if re.search(r'(wie komme ich.*nach|fahrt von.*nach|strecke von.*nach|route.*nach|zug.*nach|bus.*nach|wie lange.*nach|entfernung.*nach|anreise|abfahrt.*nach)', low):
analysis.update(role='research', task_type='explain', tools=['web_research'], needs_web_research=True, confidence=0.87)
return analysis
# Fix N2: News/Nachrichten → web_research
if re.search(r'(nachrichten|news|schlagzeilen|headline|meldungen)', low) and re.search(r'(aktuell|heute|neueste|latest|was sind die|letzte)', low):
analysis.update(role='research', task_type='explain', tools=['web_research'], needs_web_research=True, confidence=0.86)
return analysis
# Fix SHOP: product/hardware recommendations → web_research
if re.search(r'(empfiehl mir|empfehle mir|was ist (ein )?guter|welcher.*ist gut|welche.*ist gut|bestes.*fuer|beste.*fuer|gutes.*fuer|guten.*fuer)', low) and re.search(r'(monitor|festplatte|ssd|hdd|webcam|tastatur|maus|drucker|router|switch|kabel|nas|raspberry|grafikkarte|cpu|prozessor|ram|netzteil|gehaeuse|laptop|notebook|tablet|kopfhoerer|lautsprecher|kamera|mikrofon)', low):
analysis.update(role='research', task_type='compare', tools=['web_research'], needs_web_research=True, confidence=0.84)
return analysis
# Fix W1: Wetter/Forecast → web_research (braucht aktuelle Daten)
if re.search(r'(wetter|temperatur|regen|schnee|sonne|wind|forecast|niederschlag|grad.*celsius|celsius.*grad|wettervorhersage|wie wird.*morgen|wie ist.*wetter)', low):
analysis.update(role='research', task_type='explain', tools=['web_research'], needs_web_research=True, confidence=0.88)
return analysis
# Fix POD: podcast/audiobook discovery → web_research
if re.search(r'\bpodcast', low) or (re.search(r'\bhoerbuecher?\b|\baudiobooks?\b', low) and re.search(r'(finden|empfehle?|kostenlos|beste?|wo.*kann|listen|plattform)', low)):
analysis.update(role='research', task_type='lookup', tools=['web_research'], confidence=0.87)
return analysis
if re.search(r'(aktuell|neueste|heute|latest)', low) and not re.search(r'(rezept|abendessen|mittagessen|fruehstueck|kochen|backen|essen|mahlzeit|gericht)', low) and not any(s in low for s in ('nextcloud', 'immich', 'traefik', 'authentik', 'emby', 'unraid', 'proxmox', 'docker', 'nginx', 'tailscale', 'server', 'ram', 'cpu', 'disk', 'speicher', 'container', 'homelab')) and not re.search(r'\b(schreibe?|verfasse?|erstelle?|formuliere?|entschuldig|dankesbrief|brief|nachricht schreiben)\b', low):
analysis.update(role='research', task_type='explain', tools=['web_research', 'expert_write'], needs_web_research=True, confidence=0.75)
return analysis
# OPS-EXEC: execute commands on hosts
if re.search(r'(fuehre?\s+aus|execute|lauf\s+aus|fuhre?\s+aus)', low) or \
(re.search(r'(teste?\s+ob|teste?\s+den|pruefe?\s+ob|pruefe?\s+den|ueberpruefe?)', low) and any(s in low for s in SERVICE_HINTS)):
analysis.update(role='ops', task_type='action', tools=['ops_exec'], needs_setup_context=True, confidence=0.87)
return analysis
# OPS-INSTALL: install/configure software
if re.search(r'(installiere?|setup\s+\w|richte\s+ein|aufsetzen|deploye?)', low) and \
not re.search(r'(wie\s+(installiere|richte|setze|deploye)|anleitung|erklaer)', low):
analysis.update(role='ops', task_type='action', tools=['ops_install'], needs_setup_context=True, confidence=0.87)
return analysis
# OPS-API: query homelab APIs
if re.search(r'(api\s+(abfrage|query|status)|zeig\s+mir\s+(alle\s+)?(container|entities|services|benutzer))', low) or \
re.search(r'(docker\s+(ps|stats|images)|liste\s+(alle\s+)?(container|images|volumes))', low):
analysis.update(role='ops', task_type='final', tools=['ops_api'], needs_setup_context=True, confidence=0.88)
return analysis
# CAUSAL: warum/wieso/zusammenhang/ursache/seit/nach dem → deep causal analysis
if re.search(r'\b(warum|wieso|weshalb|reason|ursache|zusammenhang|root.?cause)\b', low) and \
SERVICE_HINTS_PATTERN.search(low):
analysis.update(role='ops', task_type='diagnose', tools=['ops_deep_analyze'],
needs_setup_context=True, confidence=0.90)
return analysis
if re.search(r'\b(seit|after|nach dem|nach dem update|nach dem neustart|nach dem stromausfall)\b', low) and \
re.search(r'(nicht mehr|funktioniert nicht|geht nicht|down|offline|error|fehler|kaputt|problem)', low):
analysis.update(role='ops', task_type='diagnose', tools=['ops_deep_analyze'],
needs_setup_context=True, confidence=0.90)
return analysis
if re.search(r'(mehrere services|alle services|viele services|ploetzlich.*nicht|gleichzeitig.*nicht)', low):
analysis.update(role='ops', task_type='diagnose', tools=['ops_deep_analyze'],
needs_setup_context=True, confidence=0.89)
return analysis
# OPS-DEEP: complex multi-service analysis
if re.search(r'(komplette?\s+analyse|tiefe?\s+analyse|vollstaendige?\s+diagnose|ursachenanalyse)', low) or \
(re.search(r'(analysiere?\s+(alle|komplett|alles)|wochenbericht|monatsbericht)', low) and any(s in low for s in SERVICE_HINTS)):
analysis.update(role='ops', task_type='diagnose', tools=['ops_deep_analyze'], needs_setup_context=True, confidence=0.88)
return analysis
# Fix KB-ENTERTAINMENT: Entertainment, Gaming, Filme & Serien für KB_ENTERTAINMENT
# Muss NACH personal/memory Patterns kommen, aber VOR dem Fallback
ENTERTAINMENT_KB_PATTERNS = (
r'\b(elden ring|witcher|fortnite|minecraft|zelda|mario|pokemon)\b',
r'\b(playstation|xbox|switch|steam|epic games|gog|battle\.net)\b',
r'\b(netflix|disney\+|hbo|prime video|apple tv|paramount|peacock)\b',
r'\b(spotify|apple music|tidal|deezer|youtube music)\b',
r'\b(twitch|youtube|streamer|lets play|walkthrough|gameplay)\b',
r'\b(mmorpg|rpg|fps|moba|battle royale|open world|indie game)\b',
r'\b(gpu|grafikkarte|rtx 40|rx 7\d{3}|gaming pc|144hz|240hz)\b',
r'(was.*spielen|spiel.*empfehl|gutes.*spiel|bestes.*spiel|spiel.*tip)',
r'(was.*schauen|film.*empfehl|serie.*empfehl|guter.*film|beste.*serie)',
r'(was.*hören|musik.*empfehl|gute.*musik|playlist.*tip|song.*empfehl)',
r'(binge.*watch|staffel.*empfehl|folge.*empfehl|streaming.*tip)',
)
if any(re.search(p, low) for p in ENTERTAINMENT_KB_PATTERNS):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix KB-HISTORY: Geschichte & Kultur Fragen für KB_HISTORY_CULTURE
# Muss NACH allen Setup/OPS Patterns kommen, aber VOR dem Fallback
HISTORY_KB_PATTERNS = (
r'^(wann war|wer war|was war|wie war|wann lebte|wer lebte|wann starb|wer starb)\b',
r'^(in welchem jahr|in welchem jahrhundert|in welcher epoche)\b',
r'^(erster weltkrieg|zweiter weltkrieg|kalter krieg|französische revolution|industrielle revolution)\b',
r'^(altes ägypten|römisches reich|antikes griechenland|mittelalter|renaissance)\b',
r'^(karl der große|napoleon|luther|hitler|caesar|augustus|cleopatra)\b',
r'^(pyramiden|kolosseum|chinesische mauer|eiffelturm|taj mahal)\b',
r'^(leonardo da vinci|michelangelo|mozart|beethoven|bach|picasso)\b',
r'^(sokrates|platon|aristoteles|philosophie|demokratie)\b',
r'^(religion|christentum|islam|judentum|buddhismus|hinduismus)\b',
r'^(bibel|koran|tora|torah)\b',
r'^(v\.chr\.|n\.chr\.|vor christus|nach christus)\b',
)
if any(re.search(p, low) for p in HISTORY_KB_PATTERNS):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix KB-SMALLTALK: Smalltalk, Gesprächsthemen, Alltagsfragen für KB_SMALLTALK
# Muss VOR dem Fallback kommen, damit Begrüßungen/Motivation nicht ungrounded landen
SMALLTALK_KB_PATTERNS = (
r'^(wie geht|wie gehts|wie geht es dir|was machst du|was gibt es neues|was gibts neues)\b',
r'^(erzähl mir|erzähl was|erzähl einen witz|erzähl mir was|erzähl mir einen witz)\b',
r'^(langweilig|mir ist langweilig|unterhaltung|gespräch|plaudern|quatschen)\b',
r'^(guten morgen|guten abend|gute nacht|hallo|hi|hey)\b',
r'^(danke|vielen dank|danke dir|danke schön|danke schoen)\b',
r'^(motivation|inspiration|aufmunterung|ermutigung|gib mir kraft)\b',
r'^(tipp|ratschlag|empfehlung|was würdest du tun|was wuerdest du tun)\b',
r'^(meinung|was denkst du|wie siehst du das|was sagst du dazu)\b',
r'^(entschuldigung|sorry|verzeihung|entschuldige)\b',
r'^(bis später|bis bald|tschüss|tschuesss|ciao|auf wiedersehen)\b',
)
if any(re.search(p, low) for p in SMALLTALK_KB_PATTERNS):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# Fix KB-TECH-SUPPORT: Technik-Support & DIY Fragen für KB_TECH_SUPPORT.md
# WICHTIG: Muss VOR dem Fallback kommen, damit Tech-Probleme nicht ungrounded landen
TECH_SUPPORT_PATTERNS = (
r'\b(smartphone|handy|iphone|android|tablet|ipad)\b.*\b(lädt nicht|lädt nicht|akku|bildschirm|display|kaputt|defekt|problem)\b',
r'\b(laptop|computer|pc|windows|mac|macbook)\b.*\b(langsam|hängt|hängt|friert|bluescreen|neustart|startet nicht)\b',
r'\b(wlan|wifi|internet|router)\b.*\b(funktioniert nicht|geht nicht|langsam|verbindung|problem)\b',
r'\b(drucker|scanner)\b.*\b(funktioniert nicht|geht nicht|verbindet|problem|fehler)\b',
r'\b(fernseher|tv|fernbedienung|streaming|netflix)\b.*\b(funktioniert nicht|geht nicht|bild|ton|problem)\b',
r'\b(usb|hdmi|kabel|anschluss)\b.*\b(funktioniert nicht|geht nicht|erkannt|problem)\b',
r'\b(virus|malware|phishing|sicherheit|passwort vergessen)\b',
r'\b(app|programm|software)\b.*\b(stürzt|ab|hängt|startet nicht|fehler)\b',
r'\b(datei|ordner)\b.*\b(löschen|löschen|nicht öffnen|nicht öffnen|zugriff)\b',
r'\b(windows|update|aktualisieren)\b.*\b(hängt|hängt|fehler|problem)\b',
r'\b(waschmaschine|staubsauger|kühlschrank)\b.*\b(funktioniert nicht|geht nicht|problem|fehler|laut)\b',
r'\b(diy|reparieren|fixen|hilfe|support)\b.*\b(technik|gerät|problem)\b',
)
if any(re.search(p, low) for p in TECH_SUPPORT_PATTERNS) and not any(s in low for s in ('homelab', 'server', 'docker', 'container', 'service', 'host', 'setup', 'infrastruktur', 'nextcloud', 'traefik', 'proxmox', 'unraid')):
analysis.update(role='general', task_type='explain', tools=['general_answer'], needs_memory=False, confidence=0.95)
return analysis
# ── IMPLICIT DOMAIN ROUTING (Thema, nicht Schluesselwort) ──────────────────
# Kein "schreib einen Artikel" noetig — das Thema bestimmt den Skill.
# IMPLICIT-CODE: Tech-Konzepte erklaeren / vergleichen / implementieren
_tech_topics = (
r'(microservice|kubernetes|k8s|docker.compose|terraform|ansible|'
r'rest.api|graphql|grpc|websocket|message.queue|rabbitmq|kafka|'
r'sql.*nosql|nosql.*sql|relational.*datenbank|datenbank.*design|'
r'design.pattern|entwurfsmuster|solid.*prinzip|dependency.*injection|'
r'event.driven|cqrs|saga.*pattern|clean.*architektur|hexagonal|'
r'ci.*cd.*pipeline|devops.*praxis|git.*workflow|branching.*modell|'
r'algorithmus|komplexitaet|big.o|sortier|graphen.*algorithmus|'
r'machine.*learning.*impl|neural.*network.*bau|embedding.*erstell|'
r'api.*rate.limit|auth.*token.*flow|oauth.*impl|jwt.*erstell)'
)
_explain_verbs = r'(erklaer|erklaere|was ist|wie funk|unterschied|vergleich|vor.*nach|wann.*nutzen|wann.*verwenden|wie.*besser|welches.*besser|empfehlung|wie.*impl|wie.*umsetz|zeig.*wie|beispiel.*fuer)'
if re.search(_tech_topics, low) and re.search(_explain_verbs, low) and not any(s in low for s in SERVICE_HINTS):
analysis.update(role='dev', task_type='explain', tools=['qdrant_search', 'expert_code'], confidence=0.87)
return analysis
# IMPLICIT-WRITE: Kommunikationsaufgaben ohne "schreib" Keyword
_comm_types = r'(\bemail\b|\bbrief\b|anschreiben|bewerbung|motivationsschreiben|\bpitch\b|proposal|\bpraesentation\b|\bpresentation\b|executive.*summary|zusammenfassung.*fuer|bericht.*fuer|protokoll.*erstell|minutes.*of|meeting.*notes|stellungnahme|offizielle.*mitteilung|pressemitteilung|newsletter|\bslogan\b|produktbeschreibung|landing.*page.*text)'
_comm_triggers = r'(kannst du|hilf mir|erstelle|formulier|schreib|verfass|mach mir|ich brauche|benotige|erstell mir|bitte.*erstell|wie.*formulier|wie.*sag ich|gib mir.*entwurf)'
if re.search(_comm_types, low) and re.search(_comm_triggers, low) and not any(s in low for s in ('homelab', 'server', 'docker', 'code', 'skript')):
analysis.update(role='research', task_type='write', tools=['expert_write'], confidence=0.89)
return analysis
# IMPLICIT-SYNTH: Komplexe Synthesis-Fragen → Web-Recherche + Expert-Schreiben
_synth_patterns = r'(was.*beste.*2\d{3}|welche.*empfehlung.*2\d{3}|wie ist der stand|aktueller stand von|neueste.*entwicklung|wohin.*entwickelt sich|wie.*zukunft|state of the art|was denken experten|expertenmeinung zu|marktfuehrer|markt.*analyse|wettbewerbsvergleich|vor.*nachteile.*von.*und.*von)'
if re.search(_synth_patterns, low) and not any(s in low for s in SERVICE_HINTS):
analysis.update(role='research', task_type='explain', tools=['web_research', 'expert_write'], needs_web_research=True, confidence=0.86)
return analysis
# IMPLICIT-STRATEGY: Planungs- und Vorgehen-Fragen ohne explizites "OKR/SWOT"
_plan_patterns = r'(wie.*aufbauen|wie gehe ich vor|schritt.*fuer.*schritt.*vorgehen|wie.*skalier|best.*approach.*fuer|bestes.*vorgehen|wie.*strukturier|wie.*priorisier|was.*zuerst|in welcher reihenfolge.*vorgehen|langfristig.*planen|strategie.*entwickeln|roadmap.*fuer|plan.*fuer.*naechsten|wie.*team.*aufstell|ressourcen.*einsetzen)'
if re.search(_plan_patterns, low) and not any(s in low for s in ('kasm', 'emby', 'sonos', 'lidarr', 'radarr', 'sonarr', 'traefik', 'nextcloud', 'immich', 'authentik', 'ldap')) and not re.search(r'(diagnos|kaputt|fehler|defekt|nicht.*laueft|nicht.*laeuft|nicht.*erreichbar)', low):
analysis.update(role='research', task_type='build', tools=['expert_strategy'], confidence=0.85)
return analysis
# IMPLICIT-EDIT: Text liegt vor und soll verbessert werden (ohne "lektoriere")
# Erkennung: langer Text + Verbesserungswunsch
_edit_triggers = r'(klingt.*gut|klingt das gut|ist das gut formuliert|passt das so|kannst du.*verbessern|kannst du.*anpassen|kannst du.*ueberarbeiten|bitte.*verbessern|bitte.*anpassen|mach.*besser|mach.*professioneller|mach.*klarer|zu lang|zu kurz|zu foermlich|zu informell|andere formulierung|anders formulieren)'
if re.search(_edit_triggers, low) and len(low) > 20 and not any(s in low for s in ('homelab', 'server', 'docker', 'code', 'skript')):
analysis.update(role='general', task_type='write', tools=['expert_edit'], confidence=0.87)
return analysis
# ── EXPERT DOMAINS ────────────────────────────────────────────────────────
# EXPERT-WRITE: Artikel, Blog, Journalismus, kreatives Schreiben, Storytelling
if re.search(r'(schreib.{0,20}(artikel|blog|essay|bericht|kolumne|reportage|feature|story|geschichte|novelle|kurzgeschichte|prolog|epilog|gedicht|rede|speech|pitch|biographie|profil|portrait|caption|post|thread)|'
r'verfasse.{0,20}(artikel|text|bericht|email|brief|nachricht|anschreiben|proposal|pitch|stellungnahme)|'
r'erstell.{0,20}(artikel|blog.*post|newsletter|pressemitteilung|content|copy|slogan|headline|landing.*page|produktbeschreibung)|'
r'schreib.{0,20}(mir|uns|einen|eine|guten|spannenden|interessanten|professionellen|kreativen|packenden|ueberzeugenden)|'
r'ich brauche.{0,30}(artikel|text|story|email|anschreiben|pressemitteilung|newsletter|pitch|proposal)|'
r'ghostwrite|narrativ|storytelling|lede|inverted pyramid|hook.*fuer.*text|teaser.*fuer)', low) and not any(s in low for s in ('homelab', 'docker', 'container', 'code', 'skript', 'script', 'programm', 'funktion', 'klasse')):
analysis.update(role='research', task_type='write', tools=['expert_write'], confidence=0.92)
return analysis
# EXPERT-CODE: Code-Architektur, Code-Review, Senior-Dev-Aufgaben
if re.search(r'(code.*review|review.*code|code.*analysier|analysier.*code|'
r'architektur.*entscheiden|technische.*schulden|tech.*debt|refactor|refaktorier|'
r'design.*pattern|entwurfsmuster|solid.*prinzip|clean.*code|clean.*architektur|'
r'performance.*optimier|bottleneck.*find|profil.*code|security.*audit.*code|'
r'algorithm.*optimier|komplexitaet.*reduzier|o.*notation|big.*o|'
r'api.*design|api.*architektur|microservice.*design|monolith.*split|'
r'test.*strategie|unit.*test.*schreib|integration.*test|tdd|bdd|'
r'was.*best.*practice|beste.*praxis.*code|code.*qualitaet|'
r'implementier.*fuer.*mich|schreib.*funktion|schreib.*klasse|schreib.*script|schreib.*skript|'
r'bau.*mir.*eine.*app|erstell.*mir.*ein.*programm|erstell.*mir.*eine.*api|'
r'wie.*implementier.*ich|wie.*programmier.*ich|wie.*baue.*ich|wie.*erstell.*ich.*code|'
r'debug.*diesen|fehler.*in.*meinem.*code|warum.*gibt.*mein.*code|was.*falsch.*code)', low):
analysis.update(role='dev', task_type='build', tools=['expert_code'], confidence=0.91)
return analysis
# EXPERT-RESEARCH: Tiefe Recherche, Marktanalyse, investigativ
if re.search(r'(recherchiere.{0,30}(tiefgr|tiefsinnig|vollst|komplett|umfassend|detailliert)|'
r'marktanalyse|wettbewerbsanalyse|kompetitor.*analyse|competitive.*analysis|'
r'investigier|investigative|vollstaendige.*analyse|comprehensive.*analysis|'
r'fact.?check|fakten.*pruefen|quellen.*pruefen|verifizier.*fakten|'
r'recherche.*bericht|research.*report|due.*diligence|'
r'was.*sagen.*experten|expertenmeinung|state.*of.*art|aktueller.*stand)', low):
analysis.update(role='research', task_type='explain', tools=['expert_research'], needs_web_research=True, confidence=0.90)
return analysis
# EXPERT-EDIT: Lektorat, Korrekturlesen, Stilverbesserung
if re.search(r'(korrigier.{0,20}(text|folgenden|meinen|diesen|den)|'
r'lektorier|lektüre|proofreading|proofread|'
r'verbessere.{0,20}(text|meinen text|folgenden text|diesen text|schreibstil)|'
r'ueberarbeite.{0,20}(text|meinen|diesen|folgenden)|'
r'überarbeite.{0,20}(text|meinen|diesen|folgenden)|'
r'rechtschreibung.*pruefen|grammatik.*pruefen|pruef.*grammatik|pruef.*rechtschreibung|'
r'stilistisch.*verbessern|stilverbesserung|besser.*formulier|umformulier|'
r'klingt.*besser|klingt.*natuerlicher|klingt.*professioneller|'
r'feedback.*text|feedback.*meinem text|text.*feedback)', low) and not any(s in low for s in ('docker', 'code', 'skript')):
analysis.update(role='general', task_type='write', tools=['expert_edit'], confidence=0.92)
return analysis
# EXPERT-STRATEGY: Strategie, Business-Planung, Management
if re.search(r'(businessplan|business.*plan|business.*strategie|go.*to.*market|gtm.*strategie|startup.*plan|gruendungsplan|'
r'okr.*(erstell|roadmap|ziel)|erstell.*okr|erstell.*roadmap|'
r'ziele.*definier|quartalsziele|jahresziele|'
r'roadmap.*(erstell|plan)|produkt.*roadmap|feature.*priorisier|'
r'swot.*analyse|swot.*erstell|staerken.*schwaechen|'
r'entscheidungs.*matrix|priorisierungs.*matrix|rice.*scoring|'
r'strategische.*planung|langfristige.*planung|wachstums.*strategie|'
r'project.*charter|projektcharter|projektplanung|meilenstein.*plan|'
r'ressourcen.*planung|kapazitaets.*planung|team.*aufbau|'
r'wie.*skalier|skalierungs.*strategie|pivot.*entscheid)', low) and not any(s in low for s in ('docker', 'container')):
analysis.update(role='research', task_type='build', tools=['expert_strategy'], confidence=0.91)
return analysis
if role_hint == 'ops':
analysis.update(role='ops', task_type='diagnose', tools=['setup_lookup'], needs_setup_context=True, confidence=0.66)
elif role_hint == 'research':
analysis.update(role='research', task_type='explain', tools=['web_research'], needs_web_research=True, confidence=0.66)
elif role_hint == 'personal':
analysis.update(role='personal', task_type='write', tools=['personal_assist'], needs_memory=True, confidence=0.66)
elif role_hint == 'dev':
analysis.update(role='dev', task_type='build', tools=['dev_build'], confidence=0.66)
else:
analysis.update(role='general', task_type='howto', tools=['general_answer'], needs_memory=False, confidence=0.6)
return analysis
def normalize_analysis(data, message: str, role_hint: str | None):
if not isinstance(data, dict):
data = heuristic_analysis(message, role_hint)
out = heuristic_analysis(message, role_hint)
# Fix RECIPE-ROUTING-13: Wenn Heuristik hohe Confidence hat (Fast-Path), behalte ihre Tools
# Heuristik-Fast-Paths (z.B. Rezepte) haben confidence >= 0.9, LLM hat confidence <= 0.8
if out.get('confidence', 0) >= 0.9:
# Heuristik-Fast-Path: behalte Tools, erlaube nur reasoning_focus vom LLM
out.update({k: v for k, v in data.items() if k in {'reasoning_focus'}})
else:
out.update({k: v for k, v in data.items() if k in out or k in {'reasoning_focus', 'tools', 'confidence'}})
role = str(out.get('role') or (role_hint or 'general')).lower()
if role not in FINAL_ROLE_MAP:
role = role_hint if role_hint in FINAL_ROLE_MAP else 'general'
out['role'] = role
tools = out.get('tools') or []
clean_tools = []
for t in tools:
name = str(t).strip()
if name in TOOLS and name not in clean_tools:
clean_tools.append(name)
out['tools'] = clean_tools[:3]
out['needs_setup_context'] = bool(out.get('needs_setup_context'))
out['needs_web_research'] = bool(out.get('needs_web_research'))
out['needs_memory'] = bool(out.get('needs_memory'))
out['dangerous_action'] = bool(out.get('dangerous_action'))
out['force_sequential'] = bool(out.get('force_sequential'))
if not out['tools']:
out['tools'] = heuristic_analysis(message, role_hint)['tools']
if out['needs_setup_context'] and 'setup_lookup' not in out['tools'] and out['role'] == 'ops' and not out['dangerous_action']:
existing_kinds = {TOOLS[t]['kind'] for t in out['tools'] if t in TOOLS}
if existing_kinds <= {'evidence'}:
out['tools'] = (['setup_lookup'] + out['tools'])[:3]
if out['needs_web_research'] and not any(t in out['tools'] for t in ['web_research', 'url_research', 'compare_selfhosted', 'security_plan']):
out['tools'] = (['web_research'] + out['tools'])[:3]
family_candidates = collect_intent_families(message, SERVICE_HINTS) if USE_INTENT_FAMILIES else []
family_info = family_candidates[0] if family_candidates else classify_intent_family(message, SERVICE_HINTS)
before_tools = list(out.get('tools') or [])
out = apply_intent_normalizer(message, out, family_info=family_info)
if family_candidates:
out['family_candidates'] = [item.get('family') for item in family_candidates if item.get('family')]
composition = compose_family_plan(message, out, family_candidates) if USE_INTENT_COMPOSITION else {'composed': False}
out, before_comp, logged_composition = _apply_composition_selection(
out,
composition,
family_candidates,
choose_plan=choose_plan,
use_policy=USE_COMPOSITION_POLICY,
)
if before_comp and logged_composition:
log_intent_composition(message, family_candidates, before_comp, out, logged_composition)
log_intent_family_shadow(message, family_info, before_tools, list(out.get('tools') or []))
if USE_META_CONTROLLER:
_log_shadow_decision(
META_CONTROLLER_LOG,
_shadow_decision(message, out, family_candidates, TOOLS),
)
return out
def apply_intent_normalizer(message: str, analysis: dict, family_info: dict | None = None):
return _normalize_intent_family(
message,
analysis,
family_info,
classify_intent_family=classify_intent_family,
service_hints=SERVICE_HINTS,
)
def log_intent_family_shadow(message: str, family_info: dict, before_tools: list[str], after_tools: list[str]):
_log_intent_family_shadow(
message,
family_info,
before_tools,
after_tools,
log_path=INTENT_FAMILY_LOG,
collect_intent_families=collect_intent_families,
service_hints=SERVICE_HINTS,
)
def log_intent_composition(message: str, family_candidates: list[dict], analysis_before: dict, analysis_after: dict, composition: dict):
_log_intent_composition(
message,
family_candidates,
analysis_before,
analysis_after,
composition,
log_path=INTENT_COMPOSITION_LOG,
)
def write_temp(prefix: str, text: str):
tmp = tempfile.NamedTemporaryFile('w', suffix=f'.{prefix}.txt', delete=False, encoding='utf-8')
tmp.write(text)
tmp.flush()
tmp.close()
return tmp.name
def load_profile_entries_local():
if not PROFILE_STORE.exists():
return []
items = []
try:
lines = PROFILE_STORE.read_text(encoding='utf-8', errors='ignore').splitlines()
except Exception:
return []
for line in lines:
line = line.strip()
if not line:
continue
try:
data = json.loads(line)
except Exception:
continue
if isinstance(data, dict) and data.get('memory_type') == 'profile':
items.append(data)
return items
def profile_memory_context(query: str, limit: int = 4):
entries = load_profile_entries_local()
if not entries:
return ''
qnorm = norm(query)
qterms = topic_terms(query)
scored = []
for entry in entries:
text = str(entry.get('text') or '').strip()
if not text:
continue
en = norm(text)
score = float(entry.get('importance', 0.5))
if qnorm and qnorm in en:
score += 5.0
for term in qterms:
if term in en:
score += 1.7
cat = str(entry.get('category') or '')
if cat == 'constraint':
score += 0.7
if cat == 'setup' and any(h in qnorm for h in SERVICE_HINTS):
score += 1.2
if cat == 'preference' and any(k in qnorm for k in ['antwort', 'stil', 'kurz', 'direkt', 'mag', 'bevorzug', 'gerne']):
score += 1.0
if score >= 1.45:
scored.append((score, entry))
if not scored:
fallback = []
for entry in reversed(entries):
cat = str(entry.get('category') or '')
if cat in {'constraint', 'preference'}:
fallback.append(entry)
if len(fallback) >= min(limit, 2):
break
if not fallback:
return ''
picked = list(reversed(fallback))
else:
picked = []
seen = set()
for _score, entry in sorted(scored, key=lambda x: (x[0], x[1].get('last_seen_at', x[1].get('created_at', ''))), reverse=True):
key = (entry.get('category', 'fact'), norm(str(entry.get('text') or '')))
if key in seen:
continue
seen.add(key)
picked.append(entry)
if len(picked) >= limit:
break
lines = ['Persoenlicher Kontext:']
for entry in picked:
lines.append(f"- [{entry.get('category', 'fact')}|profile] {str(entry.get('text') or '').strip()}")
return '\n'.join(lines) if len(lines) > 1 else ''
def memory_context(message: str, analysis: dict | None = None):
if not analysis or not AGENT_MEMORY.exists():
return ''
role = str(analysis.get('role') or '')
task_type = str(analysis.get('task_type') or '')
needs_memory = bool(analysis.get('needs_memory')) or task_type == 'memory'
if not needs_memory:
return ''
if task_type == 'memory':
rc, out, _err = run([str(AGENT_MEMORY), 'context', message], timeout=10)
if rc != 0 or not out or '(kein persoenlicher Kontext gespeichert)' in out:
return ''
return out
base = ''
if role in {'personal', 'general', 'ops'}:
base = profile_memory_context(message)
# Append goals + contacts context for personal/research roles or planning task types
if role in {'personal', 'research'} or task_type in {'strategy', 'write', 'assist', 'plan'}:
extra_parts = []
goals_bin = ROOT / 'bin' / 'goals-manage'
contacts_bin = ROOT / 'bin' / 'contacts-manage'
if goals_bin.exists():
rc_g, out_g, _ = run([str(goals_bin), 'context'], timeout=10)
if rc_g == 0 and out_g and '(Noch keine' not in out_g and '(Keine' not in out_g:
extra_parts.append(out_g)
if contacts_bin.exists():
rc_c, out_c, _ = run([str(contacts_bin), 'show-due', 'days=3'], timeout=10)
if rc_c == 0 and out_c and 'Keine Follow-ups' not in out_c:
extra_parts.append('Follow-ups (3 Tage):\n' + out_c)
rc_b, out_b, _ = run([str(contacts_bin), 'show-birthdays', 'days=7'], timeout=10)
if rc_b == 0 and out_b and 'Keine Geburtstage' not in out_b:
extra_parts.append('Geburtstage (7 Tage):\n' + out_b)
if extra_parts:
extras = '\n\n'.join(extra_parts)
base = (base + '\n\n' + extras).strip() if base else extras
return base
def remember_user_signal(message: str):
if not AGENT_MEMORY.exists():
return
try:
run([str(AGENT_MEMORY), 'ingest', message], timeout=8)
except Exception:
pass
def remember_episode(message: str):
if not AGENT_MEMORY.exists():
return
try:
run([str(AGENT_MEMORY), 'episode', message], timeout=8)
except Exception:
pass
def remember_incident(text: str):
if not AGENT_MEMORY.exists():
return
try:
run([str(AGENT_MEMORY), 'incident', text], timeout=10)
except Exception:
pass
def should_fetch_retrieval(message: str, analysis: dict | None = None, role_hint: str | None = None) -> bool:
low = norm(message)
if analysis:
role = analysis.get('role')
needs_setup = bool(analysis.get('needs_setup_context'))
needs_web = bool(analysis.get('needs_web_research'))
task_type = analysis.get('task_type')
if role == 'research' and needs_web and not needs_setup:
return False
if role in {'general', 'personal'} and not needs_setup and not any(h in low for h in SERVICE_HINTS):
return False
if role == 'research' and task_type == 'compare' and not needs_setup and not any(h in low for h in SERVICE_HINTS):
return False
if role_hint in {'ops', 'dev'}:
return True
if analysis and (analysis.get('needs_setup_context') or analysis.get('role') in {'ops', 'dev'}):
return True
if any(h in low for h in SERVICE_HINTS):
return True
if re.search(r'\b(set?up|host|service|dienste|logs?|share|smb|bibliothek|container|agent|bibabot|gedaechtnis|memory|router|subagent)\b', low):
return True
return False
def retrieval_context(message: str, analysis: dict | None = None, role_hint: str | None = None):
if not CONTEXT_PACK.exists() or not should_fetch_retrieval(message, analysis, role_hint):
return ''
queries = [message]
if is_followup_query(message):
topic = recent_topic_event(message) or load_topic_state()
if topic:
root = str(topic.get('root_message') or topic.get('message') or '').strip()
if root and norm(root) != norm(message):
queries.append(f'{root}\n{message}')
if analysis:
focus = ' '.join(str(x).strip() for x in analysis.get('reasoning_focus', [])[:2] if str(x).strip())
if focus:
queries.append(f'{message}\n{focus}')
low = norm(message)
hinted = [term for term in SERVICE_HINTS if term in low]
if hinted:
queries.append(' '.join(dict.fromkeys(hinted + [message])))
lines = []
seen = set()
for query in queries[:3]:
rc, out, _err = run([str(CONTEXT_PACK), query], timeout=35)
if rc != 0 or not out:
continue
for line in out.splitlines():
line = line.strip()
if not line or line == 'Kontext-Snippets:' or line in seen:
continue
seen.add(line)
lines.append(line)
if len(lines) >= 8:
return 'Kontext-Snippets:\n' + '\n'.join(lines)
return 'Kontext-Snippets:\n' + '\n'.join(lines) if lines else ''
def append_context(base: str, extra: str) -> str:
extra = (extra or '').strip()
base = (base or '').strip()
if not extra:
return base
if not base:
return extra
if extra in base:
return base
return base + '\n\n' + extra
def playbook_context(message: str, analysis: dict | None = None) -> str:
if not USE_PLAYBOOKS:
return ''
try:
return format_playbook_context(message, analysis=analysis)
except Exception:
return ''
def apply_plan_first(analysis: dict, plan: dict | None) -> dict:
if not plan:
return analysis
out = dict(analysis or {})
tool_override = [str(t).strip() for t in plan.get('tool_override') or [] if str(t).strip() in TOOLS]
if tool_override:
out['tools'] = tool_override[:3]
focus = []
for item in (out.get('reasoning_focus') or []) + (plan.get('reasoning_focus') or []):
text = str(item).strip()
if text and text not in focus:
focus.append(text)
if focus:
out['reasoning_focus'] = focus[:6]
gaps = set(plan.get('evidence_gaps') or [])
if 'setup_or_inventory' in gaps:
out['needs_setup_context'] = True
if 'personal_memory' in gaps:
out['needs_memory'] = True
out['plan_first_mode'] = plan.get('mode', 'plan_first')
return out
def response_contract(analysis: dict):
task_type = analysis.get('task_type', 'explain')
if task_type == 'diagnose':
return 'Antwortvertrag: Erklaere kurz den relevanten Mechanismus, nenne die wahrscheinlichste Ursache, belege sie mit den vorhandenen Fakten und gib die naechsten sinnvollen Checks oder Fixes an.'
if task_type == 'compare':
return 'Antwortvertrag: Gib eine klare Empfehlung, vergleiche die besten Optionen mit konkreten Tradeoffs und begruende, warum die Empfehlung zur Situation passt.'
if task_type == 'howto':
return 'Antwortvertrag: Gib konkrete, brauchbare Schritte statt Meta-Erklaerungen. Nenne Voraussetzungen und typische Stolpersteine kurz.'
if task_type == 'build':
return 'Antwortvertrag: Zerlege das Vorhaben in sinnvolle Schritte, nenne den vorgesehenen Output und sage offen, was bereits verifiziert ist.'
if task_type == 'status':
return 'Antwortvertrag: Nenne zuerst den aktuellen Zustand und danach nur die Punkte, die wirklich Aufmerksamkeit brauchen.'
if task_type == 'action':
return 'Antwortvertrag: Sage klar, was getan wurde oder was blockiert war, und nenne den verifizierten Effekt.'
if task_type == 'memory':
return 'Antwortvertrag: Antworte persoenlich, konkret und nur mit wirklich gespeichertem Kontext.'
return 'Antwortvertrag: Antworte direkt auf die eigentliche Frage, dann kurz die wichtigsten Begruendungen oder Beispiele.'
def postprocess_final_answer(message: str, analysis: dict, final_text: str) -> str:
return _policy_postprocess_final_answer(message, analysis, final_text)
def compact_episode_text(message: str, analysis: dict, final_text: str, evidence_items: list[dict]) -> str:
lines = []
for line in (final_text or '').splitlines():
line = line.strip()
if not line or line.lower() in {'befund:', 'ursache:', 'wahrscheinlichste ursache:', 'naechste pruefungen/aktionen:', 'naechster schritt:', 'aktion:'}:
continue
lines.append(line)
if len(lines) >= 3:
break
if not lines and evidence_items:
lines = [item.get('summary', '') for item in evidence_items if item.get('summary')]
snippet = ' | '.join(line for line in lines if line)[:700]
return f'{message}\nAntwortkern: {snippet}' if snippet else message
def evidence_object(tool: str, rc: int, output: str, err: str = ''):
# Fix GENERAL-GROUNDEDNESS-5: rc kann None sein, was alle Heuristiken deaktiviert
# Normalisiere rc zu int (0 = success, 1 = failure)
if rc is None:
rc = 0 if (output or '').strip() else 1
clean = (output or '').strip()
# Fix EVIDENCE-STDERR-1: Wenn stdout leer ist, aber stderr substanzielle Antwort hat, nutze stderr
# Problem: Tools wie research-web-answer geben bei Fehlern 'nicht_verifiziert' auf stderr aus
# Lösung: Fallback auf stderr wenn stdout leer und stderr nicht nur Fehlermeldung ist
if not clean and err and len(err.strip()) > 20 and 'nicht_verifiziert' not in err.lower():
clean = err.strip()
low = norm(clean)
kind = EVIDENCE_KIND_BY_TOOL.get(tool, 'general')
grounded = False
if tool in GROUNDED_FASTPATH_TOOLS:
grounded = bool(clean)
elif kind == 'web':
# Fix WEB-GROUNDEDNESS-7: Entschärfte Heuristik für web_research
# Problem: Zu strenge Kombination von Bedingungen hat 32% (12/37) als ungrounded markiert
# Lösung: Einfachere, robustere Prüfung - akzeptiere mehr Antworten als grounded
# Basis: Kein Fehler und substanzieller Output
# Fix WEB-GROUNDEDNESS-8: Erkenne nicht_verifiziert überall im String, nicht nur am Anfang
has_valid_output = bool(clean) and rc == 0 and not clean.startswith('(fehler:') and 'nicht_verifiziert' not in clean.lower()
# Explizite Marker (Kurzantwort/Quellen Format von research-web-answer)
# Fix WEB-GROUNDEDNESS-8: Prüfe ob nach Marker auch tatsächlich Inhalt kommt
has_quellen = 'Quellen:' in clean and len(clean.split('Quellen:')[-1].strip()) > 10
# Kurzantwort muss gefolgt sein von substanziellem Inhalt (nicht nur Formatierung)
if 'Kurzantwort:' in clean:
after_marker = clean.split('Kurzantwort:')[-1].strip()
# Entferne Markdown-Formatierung für die Prüfung
after_clean = after_marker.replace('**', '').replace('*', '').replace('- ', '').strip()
has_kurzantwort = len(after_clean) > 20 and '.' in after_clean
else:
has_kurzantwort = False
has_explicit_markers = has_quellen or has_kurzantwort
# Substanzieller Content (ENTSCHÄRFT: von 150 auf 100 Zeichen, von 2 auf 1 Sätze)
has_substantial_content = (
has_valid_output and len(clean) > 100 and clean.count('.') >= 1
)
# Web-typische Strukturmerkmale (erweitert)
web_markers = ['http', 'www.', '.de', '.com', '.org', ' - ', '•', '→', '1.', '2.', '3.', '- ', '**', '*']
has_web_structure = any(m in clean.lower() for m in web_markers)
# Markdown-Struktur als Qualitätsindikator
has_markdown_structure = has_valid_output and any(c in clean for c in ['**', '*', '- ', '1.', '2.', '###', '##'])
# Mehrzeilige Antworten sind wahrscheinlich gut
has_multiple_lines = has_valid_output and len(clean.splitlines()) >= 2
# Kombinierte Entscheidung (ENTSCHÄRFT: OR statt AND-Kombinationen)
grounded = (
has_explicit_markers or
(has_substantial_content and (has_web_structure or has_markdown_structure)) or
(has_valid_output and len(clean) > 200 and has_multiple_lines)
)
elif kind in {'memory', 'status', 'action'}:
grounded = bool(clean)
elif kind == 'general':
# Fix PERSONAL-GROUNDEDNESS-1: Spezielle Heuristik für personal_assist
# Problem: personal_assist Antworten (Rezepte, Wochenplanung, etc.) wurden als ungrounded markiert
# Lösung: Eigene, weniger strenge Heuristik für personal_assist
if tool == 'personal_assist':
# personal_assist gibt strukturierte Antworten (Rezepte, Listen, How-tos)
# Akzeptiere wenn: Output existiert, kein Fehler, und substanziell (>50 Zeichen)
has_valid_output = bool(clean) and rc == 0 and not clean.startswith('(fehler:') and not clean.startswith('nicht_verifiziert')
# Persönliche Assistenz-Marker (Rezepte, Listen, How-tos)
# Fix PERS-GROUNDED-2: Erweiterte Marker für Abendessen-Ideen und Essens-Themen
personal_markers = [
'zutaten', 'zubereitung', 'zutatenliste', # Rezepte
'schritt', 'schritte', 'anleitung', 'vorgehen', # How-tos
'kurzbeschreibung', 'beschreibung', # Allgemein
'portione', 'portionen', # Rezepte
'tipps', 'hinweise', 'beispiel', # Listen/Tipps
'1.', '2.', '3.', '- ', # Aufzählungen
'**', '*', # Markdown-Formatierung
'hier sind', 'wichtige punkte', 'punkte', # Einleitungen
'optionen/hinweise', 'optionen', 'hinweise', # Rezept-Ende
'einfaches', 'grundrezept', # Rezept-Einleitung
'fuer', 'portionen', # Rezept-Header
'wochenplan', 'wochenplanung', 'pragmatisch', # Wochenplanung
'prioritaeten', 'prioritäten', 'kalenderblock', # Planung
'brain-dump', 'tagesabschluss', 'review', # Produktivität
# Fix PERS-GROUNDED-2: Neue Marker für Abendessen-Ideen
'abendessen', 'abend-essen', 'ideen', 'schnell', 'einfach', # Essens-Ideen
'one-pan', 'one pan', 'pasta', 'bowl', 'wrap', 'sandwich', # Gerichtstypen
'protein', 'gemuese', 'gemüse', 'fleisch', 'fisch', 'tofu', # Zutaten-Kategorien
'gerichte', 'variationen', 'klassiker', 'optionen', # Listen-Typen
'meal-prep', 'meal prep', 'vorbereiten', 'vorkochen', # Planung
'min', 'minute', 'minuten', 'stunde', 'zeit', # Zeitangaben
]
has_personal_structure = has_valid_output and any(m in low for m in personal_markers)
# Substantielle Antworten sind grounded (ENTSCHÄRFT für personal)
is_substantial = (
has_valid_output and
(len(clean) > 50 and clean.count('.') >= 1) or
(len(clean) > 30 and has_personal_structure) or
(len(clean.splitlines()) >= 2 and has_personal_structure)
)
# Mehrzeilige Antworten mit Struktur sind grounded
has_multiple_lines = has_valid_output and len(clean.splitlines()) >= 2
has_markdown = has_valid_output and any(c in clean for c in ['**', '*', '- ', '1.', '2.', '###', '##'])
# Kombinierte Entscheidung für personal_assist
grounded = (
has_personal_structure or
is_substantial or
(has_multiple_lines and len(clean) > 40) or
(has_markdown and len(clean) > 40)
)
# Fix PERS-GROUNDED-3: Entferne early return - Summary wird jetzt unten erstellt
# Der Code fällt durch zu den gemeinsamen Summary/Return-Blöcken am Ende der Funktion
# Fix GENERAL-GROUNDEDNESS-6: Entschärfte Heuristik für general_answer
# Problem: Zu strenge Prüfung hat viele valide Antworten als ungrounded markiert
# Lösung: Akzeptiere mehr Antworten als grounded, besonders wenn sie aus general-local-assist kommen
# Basis: Output existiert und ist kein Fehler
has_valid_output = bool(clean) and rc == 0 and not clean.startswith('(fehler:') and not clean.startswith('nicht_verifiziert')
# Heuristik 1: Strukturierte Inhalte (Rezepte, How-tos, Erklärungen)
# Fix GENERAL-GROUNDEDNESS-7: Erweiterte Marker für general-local-assist Output-Format
structure_markers = [
'zutaten', 'zubereitung', 'zutatenliste', # Rezepte
'schritt', 'schritte', 'anleitung', 'vorgehen', # How-tos
'kurzbeschreibung', 'beschreibung', # Allgemein
'portione', 'portionen', # Rezepte
'vorteile', 'nachteile', 'fazit', # Vergleiche
'erklärung', 'erklaerung', 'warum', 'wie', # Erklärungen
'tipps', 'hinweise', 'beispiel', # Listen/Tipps
'1.', '2.', '3.', '- ', # Aufzählungen
'**', '*', # Markdown-Formatierung
'hier sind', 'wichtige punkte', 'punkte', # Einleitungen
# Fix GENERAL-GROUNDEDNESS-7: Zusätzliche Marker für Rezept-Format
'optionen/hinweise', 'optionen', 'hinweise', # Rezept-Ende
'einfaches', 'grundrezept', # Rezept-Einleitung
'fuer', 'portionen', # Rezept-Header
]
has_structure = has_valid_output and any(m in low for m in structure_markers)
# Heuristik 2: Substantielle Antworten sind grounded (ENTSCHÄRFT)
# Vorher: >100 Zeichen, 2+ Sätze
# Jetzt: >80 Zeichen, 1+ Sätze (oder >50 Zeichen mit Struktur)
# Fix GENERAL-GROUNDEDNESS-8: Mehrzeilige Antworten mit Struktur sind immer grounded
is_substantial = (
has_valid_output and
(len(clean) > 80 and clean.count('.') >= 1) or
(len(clean) > 50 and any(m in low for m in structure_markers)) or
# Fix: Mehrzeilige Antworten mit Aufzählungen sind grounded
(len(clean.splitlines()) >= 3 and any(m in low for m in ['-', '1.', '2.', 'zutaten', 'zubereitung', 'kurzbeschreibung']))
)
# Heuristik 3: Definition/Erklärung-Form
explanation_patterns = [
r'^\w+\s+ist\s+ein', # "Docker ist ein..."
r'^\w+\s+ist\s+die', # "Das ist die..."
r'^\w+\s+sind\s+', # "Y sind..."
r'^ein\s+\w+\s+ist', # "Ein Lichtjahr ist..."
r'^der\s+\w+\s+ist', # "Der Unterschied ist..."
r'^die\s+\w+\s+ist', # "Die Antwort ist..."
r'^\w+\s+bedeutet', # "X bedeutet..."
]
has_explanation_form = has_valid_output and any(re.search(p, low) for p in explanation_patterns)
# Heuristik 4: Vergleichs-Marker
comparison_markers = [' vs ', ' versus ', 'unterschied', 'vergleich', ' vs. ', ' im vergleich zu ', ' gegenüber ']
has_comparison = has_valid_output and any(m in low for m in comparison_markers)
# Heuristik 5: Wissenschaft/Fakt-Marker
fact_markers = [
'lichtjahr', 'lichtjahre', 'photosynthese', 'dna', 'protein',
'kalorie', 'kilokalorie', 'kcal', 'gramm', 'meter', 'sekunde',
'prozent', '°c', 'grad', 'jahr', 'monat', 'tag', 'stunde',
'wasser', 'sauerstoff', 'kohlenstoff', 'stickstoff',
'erde', 'mond', 'sonne', 'planet', 'stern', 'galaxie',
'atom', 'molekül', 'molekuel', 'zelle', 'organismus',
'temperatur', 'druck', 'geschwindigkeit', 'masse', 'gewicht',
]
has_fact_content = has_valid_output and any(m in low for m in fact_markers)
# Heuristik 6: Allgemeine Erklärungs-Wörter
explanation_words = ['weil', 'denn', 'daher', 'deshalb', 'folglich', 'somit', 'also']
has_explanation_words = has_valid_output and any(m in low for m in explanation_words)
# Heuristik 7: Qualitäts-Indikatoren (NEU)
# Antworten mit mehreren Zeilen oder Markdown sind wahrscheinlich gut
has_multiple_lines = has_valid_output and len(clean.splitlines()) >= 2
has_markdown = has_valid_output and any(c in clean for c in ['**', '*', '- ', '1.', '2.', '###', '##'])
# Kombinierte Entscheidung: grounded wenn mindestens eine Bedingung erfüllt (ENTSCHÄRFT)
# Vorher: strengere Schwellen (60-80 Zeichen)
# Jetzt: akzeptiere auch kurze aber strukturierte Antworten
grounded = (
has_structure or
is_substantial or
(has_explanation_form and len(clean) > 50) or
(has_comparison and len(clean) > 50) or
(has_fact_content and len(clean) > 50) or
(has_explanation_words and len(clean) > 60) or
(has_multiple_lines and len(clean) > 60) or
(has_markdown and len(clean) > 60)
)
summary = ''
for line in clean.splitlines():
line = line.strip()
if not line or line.lower() in {'befund:', 'ursache:', 'aktion:', 'kurzantwort:', 'quellen:'}:
continue
summary = line[:220]
break
if not summary:
summary = clean.splitlines()[0][:220] if clean else f'(fehler: {(err or "").strip()})'
return {
'tool': tool,
'rc': rc,
'kind': kind,
'grounded': grounded,
'output': clean if clean else (err or '').strip(),
'error': (err or '').strip(),
'summary': summary,
'confidence': 0.92 if grounded else 0.72 if clean else 0.15,
'signals': [sig for sig in ['ldap', 'authentik', 'oidc', 'dns', 'smb', 'share', 'warning', 'error', 'quelle', 'quellen'] if sig in low],
}
def render_evidence(evidence_items: list[dict], memory_ctx: str = '', retrieval_ctx: str = '') -> str:
blocks = []
if memory_ctx:
blocks.append('Persoenlicher Kontext:\n' + memory_ctx)
if retrieval_ctx:
blocks.append(retrieval_ctx)
if evidence_items:
rendered = []
for idx, item in enumerate(evidence_items, start=1):
rendered.append(
f"Evidence {idx}: tool={item.get('tool')} kind={item.get('kind')} grounded={item.get('grounded')} rc={item.get('rc')}\n"
f"{item.get('output') or item.get('error') or '(leer)'}"
)
blocks.append('Evidence-Objekte:\n' + '\n\n'.join(rendered))
return '\n\n'.join(blocks) if blocks else '(keine)'
def record_incident_if_useful(message: str, analysis: dict, final_text: str, evidence_items: list[dict]):
if analysis.get('task_type') != 'diagnose':
return
grounded = [item for item in evidence_items if item.get('grounded') and item.get('tool') in INCIDENT_MEMORY_TOOLS]
if not grounded:
return
subject = ''
for token in [analysis.get('role', ''), *analysis.get('tools', [])]:
token = str(token or '').strip()
if token in {'ops', 'research', 'personal', 'dev', 'general'}:
continue
if token:
subject = token
break
lead = ''
for line in (final_text or '').splitlines():
line = line.strip()
if not line or line.lower() in {'befund:', 'ursache:', 'aktion:'}:
continue
lead = line[:260]
break
if not lead:
lead = grounded[0].get('summary', '')[:260]
text = f"{message}\nLoesungskern: {lead}"
remember_incident(text)
def refresh_composition_policy_async():
if not COMPOSITION_POLICY_REFRESH.exists():
return
try:
subprocess.Popen(
[str(COMPOSITION_POLICY_REFRESH)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
start_new_session=True,
)
except Exception:
pass
def record_task_outcome(message: str, analysis: dict, final_text: str, evidence_items: list[dict], status: str = 'success', error_text: str = ''):
_record_task_outcome(
message,
analysis,
final_text,
evidence_items,
status=status,
error_text=error_text,
log_path=TASK_OUTCOME_LOG,
classify_intent_family=classify_intent_family,
collect_intent_families=collect_intent_families,
service_hints=SERVICE_HINTS,
refresh_composition_policy_async=refresh_composition_policy_async,
)
def persist_result_state(message: str, analysis: dict, final_text: str, evidence_items: list[dict]):
_persist_result_state(
message,
analysis,
final_text,
evidence_items,
should_skip_heavy_persistence=should_skip_heavy_persistence,
update_topic_state=update_topic_state,
compact_episode_text=compact_episode_text,
remember_episode=remember_episode,
record_incident_if_useful=record_incident_if_useful,
)
def verify_answer(message: str, analysis: dict, draft: str, evidence_items: list[dict], memory_ctx: str = '', retrieval_ctx: str = ''):
if not draft.strip():
return None
if analysis.get('task_type') in {'action', 'memory'}:
return {'verdict': 'keep', 'grounded': True, 'addresses_request': True, 'wrong_entity_jump': False, 'retry_tool': '', 'missing_points': [], 'rationale': 'Direkter Spezialpfad.', 'revised_answer': ''}
if len(evidence_items) == 1 and evidence_items[0].get('tool') in GROUNDED_FASTPATH_TOOLS and evidence_items[0].get('grounded'):
return {'verdict': 'keep', 'grounded': True, 'addresses_request': True, 'wrong_entity_jump': False, 'retry_tool': '', 'missing_points': [], 'rationale': 'Direkter grounded Tool-Fastpath.', 'revised_answer': ''}
if len(evidence_items) == 1 and evidence_items[0].get('tool') == 'web_research':
raw = evidence_items[0].get('output', '')
if raw.strip().startswith('Kurzantwort:') and 'Quellen:' in raw:
return {'verdict': 'keep', 'grounded': True, 'addresses_request': True, 'wrong_entity_jump': False, 'retry_tool': '', 'missing_points': [], 'rationale': 'Direkter Research-Fastpath mit Quellen.', 'revised_answer': ''}
if analysis.get('role') == 'general' and len(evidence_items) == 1 and evidence_items[0].get('tool') == 'general_answer' and not memory_ctx and not retrieval_ctx:
return {'verdict': 'keep', 'grounded': True, 'addresses_request': True, 'wrong_entity_jump': False, 'retry_tool': '', 'missing_points': [], 'rationale': 'Direkter General-Fastpath ohne riskante Evidenzspruenge.', 'revised_answer': ''}
# Fix PERS-GROUNDED-3: Fastpath für personal_assist (wie general_answer)
# Problem: personal_assist läuft durch LLM-Verifier und wird manchmal falsch als ungrounded markiert
# Lösung: Direkter Fastpath wenn personal_assist grounded ist und keine Memory/Retrieval-Kontext
if analysis.get('role') == 'personal' and len(evidence_items) == 1 and evidence_items[0].get('tool') == 'personal_assist' and evidence_items[0].get('grounded') and not memory_ctx and not retrieval_ctx:
return {'verdict': 'keep', 'grounded': True, 'addresses_request': True, 'wrong_entity_jump': False, 'retry_tool': '', 'missing_points': [], 'rationale': 'Direkter Personal-Fastpath ohne riskante Evidenzspruenge.', 'revised_answer': ''}
system = (
'Du bist das Qualitaetspruef-Modul von J.A.R.V.I.S. '
'Pruefe eine Antwort nur gegen Nutzerfrage, Analyse und vorhandene Evidence-Objekte. '
'Eine schlechte Antwort springt auf falsche Objekte oder Mechanismen, etwa Sonos -> Audiobookshelf. '
'Wenn die Antwort mit der vorhandenen Evidenz reparierbar ist, gib verdict=revise und liefere revised_answer. '
'Wenn das falsche Tool verwendet wurde und ein anderes vorhandenes Tool klar besser passt, gib verdict=retry und retry_tool. '
'Erfinde keine neuen Fakten.'
)
user = (
f'Nutzerfrage:\n{message}\n\n'
f'Analyse:\n{json.dumps(analysis, ensure_ascii=False)}\n\n'
f'Evidenz:\n{render_evidence(evidence_items, memory_ctx=memory_ctx, retrieval_ctx=retrieval_ctx)}\n\n'
f'Antwortentwurf:\n{draft}\n\n'
'Verfuegbare Tools fuer retry:\n'
f'{tool_catalog_text()}\n\n'
'Pruefe: trifft die Antwort das eigentliche Objekt der Anfrage, beantwortet sie die Teilfragen, bleibt sie bei belegten Aussagen, und fehlt etwas Wesentliches?'
)
for model in VERIFY_MODELS:
if not model_exists(model):
continue
rc, out, _err = call_llm_schema(model, VERIFY_SCHEMA, system, user, timeout=60)
if rc != 0 or not out:
continue
try:
data = json.loads(out)
except Exception:
continue
if not isinstance(data, dict):
continue
retry_tool = str(data.get('retry_tool') or '').strip()
if retry_tool and retry_tool not in TOOLS:
data['retry_tool'] = ''
return data
return None
def finalize_answer(message: str, analysis: dict, evidence_items: list[dict], draft: str, memory_ctx: str = '', retrieval_ctx: str = '', allow_retry: bool = True, allow_synthesize: bool = True):
tool_kinds = {name: cfg.get('kind', '') for name, cfg in TOOLS.items()}
_skip_verify = _should_skip_verify(analysis, evidence_items, tool_kinds)
review = ({} if _skip_verify else verify_answer(message, analysis, draft, evidence_items, memory_ctx=memory_ctx, retrieval_ctx=retrieval_ctx)) or {}
verdict = review.get('verdict', 'keep')
revised = str(review.get('revised_answer') or '').strip()
retry_tool = str(review.get('retry_tool') or '').strip()
if verdict == 'retry' and allow_retry and retry_tool and retry_tool not in [item.get('tool') for item in evidence_items] and TOOLS.get(retry_tool, {}).get('kind') != 'action':
rc, out, err = run_tool(retry_tool, message)
if rc == 0 and out:
merged = evidence_items + [evidence_object(retry_tool, rc, out, err)]
new_draft = synthesize(message, analysis, merged, memory_ctx=memory_ctx, retrieval_ctx=retrieval_ctx)
return finalize_answer(message, analysis, merged, new_draft, memory_ctx=memory_ctx, retrieval_ctx=retrieval_ctx, allow_retry=False, allow_synthesize=False)
if err:
evidence_items = evidence_items + [evidence_object(retry_tool, rc, out, err)]
if verdict == 'revise' and revised:
return revised, evidence_items
if verdict == 'keep':
return draft, evidence_items
if allow_synthesize and evidence_items and draft == evidence_items[-1].get('output', ''):
new_draft = synthesize(message, analysis, evidence_items, memory_ctx=memory_ctx, retrieval_ctx=retrieval_ctx)
if new_draft and new_draft != draft:
return finalize_answer(message, analysis, evidence_items, new_draft, memory_ctx=memory_ctx, retrieval_ctx=retrieval_ctx, allow_retry=False, allow_synthesize=False)
return draft, evidence_items
def run_tool(name: str, message: str):
trace_stage('run_tool', name)
if name == 'memory_profile':
low = norm(message)
if re.search(r'^(bitte\s+)?(merke dir|merk dir|denk daran|behalte im kopf)\b', low):
return run([str(AGENT_MEMORY), 'ingest', message], timeout=20)
if '?' not in message and re.search(r'\b(ich will|ich bevorzuge|mir ist wichtig|achte darauf|das ist falsch|es sollte|er sollte|wenn es um .* geht|ich .* wichtig finde)\b', low):
return run([str(AGENT_MEMORY), 'ingest', message], timeout=20)
if 'was weisst du ueber mich' in low or 'was weisst du über mich' in message.lower() or 'mein profil' in low or 'gespeichertes profil' in low:
return run([str(AGENT_MEMORY), 'summary'], timeout=20)
return run([str(AGENT_MEMORY), 'answer', message], timeout=20)
if name == 'briefing':
return run([str(ROOT / 'bin' / 'agent-briefing')], timeout=70)
if name == 'setup_lookup':
return run([str(ROOT / 'bin' / 'ops-setup-qa'), message], timeout=70)
if name == 'public_services_health':
return run([str(ROOT / 'bin' / 'ops-public-services-health')], timeout=70)
if name == 'log_hotspots':
return run([str(ROOT / 'bin' / 'ops-log-hotspots'), message], timeout=90)
if name == 'host_slow_diagnose':
return run([str(ROOT / 'bin' / 'ops-host-slow-diagnose'), message], timeout=120)
if name == 'kasm_diagnose':
return run([str(ROOT / 'bin' / 'ops-kasm-diagnose'), message], timeout=120)
if name == 'user_service_access_diagnose':
return run([str(ROOT / 'bin' / 'ops-user-service-access-diagnose'), message], timeout=120)
if name == 'web_root_cause':
return run([str(ROOT / 'bin' / 'ops-web-root-cause'), message], timeout=120)
if name == 'sonos_library_diagnose':
return run([str(ROOT / 'bin' / 'ops-sonos-library-diagnose'), message], timeout=120)
if name == 'emby_user_provision':
return run([str(ROOT / 'bin' / 'ops-emby-user-provision-advice'), message], timeout=120)
if name == 'light_status':
return run([str(ROOT / 'bin' / 'ops-lights-on')], timeout=40)
if name == 'light_usage':
return run([str(ROOT / 'bin' / 'ops-light-usage-advice'), '168'], timeout=90)
if name == 'light_control':
return run([str(ROOT / 'bin' / 'home-lightctl'), message], timeout=40)
if name == 'web_research':
quick = ROOT / 'bin' / 'research-quick-answer'
low = norm(message)
if quick.exists() and re.search(r'^(was ist|wer ist|wo ist|wo liegt|wo befindet sich|warum|wieso|weshalb|wie leben|wie lebt|wo leben|wo lebt|was essen|was frisst|wie gross ist|wie groß ist|wie alt wird|wie funktioniert)\b', low):
rc, out, err = run([str(quick), message], timeout=25)
if rc == 0 and out.strip() and 'nicht_verifiziert' not in out.lower():
return rc, out, err
# Fix WEB-RESEARCH-FALLBACK: Wenn Web-Recherche fehlschlägt, versuche general-local-assist
# Problem: 26% der web_research Aufrufe waren ungrounded wegen Netzwerkfehlern
# Lösung: Fallback auf lokales Wissen statt nicht_verifiziert zurückgeben
rc, out, err = run([str(ROOT / 'bin' / 'research-web-answer'), message], timeout=180)
# Fix WEB-STDERR-1: Prüfe auch stderr wenn stdout leer ist
# Problem: research-web-answer gibt 'nicht_verifiziert' auf stderr aus bei Fehlern
combined_output = out.strip() if out.strip() else err.strip()
if rc != 0 or not combined_output or 'nicht_verifiziert' in combined_output.lower():
# Web-Recherche fehlgeschlagen - versuche general-local-assist als Fallback
# Fix WEB-FALLBACK-1: Akzeptiere auch Exit Code 1 wenn substanzielle Antwort vorhanden
# Problem: general-local-assist gibt oft rc=1 zurück auch bei brauchbarer Antwort
env = os.environ.copy()
env['BIBABOT_SELF_CRITIQUE'] = '0'
env.setdefault('MAC_OLLAMA_MODEL_GENERAL', os.environ.get('BIBABOT_GENERAL_CHAIN', GENERAL_CHAIN_DEFAULT))
env['MAC_OLLAMA_NUM_CTX_CHAT'] = env.get('MAC_OLLAMA_NUM_CTX_CHAT', '24576')
rc2, out2, err2 = run([str(ROOT / 'bin' / 'general-local-assist'), message], timeout=70, env=env)
# Akzeptiere Antwort wenn: Exit Code 0 ODER (Exit Code 1 UND substanzielle Antwort >100 Zeichen)
has_substantial_output = out2.strip() and len(out2.strip()) > 100 and 'nicht_verifiziert' not in out2.lower()
if (rc2 == 0 and out2.strip()) or (rc2 == 1 and has_substantial_output):
# Markiere als Web-Antwort mit Hinweis auf Fallback
return 0, f"{out2.strip()}\n\n(Hinweis: Web-Recherche nicht verfügbar, Antwort basiert auf lokalem Wissen)", ""
return rc, out, err
if name == 'url_research':
return run([str(ROOT / 'bin' / 'research-url-answer'), message], timeout=180)
if name == 'compare_selfhosted':
return run([str(ROOT / 'bin' / 'research-compare-selfhosted'), message], timeout=150)
if name == 'security_plan':
return run([str(ROOT / 'bin' / 'research-security-plan')], timeout=150)
if name == 'media_advice':
low = norm(message)
mode = 'audiobooks'
if 'lidarr' in low:
mode = 'lidarr'
elif 'emby' in low:
mode = 'emby'
return run([str(ROOT / 'bin' / 'ops-media-advice'), mode], timeout=90)
if name == 'personal_assist':
return run([str(ROOT / 'bin' / 'personal-local-assist'), message], timeout=150)
if name == 'dev_build':
return run([str(ROOT / 'bin' / 'dev-local-assist'), message], timeout=240)
if name == 'general_answer':
env = os.environ.copy()
env['BIBABOT_SELF_CRITIQUE'] = '0'
env['MAC_OLLAMA_CALL_TIMEOUT_SEC'] = '75'
env.setdefault('MAC_OLLAMA_MODEL_GENERAL', os.environ.get('BIBABOT_GENERAL_CHAIN', GENERAL_CHAIN_DEFAULT))
env['MAC_OLLAMA_NUM_CTX_CHAT'] = env.get('MAC_OLLAMA_NUM_CTX_CHAT', '24576')
return run([str(ROOT / 'bin' / 'general-local-assist'), message], timeout=100, env=env)
if name == 'ops_exec':
return run([str(ROOT / 'bin' / 'ops-exec-action'), message], timeout=120)
if name == 'ops_install':
return run([str(ROOT / 'bin' / 'ops-install-action'), message], timeout=180)
if name == 'ops_api':
return run([str(ROOT / 'bin' / 'ops-api-query'), message], timeout=100)
if name == 'ops_deep_analyze':
return run([str(ROOT / 'bin' / 'ops-deep-analyze'), message], timeout=130)
if name == 'qdrant_search':
return run([str(ROOT / 'bin' / 'qdrant-context-search'), message], timeout=40)
if name == 'expert_write':
return run([str(ROOT / 'bin' / 'expert-write-assist'), message], timeout=180)
if name == 'expert_code':
return run([str(ROOT / 'bin' / 'expert-code-assist'), message], timeout=180)
if name == 'expert_research':
return run([str(ROOT / 'bin' / 'expert-deep-research'), message], timeout=200)
if name == 'expert_edit':
return run([str(ROOT / 'bin' / 'expert-proofread'), message], timeout=180)
if name == 'expert_strategy':
return run([str(ROOT / 'bin' / 'expert-strategy-assist'), message], timeout=180)
if name == 'goals_manage':
return run([str(ROOT / 'bin' / 'goals-manage'), message], timeout=15)
if name == 'contacts_manage':
return run([str(ROOT / 'bin' / 'contacts-manage'), message], timeout=15)
if name == 'agent_introspect':
return run([str(ROOT / 'bin' / 'agent-introspect'), 'analyze'], timeout=30)
if name == 'agent_self_correct':
return run([str(ROOT / 'bin' / 'agent-self-correct'), 'check', message], timeout=30)
if name == 'agent_reason':
return run([str(ROOT / 'bin' / 'agent-reason'), 'start', message, '--max-depth', '10'], timeout=60)
if name == 'agent_anticipate':
return run([str(ROOT / 'bin' / 'agent-anticipate'), 'predict', message], timeout=30)
if name == 'agent_emotion':
return run([str(ROOT / 'bin' / 'agent-emotion'), 'analyze', message], timeout=30)
if name == 'agent_simulate':
return run([str(ROOT / 'bin' / 'agent-simulate'), 'action', message], timeout=45)
if name == 'agent_synthesize':
return run([str(ROOT / 'bin' / 'agent-synthesize'), 'solve', message, '--domains', 'computing,homelab'], timeout=60)
# AUTO-LEARN TRIGGER if name.startswith("expert_") or name.startswith("kb_"): import subprocess as _sp _sp.run([str(ROOT / "bin" / "agent-learn"), "queue", name, "--priority", "5"], capture_output=True)
return 1, '', f'unknown_tool:{name}'
def synthesize(message: str, analysis: dict, evidence_items: list[dict], memory_ctx: str = '', retrieval_ctx: str = ''):
role = FINAL_ROLE_MAP.get(analysis.get('role', 'general'), 'general')
files = []
analysis_text = ['Orchestrator-Analyse:', f"- role={analysis.get('role')}", f"- task_type={analysis.get('task_type')}", f"- tools={', '.join(analysis.get('tools', []))}"]
for item in analysis.get('reasoning_focus', [])[:4]:
analysis_text.append(f'- focus={item}')
files.append(write_temp('analysis', '\n'.join(analysis_text)))
mem = memory_ctx or memory_context(message)
if mem:
files.append(write_temp('memory', mem))
if retrieval_ctx:
files.append(write_temp('retrieval', retrieval_ctx))
files.append(write_temp('contract', response_contract(analysis)))
files.append(write_temp('evidence', render_evidence(evidence_items)))
for idx, item in enumerate(evidence_items, start=1):
files.append(write_temp(f'tool{idx}', f"Tool: {item.get('tool')}\nKind: {item.get('kind')}\nGrounded: {item.get('grounded')}\n\n{item.get('output') or item.get('error') or '(leer)'}"))
cmd = [str(LLM_LOCAL), '--role', role]
for path in files:
cmd.extend(['--extra-context-file', path])
cmd.append(message)
_synth_env = os.environ.copy()
_synth_env['BIBABOT_SELF_CRITIQUE'] = '0'
_synth_env['MAC_OLLAMA_CALL_TIMEOUT_SEC'] = '60'
# Use better synthesis models when expert tools are involved
_expert_tools = {'expert_write', 'expert_code', 'expert_research', 'expert_edit', 'expert_strategy'}
_has_expert = any(item.get('tool') in _expert_tools for item in evidence_items)
if _has_expert:
_synth_env['MAC_OLLAMA_MODEL_OPS'] = 'gemma3:27b-it-qat,qwen3:30b-a3b,mistral-small3.1:latest'
_synth_env['MAC_OLLAMA_MODEL_GENERAL'] = 'gemma3:27b-it-qat,lfm2:24b,mistral-small3.1:latest'
else:
_synth_env['MAC_OLLAMA_MODEL_OPS'] = 'mistral-small3.1:latest,llama3.1:8b,qwen2.5:3b'
_synth_env['MAC_OLLAMA_MODEL_GENERAL'] = 'mistral-small3.1:latest,llama3.1:8b,qwen2.5:3b'
try:
rc, out, err = run(cmd, timeout=90, env=_synth_env)
finally:
for path in files:
try:
os.unlink(path)
except OSError:
pass
if rc == 0 and out:
return out
if evidence_items:
return evidence_items[-1].get('output', '') or evidence_items[-1].get('error', '')
return 'nicht_verifiziert: Kein belastbares Ergebnis erzeugt.'
# ─── Quality Feedback Loop ────────────────────────────────────────────────────
FEEDBACK_TOOLS = {'expert_write', 'expert_code', 'expert_research', 'expert_edit', 'expert_strategy'}
_FEEDBACK_ROLE_MAP = {
'expert_write': 'write',
'expert_code': 'code_expert',
'expert_research': 'research_deep',
'expert_edit': 'proofread',
'expert_strategy': 'strategy',
}
def judge_output(message: str, output: str) -> tuple[int, str, str]:
"""Ruft llm-judge auf und gibt (score, critique, improve) zurueck."""
judge_bin = ROOT / 'bin' / 'llm-judge'
if not judge_bin.exists():
return 3, '', ''
try:
import json as _json
rc, out, err = run([str(judge_bin), message[:1500], output[:3500]], timeout=35)
if rc == 0 and out:
start = out.find('{')
end = out.rfind('}') + 1
if start >= 0 and end > start:
d = _json.loads(out[start:end])
score = max(1, min(5, int(d.get('score', 3))))
return score, str(d.get('critique', '')), str(d.get('improve', ''))
except Exception:
pass
return 3, '', ''
def refine_output(message: str, role: str, original: str, critique: str, improve: str) -> str:
"""Verbessert die Antwort basierend auf der Qualitaetskritik."""
orig_ctx = 'Deine erste Antwort (zu verbessern):\n' + original
crit_ctx = (
'Qualitaetskritik vom Pruefer:\n'
'Note: unzureichend\n'
'Kritik: ' + critique + '\n'
'Verbesserungspunkte: ' + improve + '\n\n'
'Erstelle jetzt eine vollstaendige, verbesserte Version deiner Antwort.'
)
orig_file = write_temp('refine_orig', orig_ctx)
critique_file = write_temp('refine_critique', crit_ctx)
try:
cmd = [str(LLM_LOCAL), '--role', role,
'--extra-context-file', orig_file,
'--extra-context-file', critique_file,
message]
env = os.environ.copy()
env['MAC_OLLAMA_MODEL_OPS'] = 'gemma3:27b-it-qat,qwen3:30b-a3b,mistral-small3.1:latest'
env['MAC_OLLAMA_MODEL_GENERAL'] = 'gemma3:27b-it-qat,lfm2:24b,mistral-small3.1:latest'
env['MAC_OLLAMA_CALL_TIMEOUT_SEC'] = '90'
env['BIBABOT_SELF_CRITIQUE'] = '0'
rc, out, err = run(cmd, timeout=120, env=env)
if rc == 0 and out and len(out) > 80:
return out
except Exception:
pass
finally:
for f in (orig_file, critique_file):
try:
os.unlink(f)
except OSError:
pass
return original
def quality_feedback_loop(message: str, analysis: dict, final_text: str, evidence_items: list[dict]) -> str:
"""Bewertet die Ausgabe mit llm-judge; bei Score < 3 wird sie mit dem Originalmodell verfeinert.
Greift nur fuer expert_* Tools ein, alles andere bleibt unveraendert."""
used_tools = {item.get('tool') for item in evidence_items}
if not used_tools & FEEDBACK_TOOLS:
return final_text
if not final_text or len(final_text) < 80:
return final_text
score, critique, improve = judge_output(message, final_text)
trace_stage('quality_judge', f'score={score}', f'critique={critique[:60]}')
if score >= 3:
# Qualitaet ausreichend — keine Aenderung
return final_text
# Score 1 oder 2 → verfeinern
tool = next((t for t in evidence_items if t.get('tool') in FEEDBACK_TOOLS), {}).get('tool', '')
role = _FEEDBACK_ROLE_MAP.get(tool, 'general')
trace_stage('quality_refine_start', f'tool={tool}', f'role={role}')
refined = refine_output(message, role, final_text, critique, improve)
trace_stage('quality_refine_done', f'len={len(refined)}')
return refined
def log_event(message: str, analysis: dict, evidence_items: list[dict], role_hint: str | None):
try:
LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
entry = {
'ts': datetime.now(timezone.utc).isoformat(),
'message': message,
'role_hint': role_hint,
'analysis': analysis,
'tools': [item.get('tool') for item in evidence_items],
'evidence': [
{
'tool': item.get('tool'),
'kind': item.get('kind'),
'grounded': item.get('grounded'),
'rc': item.get('rc'),
'summary': item.get('summary'),
'confidence': item.get('confidence'),
}
for item in evidence_items
],
}
with LOG_PATH.open('a', encoding='utf-8') as f:
f.write(json.dumps(entry, ensure_ascii=False) + '\n')
except Exception:
pass
AUTO_KB_LEARN = ROOT / "bin" / "auto-kb-learn"
def auto_learn_async(message: str, answer: str):
"""Non-blocking background learning from research tasks."""
if not AUTO_KB_LEARN.exists():
return
try:
subprocess.Popen(
[str(AUTO_KB_LEARN), message[:500], answer[:2000]],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
start_new_session=True,
)
except Exception:
pass
def main():
p = argparse.ArgumentParser()
p.add_argument('--role-hint', choices=sorted(ROLE_HINTS), default=None)
p.add_argument('message')
args = p.parse_args()
message = normalize_inbound_message(args.message)
role_hint = args.role_hint
trace_stage('start', message)
remember_user_signal(message)
trace_stage('after_remember_user_signal')
seed = heuristic_analysis(message, role_hint)
trace_stage('seed', json.dumps(seed, ensure_ascii=False))
memory_ctx = memory_context(message, seed)
trace_stage('memory_ctx', 'yes' if memory_ctx else 'no')
retrieval_ctx = retrieval_context(message, seed, role_hint)
trace_stage('retrieval_ctx', 'yes' if retrieval_ctx else 'no')
playbook_ctx = playbook_context(message, seed)
retrieval_ctx = append_context(retrieval_ctx, playbook_ctx)
trace_stage('playbook_ctx', 'yes' if playbook_ctx else 'no')
if seed.get('confidence', 0.0) >= 0.65 or seed.get('dangerous_action'):
analysis = normalize_analysis(seed, message, role_hint)
else:
raw = analyze_with_llm(message, role_hint, memory_ctx=memory_ctx, retrieval_ctx=retrieval_ctx)
analysis = normalize_analysis(raw or seed, message, role_hint)
if not memory_ctx and analysis.get('needs_memory'):
memory_ctx = memory_context(message, analysis) or memory_ctx
retrieval_ctx = retrieval_context(message, analysis, role_hint) or retrieval_ctx
playbook_ctx = playbook_ctx or playbook_context(message, analysis)
retrieval_ctx = append_context(retrieval_ctx, playbook_ctx)
plan = None
if USE_PLAN_FIRST:
try:
plan = build_plan(message, analysis, memory_ctx=memory_ctx, retrieval_ctx=retrieval_ctx, playbook_ctx=playbook_ctx)
except Exception:
plan = None
if plan:
trace_stage('plan_first', json.dumps(plan, ensure_ascii=False))
analysis = apply_plan_first(analysis, plan)
if not memory_ctx and analysis.get('needs_memory'):
memory_ctx = memory_context(message, analysis) or memory_ctx
refreshed = retrieval_context(message, analysis, role_hint)
retrieval_ctx = append_context(refreshed or retrieval_ctx, playbook_ctx)
retrieval_ctx = append_context(retrieval_ctx, render_plan_context(plan))
if os.environ.get('SUBAGENT_DEBUG', '0') == '1':
print(json.dumps(analysis, ensure_ascii=False), file=sys.stderr)
tools = analysis.get('tools', [])
if not tools:
tools = ['general_answer']
analysis['tools'] = tools
trace_stage('tools', ','.join(tools))
if len(tools) == 1:
rc, out, err = run_tool(tools[0], message)
trace_stage('tool_result', tools[0], f'rc={rc}', f'out={len(out)}', f'err={len(err)}')
evidence_items = [evidence_object(tools[0], rc, out, err)]
# Auto-escalate: ops_api found critical issues + user wants solutions → add ops_deep_analyze
_low = norm(message)
if (tools[0] == 'ops_api' and rc == 0 and out and
analysis.get('task_type') in {'diagnose', 'status'} and
any(k in out.lower() for k in ['unhealthy', 'error', 'exited', 'failed']) and
re.search(r'(fix|reparier|beheb|wie|was soll|was tun|wie kann|loesung|schritte|warum|analyse)', _low)):
trace_stage('auto_escalate', 'ops_api→ops_deep_analyze')
_deep_msg = f"{message}\n\nAPI Daten:\n{out[:800]}"
_rc2, _out2, _err2 = run_tool('ops_deep_analyze', _deep_msg)
if _rc2 == 0 and _out2:
evidence_items = [evidence_object('ops_api', rc, out, err),
evidence_object('ops_deep_analyze', _rc2, _out2, _err2)]
draft = synthesize(message, analysis, evidence_items, memory_ctx=memory_ctx, retrieval_ctx=retrieval_ctx)
final, used_outputs = finalize_answer(message, analysis, evidence_items, draft, memory_ctx=memory_ctx, retrieval_ctx=retrieval_ctx)
final = postprocess_final_answer(message, analysis, final)
if not should_skip_quality_feedback(analysis, used_outputs):
final = quality_feedback_loop(message, analysis, final, used_outputs)
log_event(message, analysis, used_outputs, role_hint)
persist_result_state(message, analysis, final, used_outputs)
if USE_BUENZLI:
final = buenzli_response(final)
record_task_outcome(message, analysis, final, used_outputs, status='success')
print(final)
return 0
log_event(message, analysis, evidence_items, role_hint)
if rc == 0 and out:
if can_fast_exit_single_tool(analysis, evidence_items, memory_ctx=memory_ctx, retrieval_ctx=retrieval_ctx):
final, used_outputs = out, evidence_items
else:
final, used_outputs = finalize_answer(message, analysis, evidence_items, out, memory_ctx=memory_ctx, retrieval_ctx=retrieval_ctx)
log_event(message, analysis, used_outputs, role_hint)
final = postprocess_final_answer(message, analysis, final)
if not should_skip_quality_feedback(analysis, used_outputs):
final = quality_feedback_loop(message, analysis, final, used_outputs)
persist_result_state(message, analysis, final, used_outputs)
# Auto-learn from research results
if tools[0] in {"web_research", "url_research", "compare_selfhosted"} and final and len(final) > 50:
auto_learn_async(message, final)
if USE_BUENZLI:
final = buenzli_response(final)
record_task_outcome(message, analysis, final, used_outputs, status='success')
print(final)
return 0
if out:
record_task_outcome(message, analysis, out, evidence_items, status='tool_output_unverified')
print(out)
return 0
fallback_tools = []
if tools[0] == 'general_answer' and analysis.get('task_type') in {'explain', 'howto'} and '?' in message:
fallback_tools = ['web_research']
elif tools[0] == 'web_research':
fallback_tools = ['general_answer']
elif tools[0] == 'user_service_access_diagnose':
fallback_tools = ['setup_lookup']
for fb in fallback_tools:
frc, fout, ferr = run_tool(fb, message)
if frc == 0 and fout:
merged = evidence_items + [evidence_object(fb, frc, fout, ferr)]
trace_stage('fallback_tool_result', fb, f'rc={frc}', f'out={len(fout)}', f'err={len(ferr)}')
final, used_outputs = finalize_answer(message, analysis, merged, fout, memory_ctx=memory_ctx, retrieval_ctx=retrieval_ctx)
log_event(message, analysis, used_outputs, role_hint)
final = postprocess_final_answer(message, analysis, final)
if not should_skip_quality_feedback(analysis, used_outputs):
final = quality_feedback_loop(message, analysis, final, used_outputs)
persist_result_state(message, analysis, final, used_outputs)
if USE_BUENZLI:
final = buenzli_response(final)
record_task_outcome(message, analysis, final, used_outputs, status='success')
print(final)
return 0
if err:
record_task_outcome(message, analysis, '', evidence_items, status='tool_failed', error_text=err)
print(f'nicht_verifiziert: Tool {tools[0]} schlug fehl ({err}).')
return 1
record_task_outcome(message, analysis, '', evidence_items, status='no_result')
print('nicht_verifiziert: Kein belastbares Ergebnis erzeugt.')
return 1
# Sequential execution: each tool gets accumulated context from previous steps
_seq_result = _run_sequential_tools(
message,
tools,
analysis,
is_complex_task=is_complex_task,
run_tool=run_tool,
evidence_object=evidence_object,
trace_stage=trace_stage,
sequential_tool_budget=sequential_tool_budget,
summarize_sequential_context=summarize_sequential_context,
)
collected = list(_seq_result.get('collected') or [])
_failed_ev = _seq_result.get('failed_ev')
_is_seq = bool(_seq_result.get('is_sequential'))
if _failed_ev is not None:
_tool = _failed_ev.get('tool', '')
_err = _failed_ev.get('error', '') or ('timeout' if _failed_ev.get('rc') == 124 else 'tool_failed')
record_task_outcome(message, analysis, '', collected, status='tool_failed', error_text=f'{_tool}:{_err}')
print(f'nicht_verifiziert: Kompositionsschritt {_tool} schlug fehl ({_err}).')
return 1
# Skip expensive synthesize when final tool already produced a complete structured answer
_last_ev = collected[-1] if collected else {}
_last_out = _last_ev.get('output', '')
_last_kind = TOOLS.get(_last_ev.get('tool', ''), {}).get('kind', '')
_prep_direct = bool(analysis.get('composition_reason')) and _last_kind == 'final' and _last_out and all(bool(item.get('grounded')) for item in collected[:-1])
_light_composed = build_lightweight_composed_answer(message, analysis, collected)
if _light_composed:
draft = _light_composed
trace_stage('synthesize_skipped', 'lightweight_composed_answer')
elif _is_seq and _last_kind == 'final' and _last_out and (len(_last_out) > 100 or _prep_direct):
draft = _last_out
trace_stage('synthesize_skipped', 'final_tool_output_used_directly' if len(_last_out) > 100 else 'composition_final_output_used_directly')
else:
draft = synthesize(message, analysis, collected, memory_ctx=memory_ctx, retrieval_ctx=retrieval_ctx)
trace_stage('after_synthesize', f'draft={len(draft)}')
final, used_outputs = finalize_answer(message, analysis, collected, draft, memory_ctx=memory_ctx, retrieval_ctx=retrieval_ctx)
final = postprocess_final_answer(message, analysis, final)
final = quality_feedback_loop(message, analysis, final, used_outputs)
log_event(message, analysis, used_outputs, role_hint)
persist_result_state(message, analysis, final, used_outputs)
if USE_BUENZLI:
final = buenzli_response(final)
record_task_outcome(message, analysis, final, used_outputs, status='success')
print(final)
return 0
if __name__ == '__main__':
raise SystemExit(main())