7d8c226da2
Admin-Plattform für die Tenants von Onyscribe: Editor mit Live-Vorschau (echter Core in Sandbox), Versionen, KI-Agent (Cortecs), OIDC-Abgleich über die Login-Developer-API, Deploy-Engine (Mittwald, gesperrt ohne DEPLOY_LIVE, Freigabe pro Tenant), Core-Vertrag (Schema/Manifest), Core-Releases mit Kompatibilitätsprüfung, Canary-Rollout und Rollback, Zustandsabgleich. Runbook: docs/RUNBOOK_PARALLELTEST.md Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
130 lines
6.0 KiB
Python
130 lines
6.0 KiB
Python
"""LLM-Backends für den Agenten. Interne Nachrichten im Anthropic-Block-Format:
|
||
assistant: [{"type":"text"|"tool_use",...}] · user: str | [{"type":"tool_result","tool_use_id","content"}]
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import uuid
|
||
from dataclasses import dataclass, field
|
||
|
||
|
||
@dataclass
|
||
class Turn:
|
||
blocks: list[dict]
|
||
usage: dict = field(default_factory=dict)
|
||
|
||
|
||
class MessagesBackend:
|
||
"""Anthropic-Messages-Protokoll – Cortecs bietet es unter https://api.cortecs.ai/v1/messages an."""
|
||
name = "messages"
|
||
|
||
def __init__(self, api_key: str, base_url: str, model: str):
|
||
import anthropic
|
||
root = re.sub(r"/v1/?$", "", base_url.rstrip("/")) # SDK hängt /v1/messages selbst an
|
||
self.client = anthropic.Anthropic(api_key=api_key, auth_token=api_key, base_url=root, max_retries=2, timeout=180)
|
||
self.model = model
|
||
|
||
def step(self, system: str, messages: list[dict], tools: list[dict]) -> Turn:
|
||
r = self.client.messages.create(
|
||
model=self.model, max_tokens=8000, system=system, messages=messages,
|
||
tools=[{"name": t["name"], "description": t["description"], "input_schema": t["schema"]} for t in tools])
|
||
blocks = []
|
||
for b in r.content:
|
||
if b.type == "text":
|
||
blocks.append({"type": "text", "text": b.text})
|
||
elif b.type == "tool_use":
|
||
blocks.append({"type": "tool_use", "id": b.id, "name": b.name, "input": b.input})
|
||
return Turn(blocks, {"in": r.usage.input_tokens, "out": r.usage.output_tokens})
|
||
|
||
|
||
class ChatBackend:
|
||
"""OpenAI-Chat-Completions (Fallback, falls ein Modell nur diesen Weg mit Tools kann)."""
|
||
name = "chat"
|
||
|
||
def __init__(self, api_key: str, base_url: str, model: str):
|
||
from openai import OpenAI
|
||
self.client = OpenAI(api_key=api_key, base_url=base_url, max_retries=2, timeout=180)
|
||
self.model = model
|
||
|
||
@staticmethod
|
||
def _convert(system: str, messages: list[dict]) -> list[dict]:
|
||
out = [{"role": "system", "content": system}]
|
||
for m in messages:
|
||
c = m["content"]
|
||
if isinstance(c, str):
|
||
out.append({"role": m["role"], "content": c})
|
||
elif m["role"] == "assistant":
|
||
text = "".join(b["text"] for b in c if b["type"] == "text")
|
||
calls = [{"id": b["id"], "type": "function", "function": {"name": b["name"], "arguments": json.dumps(b["input"])}}
|
||
for b in c if b["type"] == "tool_use"]
|
||
msg = {"role": "assistant", "content": text or None}
|
||
if calls:
|
||
msg["tool_calls"] = calls
|
||
out.append(msg)
|
||
else:
|
||
for b in c:
|
||
out.append({"role": "tool", "tool_call_id": b["tool_use_id"], "content": b["content"]})
|
||
return out
|
||
|
||
def step(self, system, messages, tools) -> Turn:
|
||
r = self.client.chat.completions.create(
|
||
model=self.model, max_tokens=8000, temperature=0.2, messages=self._convert(system, messages),
|
||
tools=[{"type": "function", "function": {"name": t["name"], "description": t["description"], "parameters": t["schema"]}} for t in tools])
|
||
msg = r.choices[0].message
|
||
blocks = []
|
||
if msg.content:
|
||
blocks.append({"type": "text", "text": msg.content})
|
||
for tc in msg.tool_calls or []:
|
||
try:
|
||
args = json.loads(tc.function.arguments or "{}")
|
||
except json.JSONDecodeError:
|
||
args = {}
|
||
blocks.append({"type": "tool_use", "id": tc.id, "name": tc.function.name, "input": args})
|
||
u = r.usage
|
||
return Turn(blocks, {"in": getattr(u, "prompt_tokens", 0), "out": getattr(u, "completion_tokens", 0)})
|
||
|
||
|
||
HEX = re.compile(r"#[0-9a-fA-F]{6}\b")
|
||
|
||
|
||
def _shade(h, f):
|
||
h = h.lstrip("#"); c = [max(0, min(255, int(int(h[i:i + 2], 16) * f))) for i in (0, 2, 4)]
|
||
return "#%02X%02X%02X" % tuple(c)
|
||
|
||
|
||
def _tint(h, a):
|
||
h = h.lstrip("#"); c = [int(int(h[i:i + 2], 16) + (255 - int(h[i:i + 2], 16)) * a) for i in (0, 2, 4)]
|
||
return "#%02X%02X%02X" % tuple(c)
|
||
|
||
|
||
class MockBackend:
|
||
"""Demo ohne Cortecs-Key: spielt über dieselben Tools zwei einfache Aufgaben durch."""
|
||
name = "mock"
|
||
model = "mock"
|
||
|
||
def step(self, system, messages, tools) -> Turn:
|
||
used = sum(1 for m in messages if m["role"] == "assistant")
|
||
user = next(m["content"] for m in messages if m["role"] == "user" and isinstance(m["content"], str))
|
||
low = user.lower()
|
||
tid = lambda: "mock_" + uuid.uuid4().hex[:8] # noqa: E731
|
||
call = lambda name, **a: {"type": "tool_use", "id": tid(), "name": name, "input": a} # noqa: E731
|
||
hexes = HEX.findall(user)
|
||
if hexes and any(w in low for w in ("farbe", "akzent", "color", "design")):
|
||
a = hexes[0].upper()
|
||
plan = [[call("set_config", key=f"colors.{k}", value=v) for k, v in
|
||
{"accent": a, "accent_dark": _shade(a, .72), "accent_mid": _shade(a, 1.12),
|
||
"accent_light": _tint(a, .88), "accent_xlight": _tint(a, .94)}.items()],
|
||
[call("validate")]]
|
||
final = f"Akzentfarbe auf {a} gesetzt und die abgeleiteten Töne angepasst."
|
||
elif "prompt" in low or "regel" in low:
|
||
rule = re.sub(r"^(bitte\s+)?(füge\s+)?(dem\s+)?prompt\s*(hinzu)?[:,]?\s*", "", user.strip(), flags=re.I)
|
||
plan = [[call("read_core_default", name="prompt.txt")], [call("__append_prompt__", rule=rule)], [call("validate")]]
|
||
final = "Zusatzregel im Prompt ergänzt."
|
||
else:
|
||
return Turn([{"type": "text", "text": "Demo-Modus – es wurde kein Cortecs-Key gefunden (geprüft: Einstellungen → Geteilte Secrets, Env/.env CORTECS_API_KEY, Key des Tenants). "
|
||
"Key setzen und die Seite neu laden. Ohne Key verstehe ich nur „Setze die Akzentfarbe auf #RRGGBB“ und „Füge dem Prompt hinzu: …“."}])
|
||
if used < len(plan):
|
||
return Turn(plan[used])
|
||
return Turn([{"type": "text", "text": final}])
|