Gowler/Masterarbeit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tensorx adapter

Browse Source
This commit is contained in:
Christoph Schwörer
2026-08-28 07:38:12 +02:00
parent ea1f3caff7
commit 37275c96d6
3 changed files with 749 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.claude/skills/run-experiment/glm-kimi-adapter.py
+628
View File
@@ -0,0 +1,628 @@
#!/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"],
},
},
},
]
# ---------------------------------------------------------------------------
# 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,
)
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"
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}"
# ---------------------------------------------------------------------------
# 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):
"""Fuehrt den Agent-Loop durch und sammelt Metriken."""
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()
while turns < max_turns:
turns += 1
try:
response = call_api(provider, model, messages, api_key,
effort, temperature, timeout)
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
# Reasoning/Thinking-Tokens aus completion_tokens_details
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})
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.0.0",
}
# ---------------------------------------------------------------------------
# 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("--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"
"Die 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")
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,
)
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.0.0",
}
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")
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()