Gowler/Masterarbeit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8a22d586f1b162c113e6af897373b91eb0bf17b1
Masterarbeit/.claude/skills/run-experiment/glm-kimi-adapter.py
T
Christoph Schwörer 8a22d586f1 iteration 8
2026-08-28 19:41:13 +02:00

977 lines
42 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""
TensorX API Adapter fuer den run-experiment Skill.
Dieser Adapter fuehrt einen Headless-Versuchslauf mit einem OpenAI-kompatiblen
Modell (Z.AI GLM oder Moonshot Kimi) ueber den TensorX API-Gateway durch.
Er implementiert einen minimalen Agent-Loop mit Tool-Calling und erfasst
Token-Metadaten (inkl. Reasoning-Tokens) aus jeder API-Antwort.
Der API-Key wird automatisch aus der Cline providers.json gelesen
(~/.cline/data/settings/providers.json, Provider "tensorx").
Verwendung:
python glm-kimi-adapter.py \
--prompt <Pfad zur combined_prompt.md> \
--root <Root-Verzeichnis der Codebasis> \
--output <Laufverzeichnis/Ergebnisse> \
--model <Modell-ID, z.B. z-ai/glm-5.2 oder moonshotai/kimi-k3> \
--effort <low|medium|high|xhigh> \
[--max-turns 50] \
[--temperature 1.0] \
[--timeout 0]
Ausgaben:
<Laufverzeichnis>/RawResult.json – normalisierte Messdaten
<Laufverzeichnis>/Stderr.log – Fehler- und Debug-Ausgaben
Der Adapter ist bewusst eigenstaendig (nur Python-Standardbibliothek + requests).
"""
import argparse
import hashlib
import json
import os
import re
import subprocess
import sys
import time
import traceback
from datetime import datetime, timezone
from pathlib import Path
try:
import requests
except ImportError:
sys.stderr.write("FEHLER: 'requests' ist nicht installiert.\n")
sys.exit(2)
# ---------------------------------------------------------------------------
# Provider-Konfiguration
# ---------------------------------------------------------------------------
PROVIDERS = {
"tensorx": {
"name": "TensorX API Gateway",
"base_url": "https://api.tensorx.ai/v1",
"env_key": "TENSORX_API_KEY",
},
}
# Modell-Praefix -> Effort-Parameter-Typ
# z-ai/* Modelle nutzen den 'thinking'-Parameter (level: none|low|medium|high|xhigh)
# moonshotai/* Modelle nutzen 'reasoning_effort' (low|medium|high)
MODEL_EFFORT_TYPE = {
"z-ai": "thinking",
"moonshotai": "reasoning_effort",
}
# Effort-Mapping: Skill-Effort -> API-Wert je Effort-Typ
EFFORT_MAP = {
"low": {"thinking": "low", "reasoning_effort": "low"},
"medium": {"thinking": "medium", "reasoning_effort": "medium"},
"high": {"thinking": "high", "reasoning_effort": "high"},
"xhigh": {"thinking": "xhigh", "reasoning_effort": "high"},
"max": {"thinking": "xhigh", "reasoning_effort": "high"},
}
def load_cline_api_key():
"""
Liest den TensorX API-Key aus der Cline providers.json.
Pfad: ~/.cline/data/settings/providers.json
Rueckgabe: (api_key, base_url) oder (None, None).
"""
home = Path.home()
providers_file = home / ".cline" / "data" / "settings" / "providers.json"
if not providers_file.is_file():
return None, None
try:
data = json.loads(providers_file.read_text(encoding="utf-8"))
tx = data.get("providers", {}).get("tensorx", {}).get("settings", {})
return tx.get("apiKey"), tx.get("baseUrl")
except (json.JSONDecodeError, KeyError):
return None, None
# Denylist fuer schreibende/bauende Kommandos
DENIED_COMMAND_PATTERNS = [
r"\brm\b", r"\brmdir\b", r"\bmv\b", r"\bcp\b", r"\bdd\b",
r"\btruncate\b", r"\bchmod\b", r"\bchown\b", r"\bln\b", r"\btee\b",
r"\bsed\s+-i\b", r"\bgit\s+checkout\b", r"\bgit\s+restore\b",
r"\bgit\s+clean\b", r"\bgit\s+reset\b", r"\bgit\s+add\b",
r"\bgit\s+commit\b", r"\bgit\s+push\b", r"\bgit\s+fetch\b",
r"\bdotnet\b", r"\bmsbuild\b", r"\bnpm\s+install\b", r"\bnuget\b",
r"\bpip\s+install\b", r">\s*", r">>\s*",
]
# ---------------------------------------------------------------------------
# Tool-Definitionen (OpenAI Function Calling Format)
# ---------------------------------------------------------------------------
TOOLS = [
{
"type": "function",
"function": {
"name": "read_file",
"description": (
"Lies den Inhalt einer Textdatei. Der Pfad ist relativ zum "
"Arbeitsverzeichnis (Root der Codebasis)."
),
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Relativer Pfad zur Datei (z.B. 'src/Program.cs')",
},
},
"required": ["path"],
},
},
},
{
"type": "function",
"function": {
"name": "list_directory",
"description": "Liste den Inhalt eines Verzeichnisses mit Typ-Kennzeichnung.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Relativer Pfad zum Verzeichnis (leer = Root)",
},
},
"required": ["path"],
},
},
},
{
"type": "function",
"function": {
"name": "search_files",
"description": (
"Durchsuche Dateien mit einem Regex-Muster (aehnlich grep -rn). "
"Gibt Treffer mit Dateiname, Zeilennummer und Zeileninhalt zurueck."
),
"parameters": {
"type": "object",
"properties": {
"pattern": {"type": "string", "description": "Regex-Suchmuster"},
"path": {
"type": "string",
"description": "Relativer Pfad zum Startverzeichnis (leer = Root)",
},
"file_pattern": {
"type": "string",
"description": "Dateifilter (z.B. '*.cs'), optional",
},
},
"required": ["pattern"],
},
},
},
{
"type": "function",
"function": {
"name": "execute_command",
"description": (
"Fuehre einen schreibgeschuetzten Shell-Befehl im Arbeitsverzeichnis "
"aus. Schreibende und bauende Kommandos werden abgelehnt."
),
"parameters": {
"type": "object",
"properties": {
"command": {"type": "string", "description": "Der auszufuehrende Befehl"},
},
"required": ["command"],
},
},
},
{
"type": "function",
"function": {
"name": "write_file",
"description": (
"Schreibe eine Ergebnisdatei in das Ausgabeverzeichnis. Der Pfad "
"ist relativ zum Ausgabeverzeichnis (z.B. 'StRS.md')."
),
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "Relativer Pfad"},
"content": {"type": "string", "description": "Vollstaendiger Dateiinhalt"},
},
"required": ["path", "content"],
},
},
},
{
"type": "function",
"function": {
"name": "spawn_subagent",
"description": (
"Starte einen Subagenten mit eigenem Kontext fuer eine isolierte Teilaufgabe. "
"Der Subagent kann Dateien lesen, Verzeichnisse auflisten, suchen und "
"Befehle ausfuehren – aber keine Ergebnisdateien schreiben. Verwende dies, "
"um einen Teil der Codebasis parallel oder isoliert zu analysieren. "
"Der Subagent erhaelt nur die Beschreibung, nicht den bisherigen "
"Konversationsverlauf. Gib eine praegnante Aufgabenbeschreibung."
),
"parameters": {
"type": "object",
"properties": {
"description": {
"type": "string",
"description": "Die Aufgabe fuer den Subagenten (z.B. 'Analysiere alle Berechtigungspruefungen in src/backend/Centron.BL/Security und erstelle eine Zusammenfassung der gefundenen Pruefungen mit Dateipfaden und Methodennamen')",
},
"subagent_type": {
"type": "string",
"description": "Typ des Subagenten: 'explore' fuer Code-Erkundung, 'general-purpose' fuer allgemeine Analyse",
},
},
"required": ["description"],
},
},
},
]
# Read-Only-Tools fuer Subagenten (kein write_file, kein spawn_subagent)
SUBAGENT_TOOLS = [t for t in TOOLS if t["function"]["name"] not in ("write_file", "spawn_subagent")]
# Standard-Subagent-Typen (fuer Modus 'builtin')
BUILTIN_SUBAGENT_PROMPTS = {
"explore": (
"Du bist ein Code-Explorations-Agent. Deine Aufgabe ist es, einen Teil der "
"Codebasis zu untersuchen und eine strukturierte Zusammenfassung deiner "
"Erkenntnisse zurueckzugeben. Nutze die Werkzeuge aktiv, um Dateien zu lesen "
"und zu durchsuchen. Gib am Ende eine kompakte Zusammenfassung mit konkreten "
"Dateipfaden, Klassennamen und Methodennamen zurueck."
),
"general-purpose": (
"Du bist ein Analyse-Agent. Du untersuchst die Codebasis und beantwortest "
"die dir gestellte Aufgabe. Nutze die Werkzeuge aktiv. Gib am Ende eine "
"praegnante Antwort mit konkreten Belegen (Dateipfade, Methodennamen) zurueck."
),
}
# Aktive Subagent-Prompts (wird je Modus gesetzt)
SUBAGENT_SYSTEM_PROMPTS = dict(BUILTIN_SUBAGENT_PROMPTS)
def load_custom_agents(agents_file):
"""
Laedt Agenten-Definitionen aus einer JSON-Datei fuer Modus 'custom'.
Format: { "agent_name": { "description": "...", "prompt": "..." }, ... }
Rueckgabe: dict agent_name -> system_prompt
"""
agents_path = Path(agents_file)
if not agents_path.is_file():
raise FileNotFoundError(f"Agenten-Datei nicht gefunden: {agents_file}")
data = json.loads(agents_path.read_text(encoding="utf-8"))
prompts = {}
for name, spec in data.items():
desc = spec.get("description", "")
prompt = spec.get("prompt", "")
prompts[name] = prompt
return prompts
# ---------------------------------------------------------------------------
# Pfad-Sicherheit
# ---------------------------------------------------------------------------
def safe_join(root: str, rel_path: str) -> Path:
"""Verbindet root und rel_path, verhindert Path-Traversal."""
root_resolved = Path(root).resolve()
target = (root_resolved / rel_path).resolve()
if not str(target).startswith(str(root_resolved)):
raise ValueError(f"Pfad '{rel_path}' verlaesst das Root-Verzeichnis")
return target
# ---------------------------------------------------------------------------
# Tool-Implementierungen
# ---------------------------------------------------------------------------
def tool_read_file(root: str, args: dict) -> str:
path = args.get("path", "")
try:
full = safe_join(root, path)
if not full.is_file():
return f"FEHLER: Datei nicht gefunden: {path}"
content = full.read_text(encoding="utf-8", errors="replace")
if len(content) > 200000:
content = content[:200000] + "\n\n[... Datei abgeschnitten bei 200.000 Zeichen ...]"
return content
except ValueError as e:
return f"FEHLER: {e}"
except Exception as e:
return f"FEHLER beim Lesen von {path}: {e}"
def tool_list_directory(root: str, args: dict) -> str:
path = args.get("path", "")
try:
full = safe_join(root, path) if path else Path(root).resolve()
if not full.is_dir():
return f"FEHLER: Verzeichnis nicht gefunden: {path}"
entries = []
for child in sorted(full.iterdir(), key=lambda c: (c.is_file(), c.name.lower())):
typ = "[DIR] " if child.is_dir() else "[FILE]"
size = ""
if child.is_file():
try:
size = f" ({child.stat().st_size:,} bytes)"
except OSError:
pass
entries.append(f"{typ} {child.name}{size}")
return "\n".join(entries) if entries else "(leeres Verzeichnis)"
except ValueError as e:
return f"FEHLER: {e}"
except Exception as e:
return f"FEHLER beim Auflisten von {path}: {e}"
def tool_search_files(root: str, args: dict) -> str:
pattern = args.get("pattern", "")
path = args.get("path", "")
file_pattern = args.get("file_pattern", "")
if not pattern:
return "FEHLER: Kein Suchmuster angegeben"
try:
regex = re.compile(pattern, re.IGNORECASE)
search_root = safe_join(root, path) if path else Path(root).resolve()
if not search_root.is_dir():
return f"FEHLER: Verzeichnis nicht gefunden: {path}"
results = []
max_results = 500
max_file_size = 5 * 1024 * 1024
for file_path in search_root.rglob("*"):
if not file_path.is_file():
continue
if file_pattern:
import fnmatch
if not fnmatch.fnmatch(file_path.name, file_pattern):
continue
try:
if file_path.stat().st_size > max_file_size:
continue
except OSError:
continue
try:
rel = file_path.relative_to(Path(root).resolve())
except ValueError:
continue
try:
with open(file_path, "r", encoding="utf-8", errors="replace") as f:
for line_no, line in enumerate(f, 1):
if regex.search(line):
results.append(f"{rel}:{line_no}: {line.rstrip()[:300]}")
if len(results) >= max_results:
results.append(f"\n[... Suche bei {max_results} Treffern abgeschnitten ...]")
return "\n".join(results)
except Exception:
continue
return "\n".join(results) if results else "Keine Treffer."
except re.error as e:
return f"FEHLER: Ungueltiges Regex-Muster: {e}"
except ValueError as e:
return f"FEHLER: {e}"
except Exception as e:
return f"FEHLER bei der Suche: {e}"
def tool_execute_command(root: str, args: dict) -> str:
command = args.get("command", "")
if not command:
return "FEHLER: Kein Befehl angegeben"
for pat in DENIED_COMMAND_PATTERNS:
if re.search(pat, command, re.IGNORECASE):
return "ABGELEHNT: Befehl enthaelt verbotenes Muster. Schreibende und bauende Kommandos sind gesperrt."
try:
result = subprocess.run(
command, shell=True, cwd=root, capture_output=True, text=True,
timeout=60, encoding="utf-8", errors="replace",
)
output = result.stdout or ""
if result.stderr:
output += f"\n[STDERR]\n{result.stderr}"
if len(output) > 100000:
output = output[:100000] + "\n\n[... Ausgabe abgeschnitten ...]"
return output.strip() if output.strip() else "(keine Ausgabe)"
except subprocess.TimeoutExpired:
return "FEHLER: Befehl nach 60 Sekunden abgebrochen"
except Exception as e:
return f"FEHLER bei Befehlsausfuehrung: {e}"
def tool_write_file(output_dir: str, args: dict) -> str:
path = args.get("path", "")
content = args.get("content", "")
if not path:
return "FEHLER: Kein Dateipfad angegeben"
# Modell gibt oft "Ergebnisse/<name>" als Pfad – Präfix entfernen
# da output_dir bereits das Ergebnisse-Verzeichnis ist.
path = path.replace("\\", "/")
for prefix in ("Ergebnisse/", "./Ergebnisse/", "ergebnisse/"):
if path.startswith(prefix):
path = path[len(prefix):]
break
try:
base = Path(output_dir).resolve()
target = (base / path).resolve()
if not str(target).startswith(str(base)):
return f"FEHLER: Pfad '{path}' verlaesst das Ausgabeverzeichnis"
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(content, encoding="utf-8")
return f"OK: Datei geschrieben: {path} ({len(content):,} Zeichen)"
except Exception as e:
return f"FEHLER beim Schreiben von {path}: {e}"
def execute_tool(name: str, args: dict, root: str, output_dir: str) -> str:
"""Dispatch eines Tool-Aufrufs."""
dispatch = {
"read_file": lambda a: tool_read_file(root, a),
"list_directory": lambda a: tool_list_directory(root, a),
"search_files": lambda a: tool_search_files(root, a),
"execute_command": lambda a: tool_execute_command(root, a),
"write_file": lambda a: tool_write_file(output_dir, a),
}
handler = dispatch.get(name)
if handler:
return handler(args)
return f"FEHLER: Unbekanntes Werkzeug: {name}"
# ---------------------------------------------------------------------------
# Subagent
# ---------------------------------------------------------------------------
def run_subagent(provider, model, api_key, effort, root, description,
subagent_type, temperature, timeout, max_turns=15):
"""
Startet einen Subagenten mit eigenem Kontext.
Der Subagent erhaelt Read-Only-Tools und eine begrenzte Turn-Anzahl.
Rueckgabe: dict mit result, usage, turns, tool_calls, status.
"""
sys_prompt = SUBAGENT_SYSTEM_PROMPTS.get(
subagent_type, SUBAGENT_SYSTEM_PROMPTS["general-purpose"]
)
messages = [
{"role": "system", "content": sys_prompt},
{"role": "user", "content": description},
]
sub_usage = {"prompt_tokens": 0, "completion_tokens": 0,
"total_tokens": 0, "cached_tokens": 0, "reasoning_tokens": 0}
sub_turns = 0
sub_tool_calls = 0
sub_result = ""
sub_errors = []
while sub_turns < max_turns:
sub_turns += 1
try:
resp = call_api_subagent(provider, model, messages, api_key,
effort, temperature, timeout)
except Exception as e:
sub_errors.append(str(e))
break
u = resp.get("usage", {})
sub_usage["prompt_tokens"] += u.get("prompt_tokens", 0)
sub_usage["completion_tokens"] += u.get("completion_tokens", 0)
sub_usage["total_tokens"] += u.get("total_tokens", 0)
sub_usage["cached_tokens"] += u.get("prompt_tokens_details", {}).get("cached_tokens", 0)
sub_usage["reasoning_tokens"] += u.get("completion_tokens_details", {}).get("reasoning_tokens", 0)
choices = resp.get("choices", [])
if not choices:
sub_errors.append("Keine choices in Subagent-Antwort")
break
msg = choices[0].get("message", {})
messages.append(msg)
content = msg.get("content", "")
if content:
sub_result = content
tool_calls = msg.get("tool_calls", [])
if not tool_calls:
break
for tc in tool_calls:
func = tc.get("function", {})
tname = func.get("name", "")
try:
targs = json.loads(func.get("arguments", "{}"))
except json.JSONDecodeError:
targs = {}
sub_tool_calls += 1
# Subagent darf nur Read-Only-Tools nutzen
if tname in ("read_file", "list_directory", "search_files", "execute_command"):
tresult = execute_tool(tname, targs, root, "")
else:
tresult = f"FEHLER: Werkzeug '{tname}' ist fuer Subagenten nicht freigegeben."
messages.append({"role": "tool", "tool_call_id": tc.get("id", ""),
"name": tname, "content": tresult})
return {
"result": sub_result or "(Subagent ohne Ergebnis)",
"usage": sub_usage,
"turns": sub_turns,
"tool_calls": sub_tool_calls,
"errors": sub_errors,
"status": "completed" if sub_result else "failed",
}
def call_api_subagent(provider, model, messages, api_key, effort, temperature, timeout):
"""API-Aufruf fuer Subagenten (mit SUBAGENT_TOOLS statt TOOLS)."""
url = f"{provider['base_url']}/chat/completions"
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
body = {
"model": model, "messages": messages, "tools": SUBAGENT_TOOLS,
"tool_choice": "auto", "temperature": temperature, "stream": False,
}
model_prefix = model.split("/")[0] if "/" in model else ""
effort_type = MODEL_EFFORT_TYPE.get(model_prefix, "thinking")
effort_val = EFFORT_MAP.get(effort, {}).get(effort_type, "medium")
if effort_type == "thinking":
body["thinking"] = {"type": "enabled", "level": effort_val}
elif effort_type == "reasoning_effort":
body["reasoning_effort"] = effort_val
resp = requests.post(url, headers=headers, json=body,
timeout=timeout if timeout > 0 else 1800)
if resp.status_code != 200:
raise RuntimeError(f"API-Fehler {resp.status_code}: {resp.text[:2000]}")
return resp.json()
# ---------------------------------------------------------------------------
# API-Aufruf
# ---------------------------------------------------------------------------
def call_api(provider, model, messages, api_key, effort, temperature, timeout):
"""Ruft die Chat-Completions-API auf und gibt die JSON-Antwort zurueck."""
url = f"{provider['base_url']}/chat/completions"
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
body = {
"model": model, "messages": messages, "tools": TOOLS,
"tool_choice": "auto", "temperature": temperature, "stream": False,
}
# Effort-Parameter anhand des Modell-Praefixes waehlen
model_prefix = model.split("/")[0] if "/" in model else ""
effort_type = MODEL_EFFORT_TYPE.get(model_prefix, "thinking")
effort_val = EFFORT_MAP.get(effort, {}).get(effort_type, "medium")
if effort_type == "thinking":
body["thinking"] = {"type": "enabled", "level": effort_val}
elif effort_type == "reasoning_effort":
body["reasoning_effort"] = effort_val
resp = requests.post(url, headers=headers, json=body,
timeout=timeout if timeout > 0 else 1800)
if resp.status_code != 200:
raise RuntimeError(f"API-Fehler {resp.status_code}: {resp.text[:2000]}")
return resp.json()
# ---------------------------------------------------------------------------
# Agent-Loop
# ---------------------------------------------------------------------------
def run_agent_loop(provider, model, system_prompt, user_prompt, api_key, effort,
root, output_dir, max_turns, temperature, timeout, mode="solo"):
"""Fuehrt den Agent-Loop durch und sammelt Metriken."""
# Tools je nach Modus waehlen
if mode == "builtin":
active_tools = TOOLS # inklusive spawn_subagent
else:
active_tools = [t for t in TOOLS if t["function"]["name"] != "spawn_subagent"]
# call_api mit den aktiven Tools parametrisieren
def _call_api(messages):
url = f"{provider['base_url']}/chat/completions"
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
body = {
"model": model, "messages": messages, "tools": active_tools,
"tool_choice": "auto", "temperature": temperature, "stream": False,
}
model_prefix = model.split("/")[0] if "/" in model else ""
effort_type = MODEL_EFFORT_TYPE.get(model_prefix, "thinking")
effort_val = EFFORT_MAP.get(effort, {}).get(effort_type, "medium")
if effort_type == "thinking":
body["thinking"] = {"type": "enabled", "level": effort_val}
elif effort_type == "reasoning_effort":
body["reasoning_effort"] = effort_val
resp = requests.post(url, headers=headers, json=body,
timeout=timeout if timeout > 0 else 1800)
if resp.status_code != 200:
raise RuntimeError(f"API-Fehler {resp.status_code}: {resp.text[:2000]}")
return resp.json()
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
]
total_usage = {"prompt_tokens": 0, "completion_tokens": 0,
"total_tokens": 0, "cached_tokens": 0, "reasoning_tokens": 0}
turns = 0
tool_calls_log = []
final_content = ""
model_reported = model
finish_reason = None
errors = []
start_time = time.time()
# Subagent-Tracking
subagent_stats = {"spawned": 0, "completed": 0, "failed": 0, "by_type": {}}
subagent_details = []
MAX_SUBAGENTS = 10 # Hartes Limit: danach wird spawn_subagent verweigert
write_file_count = 0 # Zaehlt write_file-Aufrufe
write_reminder_sent = False # Wurde schon eine Schreib-Erinnerung gesendet?
while turns < max_turns:
turns += 1
# Schreib-Erinnerung: Wenn nach 1/3 der Turns noch kein write_file,
# oder nach 2/3 der Turns weniger als 3 write_file-Aufrufe
if not write_reminder_sent and turns >= max_turns // 3 and write_file_count == 0:
messages.append({
"role": "user",
"content": (
"WICHTIG: Du hast bisher keine Ergebnisdateien geschrieben. "
"Beginne JETZT damit, deine Analyseergebnisse mit write_file "
"in die vorgegebenen Dateien (StRS.md, SyRS.md, SwRS.md, "
"Traceability.md, Hypothesen.md, Glossar.md, Analysebericht.md) "
"zu schreiben. Schreibe nicht weiter Subagenten — formalisiere "
"deine bisherigen Erkenntnisse in Anforderungen."
),
})
write_reminder_sent = True
sys.stderr.write(f"[glm-kimi-adapter] Schreib-Erinnerung gesendet "
f"(Turn {turns}, 0 write_file-Aufrufe)\n")
elif (not write_reminder_sent and turns >= max_turns * 2 // 3
and write_file_count < 3):
messages.append({
"role": "user",
"content": (
f"WICHTIG: Du hast bisher nur {write_file_count} Ergebnisdatei(en) "
f"geschrieben. Es fehlen noch mehrere der 7 vorgegebenen Dateien "
f"(StRS.md, SyRS.md, SwRS.md, Traceability.md, Hypothesen.md, "
f"Glossar.md, Analysebericht.md). Schreibe die fehlenden Dateien "
f"JETZT mit write_file."
),
})
write_reminder_sent = True
sys.stderr.write(f"[glm-kimi-adapter] Schreib-Erinnerung gesendet "
f"(Turn {turns}, {write_file_count} write_file-Aufrufe)\n")
try:
response = _call_api(messages)
except Exception as e:
errors.append(f"Turn {turns}: API-Fehler: {e}")
break
usage = response.get("usage", {})
total_usage["prompt_tokens"] += usage.get("prompt_tokens", 0)
total_usage["completion_tokens"] += usage.get("completion_tokens", 0)
total_usage["total_tokens"] += usage.get("total_tokens", 0)
cached = usage.get("prompt_tokens_details", {}).get("cached_tokens", 0)
total_usage["cached_tokens"] += cached
comp_details = usage.get("completion_tokens_details", {})
total_usage["reasoning_tokens"] += comp_details.get("reasoning_tokens", 0)
if response.get("model"):
model_reported = response["model"]
choices = response.get("choices", [])
if not choices:
errors.append(f"Turn {turns}: Keine choices in API-Antwort")
break
choice = choices[0]
finish_reason = choice.get("finish_reason")
msg = choice.get("message", {})
messages.append(msg)
content = msg.get("content", "")
if content:
final_content = content
tool_calls = msg.get("tool_calls", [])
if not tool_calls:
break
for tc in tool_calls:
func = tc.get("function", {})
tool_name = func.get("name", "")
tool_args_str = func.get("arguments", "{}")
tc_id = tc.get("id", "")
try:
tool_args = json.loads(tool_args_str)
except json.JSONDecodeError:
tool_args = {}
tool_calls_log.append({"turn": turns, "name": tool_name, "args": tool_args})
# write_file-Zaehler erhoehen
if tool_name == "write_file":
write_file_count += 1
# Subagent-Spawning behandeln
if tool_name == "spawn_subagent":
if subagent_stats["spawned"] >= MAX_SUBAGENTS:
# Hartes Limit erreicht: verweigere und erzwinge Schreibphase
result = (
"ABGELEHNT: Subagent-Limit erreicht (10/10). "
"Du hast bereits 10 Subagenten gestartet. "
"Schreibe JETZT deine Ergebnisdateien mit write_file: "
"StRS.md, SyRS.md, SwRS.md, Traceability.md, "
"Hypothesen.md, Glossar.md, Analysebericht.md. "
"Starte keine weiteren Subagenten."
)
sys.stderr.write(f"[glm-kimi-adapter] Subagent verweigert "
f"(Limit {MAX_SUBAGENTS} erreicht)\n")
else:
sa_desc = tool_args.get("description", "")
sa_type = tool_args.get("subagent_type", "general-purpose")
subagent_stats["spawned"] += 1
subagent_stats["by_type"][sa_type] = subagent_stats["by_type"].get(sa_type, 0) + 1
sys.stderr.write(f"[glm-kimi-adapter] Subagent "
f"{subagent_stats['spawned']}/{MAX_SUBAGENTS} "
f"gestartet (Typ: {sa_type})\n")
try:
sa_result = run_subagent(
provider, model, api_key, effort, root,
sa_desc, sa_type, temperature, timeout
)
subagent_stats["completed" if sa_result["status"] == "completed" else "failed"] += 1
sa_u = sa_result["usage"]
total_usage["prompt_tokens"] += sa_u["prompt_tokens"]
total_usage["completion_tokens"] += sa_u["completion_tokens"]
total_usage["total_tokens"] += sa_u["total_tokens"]
total_usage["cached_tokens"] += sa_u["cached_tokens"]
total_usage["reasoning_tokens"] += sa_u["reasoning_tokens"]
subagent_details.append({
"id": subagent_stats["spawned"],
"type": sa_type,
"description": sa_desc[:200],
"turns": sa_result["turns"],
"tool_calls": sa_result["tool_calls"],
"tokens": sa_u["total_tokens"],
"status": sa_result["status"],
})
result = sa_result["result"]
except Exception as e:
subagent_stats["failed"] += 1
result = f"FEHLER: Subagent fehlgeschlagen: {e}"
else:
result = execute_tool(tool_name, tool_args, root, output_dir)
messages.append({"role": "tool", "tool_call_id": tc_id,
"name": tool_name, "content": result})
end_time = time.time()
duration_s = end_time - start_time
written_files = []
if os.path.isdir(output_dir):
for f in sorted(Path(output_dir).rglob("*")):
if f.is_file():
try:
written_files.append({"path": str(f.relative_to(output_dir)),
"size": f.stat().st_size})
except OSError:
pass
tool_call_types = {}
for tc in tool_calls_log:
name = tc["name"]
tool_call_types[name] = tool_call_types.get(name, 0) + 1
return {
"is_error": len(errors) > 0 and not final_content,
"subtype": "success" if final_content else "error",
"duration_ms": int(duration_s * 1000),
"duration_api_ms": int(duration_s * 1000),
"num_turns": turns, "model": model_reported, "model_requested": model,
"provider": provider["__id"],
"usage": {
"prompt_tokens": total_usage["prompt_tokens"],
"completion_tokens": total_usage["completion_tokens"],
"total_tokens": total_usage["total_tokens"],
"cached_tokens": total_usage["cached_tokens"],
"cache_read_tokens": total_usage["cached_tokens"],
"cache_creation_tokens": 0,
"reasoning_tokens": total_usage["reasoning_tokens"],
"output_tokens_details": {
"thinking_tokens": total_usage["reasoning_tokens"],
},
},
"modelUsage": {
model_reported: {
"input_tokens": total_usage["prompt_tokens"],
"output_tokens": total_usage["completion_tokens"],
"cache_read_input_tokens": total_usage["cached_tokens"],
"cache_creation_input_tokens": 0,
"reasoning_tokens": total_usage["reasoning_tokens"],
}
},
"tool_calls": tool_calls_log,
"tool_call_count": len(tool_calls_log),
"tool_call_types": tool_call_types,
"written_files": written_files, "result": final_content,
"finish_reason": finish_reason, "errors": errors, "session_id": "",
"adapter": "python-glm-kimi", "adapter_version": "1.1.0",
"mode": mode,
"subagent_stats": subagent_stats,
"subagent_details": subagent_details,
}
# ---------------------------------------------------------------------------
# Hauptprogramm
# ---------------------------------------------------------------------------
def main():
parser = argparse.ArgumentParser(description="TensorX API Adapter fuer run-experiment (GLM/Kimi)")
parser.add_argument("--prompt", required=True, help="Pfad zur combined_prompt.md")
parser.add_argument("--root", required=True, help="Root-Verzeichnis der Codebasis")
parser.add_argument("--output", required=True, help="Ausgabeverzeichnis (Ergebnisse)")
parser.add_argument("--model", required=True, help="Modell-ID (z.B. z-ai/glm-5.2, moonshotai/kimi-k3)")
parser.add_argument("--provider", default="tensorx", help="API-Provider (default: tensorx)")
parser.add_argument("--api-key", default=None, help="API-Key (default: aus Cline providers.json)")
parser.add_argument("--effort", default="high", choices=["low", "medium", "high", "xhigh", "max"])
parser.add_argument("--mode", default="solo", choices=["solo", "builtin", "custom"], help="Agentenmodus (solo=keine Subagenten, builtin=eingebaute, custom=vordefinierte Agenten aus Datei)")
parser.add_argument("--agents", default=None, help="Pfad zu Agenten-Definitionen (JSON) fuer Modus 'custom'")
parser.add_argument("--max-turns", type=int, default=50)
parser.add_argument("--temperature", type=float, default=1.0)
parser.add_argument("--timeout", type=int, default=0, help="Timeout in Sek (0=keins)")
parser.add_argument("--result-dir", default=None, help="Verzeichnis fuer RawResult.json")
args = parser.parse_args()
provider = PROVIDERS.get(args.provider, PROVIDERS["tensorx"]).copy()
provider["__id"] = args.provider
# API-Key: erst --api-key, dann Cline providers.json, dann Env-Var
api_key = args.api_key
base_url_override = None
if not api_key:
cline_key, cline_url = load_cline_api_key()
if cline_key:
api_key = cline_key
base_url_override = cline_url
sys.stderr.write("[glm-kimi-adapter] API-Key aus Cline providers.json gelesen.\n")
else:
api_key = os.environ.get(provider.get("env_key", ""), "")
if base_url_override:
provider["base_url"] = base_url_override
if not api_key:
sys.stderr.write(
"FEHLER: Kein API-Key gefunden. Weder --api-key, noch Cline providers.json, "
f"noch Umgebungsvariable {provider.get('env_key', '')}.\n"
)
sys.exit(2)
prompt_path = Path(args.prompt)
if not prompt_path.is_file():
sys.stderr.write(f"FEHLER: Prompt-Datei nicht gefunden: {args.prompt}\n")
sys.exit(2)
user_prompt = prompt_path.read_text(encoding="utf-8")
system_prompt = (
"Du bist ein Requirements Engineer im Reverse Requirements Engineering "
"eines Legacy-ERP-Systems. Du analysierst die Codebasis im Arbeitsverzeichnis "
"und erstellst eine Anforderungsspezifikation nach ISO/IEC/IEEE 29148:2018.\n\n"
"Werkzeuge: read_file, list_directory, search_files, execute_command, write_file.\n\n"
"WICHTIG: Beginne deine Arbeit mit dem Schreiben von Ergebnisdateien (write_file), "
"nicht mit dem Erkunden. Erstelle zuerst die Skelette der 7 Ergebnisdateien "
"(StRS.md, SyRS.md, SwRS.md, Traceability.md, Hypothesen.md, Glossar.md, "
"Analysebericht.md) mit Platzhalter-Inhalt, bevor du in die Details gehst. "
"Wenn du Ergebnisse hast, schreibe sie SOFORT mit write_file — beschreibe nicht, "
"was du schreiben wirst, schreibe es."
)
if args.mode == "builtin":
system_prompt += (
"\nZusaetzlich steht spawn_subagent zur Verfuegung: Starte einen "
"Subagenten mit eigenem Kontext fuer isolierte Teilaufgaben (z.B. "
"Analyse eines einzelnen Moduls). Der Subagent kann nur lesen, nicht "
"schreiben. Nutze Subagenten, um die Breite der Analyse zu erhoehen: "
"delegiere Modul-Analysen an Subagenten, waehrend du die "
"Gesamtstruktur und die Ergebnisdateien verwaltest."
)
elif args.mode == "custom":
# Custom agents laden
if args.agents:
custom_prompts = load_custom_agents(args.agents)
SUBAGENT_SYSTEM_PROMPTS.clear()
SUBAGENT_SYSTEM_PROMPTS.update(custom_prompts)
agent_list = ", ".join(SUBAGENT_SYSTEM_PROMPTS.keys())
sys.stderr.write(f"[glm-kimi-adapter] Custom agents geladen: {agent_list}\n")
system_prompt += (
f"\nDu orchestrierst spezialisierte Subagenten. Verfuegbare Agenten-Typen: "
f"{agent_list}. Nutze spawn_subagent mit dem passenden subagent_type, "
f"um Teilaufgaben zu delegieren. Vorgehen: "
f"1. Starte 'modulinventar' fuer das vollstaendige Inventar. "
f"2. Starte 'faktenermittler' fuer Modulausschnitte, die du analysieren willst. "
f"3. Schreibe die Anforderungen selbst mit write_file (die Autoren-Agenten "
f"sind nur fuer Vorbereitung da, nicht fuer das Schreiben der Ergebnisdateien). "
f"4. Fuehre den Konsistenzcheck selbst durch. "
f"Der Subagent kann nur lesen, nicht schreiben."
)
else:
sys.stderr.write("[glm-kimi-adapter] WARNUNG: Modus 'custom' ohne --agents, falle auf 'builtin' zurueck\n")
args.mode = "builtin"
system_prompt += (
"\nZusaetzlich steht spawn_subagent zur Verfuegung."
)
system_prompt += (
"\nDie Codebasis wird ausschliesslich GELESEN. Schreibe Ergebnisdateien mit "
"write_file ins Ausgabeverzeichnis. Sprache: Deutsch fuer Anforderungen."
)
output_dir = Path(args.output).resolve()
output_dir.mkdir(parents=True, exist_ok=True)
result_dir = Path(args.result_dir) if args.result_dir else output_dir.parent
result_dir.mkdir(parents=True, exist_ok=True)
start_iso = datetime.now(timezone.utc).isoformat()
sys.stderr.write(f"[glm-kimi-adapter] Start: {start_iso}\n")
sys.stderr.write(f"[glm-kimi-adapter] Provider: {provider['name']}\n")
sys.stderr.write(f"[glm-kimi-adapter] Modell: {args.model}\n")
sys.stderr.write(f"[glm-kimi-adapter] Effort: {args.effort}\n")
sys.stderr.write(f"[glm-kimi-adapter] Mode: {args.mode}\n")
try:
result = run_agent_loop(
provider=provider, model=args.model, system_prompt=system_prompt,
user_prompt=user_prompt, api_key=api_key, effort=args.effort,
root=args.root, output_dir=str(output_dir), max_turns=args.max_turns,
temperature=args.temperature, timeout=args.timeout, mode=args.mode,
)
except Exception as e:
tb = traceback.format_exc()
sys.stderr.write(f"[glm-kimi-adapter] FEHLER: {e}\n{tb}\n")
result = {
"is_error": True, "subtype": "error", "error": str(e),
"duration_ms": 0, "num_turns": 0, "model": args.model,
"model_requested": args.model, "provider": args.provider,
"usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0,
"cached_tokens": 0, "cache_read_tokens": 0, "cache_creation_tokens": 0,
"reasoning_tokens": 0},
"modelUsage": {}, "tool_calls": [], "tool_call_count": 0,
"tool_call_types": {}, "written_files": [], "result": "",
"errors": [str(e)], "adapter": "python-glm-kimi", "adapter_version": "1.1.0",
"mode": args.mode,
"subagent_stats": {"spawned": 0, "completed": 0, "failed": 0, "by_type": {}},
"subagent_details": [],
}
end_iso = datetime.now(timezone.utc).isoformat()
result["start_time"] = start_iso
result["end_time"] = end_iso
raw_result_path = result_dir / "RawResult.json"
raw_result_path.write_text(json.dumps(result, indent=2, ensure_ascii=False), encoding="utf-8")
sys.stderr.write(f"[glm-kimi-adapter] Ende: {end_iso}\n")
sys.stderr.write(f"[glm-kimi-adapter] Turns: {result['num_turns']}\n")
sys.stderr.write(f"[glm-kimi-adapter] Tokens gesamt: {result['usage']['total_tokens']:,}\n")
sys.stderr.write(f"[glm-kimi-adapter] Tool-Calls: {result['tool_call_count']}\n")
sa = result.get("subagent_stats", {})
sys.stderr.write(f"[glm-kimi-adapter] Subagenten: {sa.get('spawned',0)} (completed: {sa.get('completed',0)}, failed: {sa.get('failed',0)})\n")
sys.stderr.write(f"[glm-kimi-adapter] Ergebnisdateien: {len(result['written_files'])}\n")
sys.stderr.write(f"[glm-kimi-adapter] RawResult: {raw_result_path}\n")
sys.exit(1 if result["is_error"] else 0)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink