133 lines
4.9 KiB
Python
133 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
task_agent_matcher.py
|
|
|
|
Ebene 3 des Skills: läuft bei Agent A (Master) zur Laufzeit, sobald ein
|
|
neuer Task ansteht. Entscheidet, welcher Agent bei welchem Standard-Modell
|
|
am besten zum Task-Typ passt - basierend auf:
|
|
|
|
- der von daily_model_check.py gebauten Prioritätsliste
|
|
(state/model_priority.json)
|
|
- aktiven Kill-Switches (state/kill_switches.json), z.B. wegen
|
|
Budget-Überschreitung -> Agent wird übersprungen bzw. auf
|
|
"postpone_to_next_day" gesetzt
|
|
- den Capability-Tags pro Agent (config/agents.yaml)
|
|
|
|
Jeder Agent bekommt exakt dieses Konzept (Config + Prioritätsliste)
|
|
mitgeliefert, damit alle konsistent nach derselben Logik laufen, egal
|
|
wer die Entscheidung am Ende ausführt.
|
|
|
|
Aufruf (Beispiel):
|
|
python3 task_agent_matcher.py --task-tag javascript
|
|
python3 task_agent_matcher.py --task-tag homebrew_code
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
CONFIG_PATH = Path(__file__).parent.parent / "config" / "agents.yaml"
|
|
STATE_DIR = Path(__file__).parent.parent / "state"
|
|
PRIORITY_PATH = STATE_DIR / "model_priority.json"
|
|
KILL_SWITCH_PATH = STATE_DIR / "kill_switches.json"
|
|
|
|
|
|
def load_yaml(path):
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
return yaml.safe_load(f)
|
|
|
|
|
|
def load_json(path, default=None):
|
|
if not path.exists():
|
|
return default if default is not None else {}
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def find_matches(task_tag, config, priority, kill_switches):
|
|
"""Gibt eine nach Preis sortierte Liste von (agent, model, price) für
|
|
Agents zurück, deren Capability-Tags zum Task-Typ passen und die
|
|
NICHT gerade per Kill-Switch auf 'postpone_to_next_day' stehen.
|
|
"""
|
|
candidates = []
|
|
postponed = []
|
|
|
|
for agent_name, agent_cfg in config["agents"].items():
|
|
if task_tag not in agent_cfg.get("capability_tags", []):
|
|
continue
|
|
|
|
if agent_name in kill_switches.get("kill_switches", {}):
|
|
ks = kill_switches["kill_switches"][agent_name]
|
|
postponed.append({
|
|
"agent": agent_name,
|
|
"reason": ks["reason"],
|
|
"action": ks["action"],
|
|
"fallback": ks.get("basic_questions_fallback"),
|
|
})
|
|
continue
|
|
|
|
ladder = priority.get("agents", {}).get(agent_name, [])
|
|
for step in ladder:
|
|
candidates.append({
|
|
"agent": agent_name,
|
|
"model": step["model"],
|
|
"cost_factor": step.get("cost_factor"),
|
|
"price_per_1k_tokens_eur": step.get("price_per_1k_tokens_eur"),
|
|
})
|
|
|
|
# Günstigste zuerst; Steps ohne Preisangabe (z.B. Free-Router) ganz vorne
|
|
candidates.sort(key=lambda c: (c["price_per_1k_tokens_eur"] is not None, c.get("price_per_1k_tokens_eur") or 0))
|
|
|
|
return candidates, postponed
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Task-zu-Agent-Matching (Ebene 3)")
|
|
parser.add_argument("--task-tag", required=True, help="Capability-Tag des Tasks, z.B. javascript, homebrew_code")
|
|
args = parser.parse_args()
|
|
|
|
config = load_yaml(CONFIG_PATH)
|
|
priority = load_json(PRIORITY_PATH)
|
|
kill_switches = load_json(KILL_SWITCH_PATH)
|
|
|
|
if not priority:
|
|
print("FEHLER: state/model_priority.json fehlt - bitte zuerst daily_model_check.py laufen lassen")
|
|
return
|
|
|
|
candidates, postponed = find_matches(args.task_tag, config, priority, kill_switches)
|
|
|
|
if not candidates:
|
|
local_fb = config.get("local_fallback", {})
|
|
if args.task_tag in local_fb.get("allowed_for_task_tags", []):
|
|
print(f"Kein Cloud-Kandidat für '{args.task_tag}' verfügbar (evtl. alle betroffenen Agents "
|
|
f"per Kill-Switch verschoben) - lokaler Fallback erlaubt:")
|
|
print(f" -> {local_fb['model']} (kostenfrei, VPS-Ollama)")
|
|
else:
|
|
print(f"Kein Kandidat für '{args.task_tag}' verfügbar und lokaler Fallback für diesen "
|
|
f"Task-Typ NICHT erlaubt (siehe config: local_fallback.allowed_for_task_tags). "
|
|
f"Task sollte pausiert werden.")
|
|
for p in postponed:
|
|
print(f" (verschoben: {p['agent']} - {p['reason']}, Fallback: {p['fallback']})")
|
|
return
|
|
|
|
best = candidates[0]
|
|
print(f"Beste Wahl für Task-Tag '{args.task_tag}':")
|
|
print(f" Agent: {best['agent']}")
|
|
print(f" Modell: {best['model']} (Kostenfaktor {best['cost_factor']}x)")
|
|
|
|
if len(candidates) > 1:
|
|
print("\nWeitere passende Kandidaten (Eskalationsreserve):")
|
|
for c in candidates[1:]:
|
|
print(f" - {c['agent']} / {c['model']} (Faktor {c['cost_factor']}x)")
|
|
|
|
if postponed:
|
|
print("\nHinweis, folgende Agents sind aktuell wegen Budget verschoben:")
|
|
for p in postponed:
|
|
print(f" - {p['agent']}: {p['reason']} -> {p['action']} (Grundfragen: {p['fallback']})")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|