diff --git a/core/pyproject.toml b/core/pyproject.toml new file mode 100644 index 0000000..5286ff1 --- /dev/null +++ b/core/pyproject.toml @@ -0,0 +1,40 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "knaster" +version = "0.1.0" +description = "Skill-Destillation von starken zu schwachen LLMs: Router, Verifier, Healer, Pflege-Agent" +requires-python = ">=3.11" +dependencies = [ + "fastapi>=0.110", + "uvicorn[standard]>=0.29", + "httpx>=0.27", + "pydantic>=2.6", + "pyyaml>=6.0", + "sqlalchemy>=2.0", + "jsonschema>=4.21", + "prometheus-client>=0.20", +] + +[project.optional-dependencies] +postgres = ["psycopg[binary]>=3.1"] +dev = ["pytest>=8", "ruff>=0.5", "yamllint>=1.35"] + +[project.scripts] +knaster = "knaster.cli:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-q" + +[tool.ruff] +line-length = 120 +target-version = "py311" + +[tool.ruff.lint] +select = ["E", "F", "I", "B", "UP"] diff --git a/core/src/knaster/__init__.py b/core/src/knaster/__init__.py new file mode 100644 index 0000000..8302ed5 --- /dev/null +++ b/core/src/knaster/__init__.py @@ -0,0 +1,3 @@ +"""knaster – Skill-Destillation von starken zu schwachen LLMs.""" + +__version__ = "0.1.0" diff --git a/core/src/knaster/budget.py b/core/src/knaster/budget.py new file mode 100644 index 0000000..3315ad9 --- /dev/null +++ b/core/src/knaster/budget.py @@ -0,0 +1,36 @@ +"""Token-Budget: Passt Prompt + Ausgabe in den effektiven Kontext eines Slots?""" + +from __future__ import annotations + +import math +from collections.abc import Callable +from dataclasses import dataclass + +MESSAGE_OVERHEAD = 64 + +Tokenizer = Callable[[str], int] + + +def estimate_tokens(text: str) -> int: + """Konservative Schätzung (eher zu hoch): ~3 Zeichen pro Token.""" + return math.ceil(len(text) / 3) + + +@dataclass(frozen=True) +class Budget: + prompt_tokens: int + output_tokens: int + limit: int + + @property + def required(self) -> int: + return self.prompt_tokens + self.output_tokens + MESSAGE_OVERHEAD + + @property + def fits(self) -> bool: + return self.required <= self.limit + + +def compute(system: str, user: str, max_output_tokens: int, limit: int, + tokenizer: Tokenizer = estimate_tokens) -> Budget: + return Budget(tokenizer(system) + tokenizer(user), max_output_tokens, limit) diff --git a/core/src/knaster/calibration.py b/core/src/knaster/calibration.py new file mode 100644 index 0000000..4234754 --- /dev/null +++ b/core/src/knaster/calibration.py @@ -0,0 +1,39 @@ +"""Kalibrierung des effektiven Kontexts aus beobachteten Pass-Rates nach Prompt-Länge.""" + +from __future__ import annotations + +from collections.abc import Iterable + +DEFAULT_BUCKETS = (2048, 4096, 8192, 16384, 32768, 65536, 131072) + + +def effective_context( + observations: Iterable[tuple[int, bool]], + *, + target: float = 0.9, + min_n: int = 20, + buckets: tuple[int, ...] = DEFAULT_BUCKETS, +) -> int | None: + """observations: (prompt_tokens, bestanden). + + Liefert die Obergrenze des größten Buckets, bis zu dem alle Buckets mit genug Daten + die Ziel-Pass-Rate erreichen. Buckets ohne genug Daten unterbrechen die Kette nicht, + zählen aber auch nicht als Nachweis. Ohne Nachweis: None. + """ + counts = {b: [0, 0] for b in buckets} # [n, passed] + for tokens, passed in observations: + for b in buckets: + if tokens <= b: + counts[b][0] += 1 + counts[b][1] += int(passed) + break + result = None + for b in buckets: + n, ok = counts[b] + if n < min_n: + continue + if ok / n >= target: + result = b + else: + break + return result diff --git a/core/src/knaster/cases.py b/core/src/knaster/cases.py new file mode 100644 index 0000000..41fc866 --- /dev/null +++ b/core/src/knaster/cases.py @@ -0,0 +1,44 @@ +"""Fälle und Präferenzpaare: modellneutral speichern, deterministisch splitten, exportieren.""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Iterator +from typing import Any + +from .store import Store + +EVAL_SHARE_PERCENT = 20 + + +def split_for(text: str) -> str: + """Fester Split pro Eingabe: gleiche Eingabe landet immer im selben Split.""" + bucket = int(hashlib.sha256(text.encode("utf-8")).hexdigest(), 16) % 100 + return "eval" if bucket < EVAL_SHARE_PERCENT else "train" + + +def export_sft(store: Store, *, cluster: str, split: str = "train") -> Iterator[str]: + """JSONL für Supervised Finetuning: nur unabhängig verifizierte Fälle.""" + for case in reversed(store.list_cases(cluster=cluster, split=split, passed=True)): + messages = list(case["messages"] or []) + messages.append({"role": "assistant", "content": case["output"]}) + meta = {k: case[k] for k in ("id", "skill", "skill_version", "stage", "model", "origin")} + yield json.dumps({"messages": messages, "meta": meta}, ensure_ascii=False) + + +def export_preferences(store: Store, *, cluster: str, split: str = "train") -> Iterator[str]: + for pair in store.list_preference_pairs(cluster=cluster): + if pair["split"] != split: + continue + yield json.dumps( + {"prompt": pair["prompt"], "chosen": pair["chosen"], "rejected": pair["rejected"]}, + ensure_ascii=False, + ) + + +def pass_rate(rows: list[dict[str, Any]]) -> float | None: + decided = [r for r in rows if r.get("passed") is not None] + if not decided: + return None + return sum(1 for r in decided if r["passed"]) / len(decided) diff --git a/core/src/knaster/classifier.py b/core/src/knaster/classifier.py new file mode 100644 index 0000000..d1f9efc --- /dev/null +++ b/core/src/knaster/classifier.py @@ -0,0 +1,50 @@ +"""Stufe 3: Entscheidungen direkt auf Embeddings (Nearest Centroid mit Konfidenz).""" + +from __future__ import annotations + +import json +import math +from collections import defaultdict +from pathlib import Path + +from .embeddings import Vector, cosine + + +class EmbeddingClassifier: + def __init__(self, centroids: dict[str, Vector], counts: dict[str, int] | None = None): + if not centroids: + raise ValueError("Classifier braucht mindestens eine Klasse") + self.centroids = centroids + self.counts = counts or {} + + @classmethod + def train(cls, examples: list[tuple[Vector, str]]) -> EmbeddingClassifier: + sums: dict[str, Vector] = {} + counts: dict[str, int] = defaultdict(int) + for vec, label in examples: + if label not in sums: + sums[label] = [0.0] * len(vec) + sums[label] = [a + b for a, b in zip(sums[label], vec, strict=True)] + counts[label] += 1 + centroids = {} + for label, total in sums.items(): + norm = math.sqrt(sum(v * v for v in total)) or 1.0 + centroids[label] = [v / norm for v in total] + return cls(centroids, dict(counts)) + + def predict(self, vec: Vector) -> tuple[str, float]: + """Liefert (Label, Konfidenz). Konfidenz = Softmax-Abstand der besten zwei Klassen (0..1).""" + scores = sorted(((cosine(vec, c), label) for label, c in self.centroids.items()), reverse=True) + best_score, best_label = scores[0] + if len(scores) == 1: + return best_label, max(0.0, best_score) + exps = [math.exp(s * 20) for s, _ in scores] # Temperatur 0.05 + return best_label, exps[0] / sum(exps) + + def save(self, path: str | Path) -> None: + Path(path).write_text(json.dumps({"centroids": self.centroids, "counts": self.counts})) + + @classmethod + def load(cls, path: str | Path) -> EmbeddingClassifier: + data = json.loads(Path(path).read_text()) + return cls(data["centroids"], data.get("counts")) diff --git a/core/src/knaster/embeddings.py b/core/src/knaster/embeddings.py new file mode 100644 index 0000000..cf4e483 --- /dev/null +++ b/core/src/knaster/embeddings.py @@ -0,0 +1,136 @@ +"""Embedding- und Rerank-Clients sowie Skill-Index.""" + +from __future__ import annotations + +import hashlib +import math +import re +from collections.abc import Sequence +from typing import Protocol + +import httpx + +Vector = list[float] + + +class Embedder(Protocol): + def embed(self, texts: Sequence[str]) -> list[Vector]: ... + + +class Reranker(Protocol): + def rerank(self, query: str, documents: Sequence[str]) -> list[float]: ... + + +def cosine(a: Sequence[float], b: Sequence[float]) -> float: + dot = sum(x * y for x, y in zip(a, b, strict=True)) + na = math.sqrt(sum(x * x for x in a)) + nb = math.sqrt(sum(y * y for y in b)) + if na == 0 or nb == 0: + return 0.0 + return dot / (na * nb) + + +class HashingEmbedder: + """Deterministischer Bag-of-Words-Embedder für Tests und als Notfall-Fallback ohne GPU/Netz.""" + + _WORD = re.compile(r"\w+", re.UNICODE) + + def __init__(self, dim: int = 512): + self.dim = dim + + def _vector(self, text: str) -> Vector: + vec = [0.0] * self.dim + for word in self._WORD.findall(text.lower()): + h = int.from_bytes(hashlib.sha1(word.encode()).digest()[:4], "big") + vec[h % self.dim] += 1.0 if (h >> 31) & 1 else -1.0 + norm = math.sqrt(sum(v * v for v in vec)) or 1.0 + return [v / norm for v in vec] + + def embed(self, texts: Sequence[str]) -> list[Vector]: + return [self._vector(t) for t in texts] + + +class OpenAIEmbedder: + """Embeddings über ein OpenAI-kompatibles API (LiteLLM → llama.cpp/TEI).""" + + def __init__(self, base_url: str, model: str, api_key: str = "", timeout: float = 30.0, + client: httpx.Client | None = None): + self.base_url = base_url.rstrip("/") + self.model = model + self.headers = {"Authorization": f"Bearer {api_key}"} if api_key else {} + self.client = client or httpx.Client(timeout=timeout) + + def embed(self, texts: Sequence[str]) -> list[Vector]: + resp = self.client.post( + f"{self.base_url}/v1/embeddings", + json={"model": self.model, "input": list(texts)}, + headers=self.headers, + ) + resp.raise_for_status() + data = sorted(resp.json()["data"], key=lambda d: d["index"]) + return [d["embedding"] for d in data] + + +class LlamaCppReranker: + """Reranking über den llama.cpp-Server (`--reranking`, Endpoint /v1/rerank).""" + + def __init__(self, base_url: str, model: str = "rerank-default", timeout: float = 30.0, + client: httpx.Client | None = None): + self.base_url = base_url.rstrip("/") + self.model = model + self.client = client or httpx.Client(timeout=timeout) + + def rerank(self, query: str, documents: Sequence[str]) -> list[float]: + resp = self.client.post( + f"{self.base_url}/v1/rerank", + json={"model": self.model, "query": query, "documents": list(documents)}, + ) + resp.raise_for_status() + scores = [0.0] * len(documents) + for item in resp.json()["results"]: + scores[item["index"]] = float(item["relevance_score"]) + return scores + + +class SkillIndex: + """Hält pro Skill die Vektoren von Beschreibung und Triggern (wenige hundert Skills → im RAM).""" + + def __init__(self, embedder: Embedder): + self.embedder = embedder + self._cache: dict[str, Vector] = {} + self._entries: dict[str, list[Vector]] = {} + + def _embed_cached(self, texts: list[str]) -> list[Vector]: + missing = [t for t in texts if t not in self._cache] + if missing: + for text, vec in zip(missing, self.embedder.embed(missing), strict=True): + self._cache[text] = vec + return [self._cache[t] for t in texts] + + def rebuild(self, skills: dict[str, list[str]]) -> None: + """skills: name -> Texte (Beschreibung + Trigger).""" + all_texts = sorted({t for texts in skills.values() for t in texts}) + self._embed_cached(all_texts) + self._entries = {name: [self._cache[t] for t in texts] for name, texts in skills.items()} + + def vector(self, name: str) -> Vector: + return self._entries[name][0] + + def search(self, query_vec: Vector, k: int = 5, only: set[str] | None = None) -> list[tuple[str, float]]: + scored = [] + for name, vecs in self._entries.items(): + if only is not None and name not in only: + continue + scored.append((name, max(cosine(query_vec, v) for v in vecs))) + scored.sort(key=lambda x: (-x[1], x[0])) + return scored[:k] + + def pairwise_similar(self, threshold: float) -> list[tuple[str, str, float]]: + names = sorted(self._entries) + pairs = [] + for i, a in enumerate(names): + for b in names[i + 1:]: + sim = cosine(self.vector(a), self.vector(b)) + if sim >= threshold: + pairs.append((a, b, sim)) + return pairs diff --git a/core/src/knaster/executor.py b/core/src/knaster/executor.py new file mode 100644 index 0000000..a46807b --- /dev/null +++ b/core/src/knaster/executor.py @@ -0,0 +1,97 @@ +"""Ausführung einer RouteDecision auf Stufe 0–3 und Erkennung von Laufzeit-Signalen.""" + +from __future__ import annotations + +from typing import Any + +from pydantic import BaseModel, Field + +from . import programs +from .classifier import EmbeddingClassifier +from .embeddings import Embedder +from .llm import ChatClient, ContextOverflowError, LLMError +from .problems import Signal, extract_json, is_degenerate +from .router import RouteDecision, system_prompt +from .skills import Skill, SkillLibrary + + +class ExecutionResult(BaseModel): + ok: bool + output: str = "" + signals: list[Signal] = Field(default_factory=list) + messages: list[dict[str, str]] = Field(default_factory=list) + finish_reason: str | None = None + usage: dict[str, Any] = Field(default_factory=dict) + error: str | None = None + label: str | None = None + confidence: float | None = None + + +def build_messages(skill: Skill | None, decision: RouteDecision, text: str) -> list[dict[str, str]]: + system = system_prompt(skill, decision.prompt_variant) + messages = [] + if system: + messages.append({"role": "system", "content": system}) + messages.append({"role": "user", "content": text}) + return messages + + +class Executor: + def __init__(self, library: SkillLibrary, chat: ChatClient, embedder: Embedder, + classifiers: dict[str, EmbeddingClassifier] | None = None): + self.library = library + self.chat = chat + self.embedder = embedder + self.classifiers = classifiers or {} + + def execute(self, decision: RouteDecision, text: str) -> ExecutionResult: + if decision.status != "routed": + return ExecutionResult(ok=False, error=f"Entscheidung nicht ausführbar: {decision.status}") + skill = self.library.get(decision.skill) if decision.skill else None + if decision.stage == 3: + return self._classify(skill, text) + if decision.stage == 2: + return self._program(skill, text) + return self._llm(skill, decision, text) + + def _classify(self, skill: Skill | None, text: str) -> ExecutionResult: + assert skill and skill.meta.classifier + clf = self.classifiers.get(skill.meta.classifier) + if clf is None: + return ExecutionResult(ok=False, signals=[Signal.PROGRAM_ERROR], + error=f"Classifier {skill.meta.classifier} nicht geladen") + label, confidence = clf.predict(self.embedder.embed([text])[0]) + if confidence < skill.meta.min_confidence: + return ExecutionResult(ok=False, output=label, label=label, confidence=confidence, + signals=[Signal.CLASSIFIER_LOW_CONFIDENCE]) + return ExecutionResult(ok=True, output=label, label=label, confidence=confidence) + + def _program(self, skill: Skill | None, text: str) -> ExecutionResult: + assert skill and skill.meta.program + try: + output = programs.get(skill.meta.program)(text) + except Exception as exc: # noqa: BLE001 – Programmfehler sind ein Signal, kein Absturz + return ExecutionResult(ok=False, signals=[Signal.PROGRAM_ERROR], error=f"{type(exc).__name__}: {exc}") + return ExecutionResult(ok=True, output=output) + + def _llm(self, skill: Skill | None, decision: RouteDecision, text: str) -> ExecutionResult: + messages = build_messages(skill, decision, text) + try: + res = self.chat.chat(decision.model or "", messages, max_tokens=decision.max_output_tokens) + except ContextOverflowError as exc: + return ExecutionResult(ok=False, messages=messages, signals=[Signal.CONTEXT_OVERFLOW], error=str(exc)) + except LLMError as exc: + return ExecutionResult(ok=False, messages=messages, signals=[Signal.ENDPOINT_ERROR], error=str(exc)) + + signals: list[Signal] = [] + if res.finish_reason == "length": + signals.append(Signal.OUTPUT_TRUNCATED) + if is_degenerate(res.content): + signals.append(Signal.DEGENERATE_OUTPUT) + if skill is not None and skill.meta.output.format == "json": + try: + extract_json(res.content) + except ValueError: + signals.append(Signal.INVALID_FORMAT) + return ExecutionResult(ok=not signals, output=res.content, signals=signals, messages=messages, + finish_reason=res.finish_reason, usage=res.usage) diff --git a/core/src/knaster/llm.py b/core/src/knaster/llm.py new file mode 100644 index 0000000..bd20698 --- /dev/null +++ b/core/src/knaster/llm.py @@ -0,0 +1,71 @@ +"""Chat-Client für das OpenAI-kompatible API von LiteLLM.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Protocol + +import httpx + +from .problems import is_context_overflow_message + + +class LLMError(Exception): + """Basisklasse; `infra=True` bedeutet: Endpunkt/Netz, nicht Modellqualität.""" + + infra = True + + +class EndpointError(LLMError): + pass + + +class ContextOverflowError(LLMError): + pass + + +@dataclass +class ChatResult: + content: str + finish_reason: str | None = None + usage: dict[str, Any] = field(default_factory=dict) + model: str | None = None + + +class ChatClient(Protocol): + def chat(self, model: str, messages: list[dict[str, str]], *, max_tokens: int | None = None, + temperature: float | None = None) -> ChatResult: ... + + +class OpenAIChatClient: + def __init__(self, base_url: str, api_key: str = "", timeout: float = 120.0, + client: httpx.Client | None = None): + self.base_url = base_url.rstrip("/") + self.headers = {"Authorization": f"Bearer {api_key}"} if api_key else {} + self.client = client or httpx.Client(timeout=timeout) + + def chat(self, model: str, messages: list[dict[str, str]], *, max_tokens: int | None = None, + temperature: float | None = None) -> ChatResult: + body: dict[str, Any] = {"model": model, "messages": messages} + if max_tokens is not None: + body["max_tokens"] = max_tokens + if temperature is not None: + body["temperature"] = temperature + try: + resp = self.client.post(f"{self.base_url}/v1/chat/completions", json=body, headers=self.headers) + except httpx.HTTPError as exc: + raise EndpointError(f"{type(exc).__name__}: {exc}") from exc + if resp.status_code == 400 and is_context_overflow_message(resp.text): + raise ContextOverflowError(resp.text[:500]) + if resp.status_code == 429 or resp.status_code >= 500: + raise EndpointError(f"HTTP {resp.status_code}: {resp.text[:300]}") + if resp.status_code >= 400: + raise EndpointError(f"HTTP {resp.status_code}: {resp.text[:300]}") + data = resp.json() + choice = data["choices"][0] + return ChatResult( + content=choice["message"].get("content") or "", + finish_reason=choice.get("finish_reason"), + usage=data.get("usage") or {}, + model=data.get("model"), + ) diff --git a/core/src/knaster/problems.py b/core/src/knaster/problems.py new file mode 100644 index 0000000..b7e3c1d --- /dev/null +++ b/core/src/knaster/problems.py @@ -0,0 +1,71 @@ +"""Signale (Fehlerklassen) und ihre Erkennung.""" + +from __future__ import annotations + +import json +import re +from enum import StrEnum + + +class Signal(StrEnum): + NO_SKILL = "no_skill" + SKILL_CONFLICT = "skill_conflict" + SLOT_UNAVAILABLE = "slot_unavailable" + CONTEXT_EXCEEDED = "context_exceeded" # präventiv vom Router erkannt + COMPACT_PROMPT = "compact_prompt" # Ausweichen auf kompakte Skill-Variante + LARGER_SLOT = "larger_slot" # Ausweichen auf Slot mit mehr Kontext + NEEDS_SPLIT = "needs_split" + OUTPUT_TRUNCATED = "output_truncated" + DEGENERATE_OUTPUT = "degenerate_output" + INVALID_FORMAT = "invalid_format" + CONTEXT_OVERFLOW = "context_overflow" # Engine meldet Overflow + ENDPOINT_ERROR = "endpoint_error" + VERIFIER_FAILED = "verifier_failed" + CLASSIFIER_LOW_CONFIDENCE = "classifier_low_confidence" + PROGRAM_ERROR = "program_error" + MISROUTE_SUSPECTED = "misroute_suspected" + + +INFRA_SIGNALS = {Signal.ENDPOINT_ERROR, Signal.CONTEXT_OVERFLOW} +QUALITY_SIGNALS = { + Signal.VERIFIER_FAILED, + Signal.INVALID_FORMAT, + Signal.DEGENERATE_OUTPUT, + Signal.OUTPUT_TRUNCATED, + Signal.CLASSIFIER_LOW_CONFIDENCE, + Signal.PROGRAM_ERROR, +} + +_OVERFLOW_PATTERNS = re.compile( + r"context length|maximum context|too many tokens|context window|prompt is too long", re.IGNORECASE +) +_WORD = re.compile(r"\w+", re.UNICODE) + + +def is_context_overflow_message(text: str) -> bool: + return bool(_OVERFLOW_PATTERNS.search(text or "")) + + +def repetition_ratio(text: str, n: int = 4) -> float: + """Anteil der n-Gramme, die Wiederholungen sind (0 = keine).""" + words = _WORD.findall(text.lower()) + if len(words) < n: + return 0.0 + grams = [tuple(words[i:i + n]) for i in range(len(words) - n + 1)] + return 1.0 - len(set(grams)) / len(grams) + + +def is_degenerate(text: str, min_words: int = 40, max_ratio: float = 0.5) -> bool: + return len(_WORD.findall(text)) >= min_words and repetition_ratio(text) > max_ratio + + +def extract_json(text: str) -> object: + """Parst JSON, toleriert ```json-Codeblöcke. Wirft ValueError bei ungültigem JSON.""" + stripped = text.strip() + fence = re.match(r"^```(?:json)?\s*\n(.*?)\n```\s*$", stripped, re.DOTALL) + if fence: + stripped = fence.group(1) + try: + return json.loads(stripped) + except json.JSONDecodeError as exc: + raise ValueError(str(exc)) from exc diff --git a/core/src/knaster/programs.py b/core/src/knaster/programs.py new file mode 100644 index 0000000..11106bf --- /dev/null +++ b/core/src/knaster/programs.py @@ -0,0 +1,33 @@ +"""Stufe 2: registrierte, geprüfte Programme (vom starken LLM erzeugt, vom Menschen freigegeben).""" + +from __future__ import annotations + +from collections.abc import Callable + +Program = Callable[[str], str] + +_REGISTRY: dict[str, Program] = {} + + +def register(name: str) -> Callable[[Program], Program]: + def deco(fn: Program) -> Program: + if name in _REGISTRY: + raise ValueError(f"Programm {name} bereits registriert") + _REGISTRY[name] = fn + return fn + + return deco + + +def get(name: str) -> Program: + return _REGISTRY[name] + + +def names() -> list[str]: + return sorted(_REGISTRY) + + +@register("echo-upper") +def _echo_upper(text: str) -> str: + """Beispielprogramm für Tests.""" + return text.upper() diff --git a/core/src/knaster/registry.py b/core/src/knaster/registry.py new file mode 100644 index 0000000..f096b6e --- /dev/null +++ b/core/src/knaster/registry.py @@ -0,0 +1,172 @@ +"""Registry: deklarative Slots/Adapter (YAML) plus Laufzeitzustand (Store).""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any, Literal + +from pydantic import BaseModel, Field + +from .store import Store +from .util import load_yaml + +SlotStatus = Literal["ok", "degraded", "unavailable"] + + +class SlotSpec(BaseModel): + name: str + tier: int + device: Literal["gpu", "cpu", "external"] = "gpu" + engine: str = "vllm" + node: str | None = None + litellm_model: str + base_model: str + max_context: int + effective_context: int | None = None + supports_lora: bool = False + fallbacks: list[str] = Field(default_factory=list) + kind: Literal["chat", "embedding", "rerank"] = "chat" + + +class AdapterSpec(BaseModel): + name: str + base_model: str + cluster: str + version: int = 1 + status: Literal["active", "candidate", "retired"] = "active" + + +class Slot(BaseModel): + """Effektive Sicht: Spezifikation + Laufzeitzustand.""" + + spec: SlotSpec + status: SlotStatus = "ok" + reason: str | None = None + set_by: str | None = None + calibrated_context: int | None = None + base_model_override: str | None = None + + @property + def name(self) -> str: + return self.spec.name + + @property + def base_model(self) -> str: + return self.base_model_override or self.spec.base_model + + @property + def effective_context(self) -> int: + for value in (self.calibrated_context, self.spec.effective_context): + if value: + return min(value, self.spec.max_context) + return self.spec.max_context + + @property + def available(self) -> bool: + return self.status == "ok" + + +class Registry: + def __init__(self, slots: dict[str, SlotSpec], adapters: list[AdapterSpec], store: Store): + self._slots = slots + self._adapters = {a.name: a for a in adapters} + self.store = store + + @classmethod + def from_file(cls, path: str | Path, store: Store) -> Registry: + data = load_yaml(path) + return cls.from_dict(data, store) + + @classmethod + def from_dict(cls, data: dict[str, Any], store: Store) -> Registry: + slots = {name: SlotSpec(name=name, **spec) for name, spec in (data.get("slots") or {}).items()} + adapters = [AdapterSpec(**a) for a in data.get("adapters") or []] + for spec in slots.values(): + for fb in spec.fallbacks: + if fb not in slots: + raise ValueError(f"Slot {spec.name}: unbekannter Fallback {fb}") + return cls(slots, adapters, store) + + # --- Slots ------------------------------------------------------------------------------ + def slot_names(self) -> list[str]: + return list(self._slots) + + def has_slot(self, name: str) -> bool: + return name in self._slots + + def slot(self, name: str) -> Slot: + spec = self._slots[name] + state = self.store.get_slot_state(name) or {} + return Slot( + spec=spec, + status=state.get("status") or "ok", + reason=state.get("reason"), + set_by=state.get("set_by"), + calibrated_context=state.get("effective_context"), + base_model_override=state.get("base_model"), + ) + + def slots(self) -> list[Slot]: + return [self.slot(n) for n in self._slots] + + def strong_slot(self) -> Slot: + chat = [s for s in self.slots() if s.spec.kind == "chat" and s.spec.tier == 0] + if not chat: + raise LookupError("Kein Slot mit tier 0 konfiguriert") + available = [s for s in chat if s.available] + return (available or chat)[0] + + def set_status(self, name: str, status: SlotStatus, reason: str = "", set_by: str = "human") -> None: + if name not in self._slots: + raise KeyError(name) + self.store.set_slot_state(name, status=status, reason=reason, set_by=set_by) + + def set_calibrated_context(self, name: str, tokens: int | None) -> None: + self.store.set_slot_state(name, effective_context=tokens) + + def set_base_model(self, name: str, base_model: str | None) -> None: + self.store.set_slot_state(name, base_model=base_model) + + # --- Adapter ---------------------------------------------------------------------------- + def adapters(self) -> list[AdapterSpec]: + merged = {n: a.model_copy() for n, a in self._adapters.items()} + for row in self.store.all_adapter_states(): + if row["name"] in merged: + if row.get("status"): + merged[row["name"]].status = row["status"] + else: + merged[row["name"]] = AdapterSpec( + name=row["name"], + base_model=row["base_model"], + cluster=row["cluster"], + version=row.get("version") or 1, + status=row.get("status") or "candidate", + ) + return list(merged.values()) + + def active_adapter(self, base_model: str, cluster: str) -> AdapterSpec | None: + found = [ + a for a in self.adapters() + if a.base_model == base_model and a.cluster == cluster and a.status == "active" + ] + return max(found, key=lambda a: a.version) if found else None + + def set_adapter(self, adapter: AdapterSpec) -> None: + self.store.set_adapter_state( + adapter.name, + base_model=adapter.base_model, + cluster=adapter.cluster, + version=adapter.version, + status=adapter.status, + ) + + def export(self) -> dict[str, Any]: + """Effektiver Stand als YAML-fähiges Dict (für Rückführung nach Git).""" + slots = {} + for s in self.slots(): + data = s.spec.model_dump(exclude={"name"}) + data["base_model"] = s.base_model + if s.calibrated_context: + data["effective_context"] = s.calibrated_context + slots[s.name] = data + return {"slots": slots, "adapters": [a.model_dump() for a in self.adapters()]} diff --git a/core/src/knaster/router.py b/core/src/knaster/router.py new file mode 100644 index 0000000..556daac --- /dev/null +++ b/core/src/knaster/router.py @@ -0,0 +1,276 @@ +"""Router: Anfrage → Skill → Stufe → Slot/Modell (+Adapter), inkl. Token-Budget und Eskalation.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal + +from pydantic import BaseModel, Field + +from . import budget as budget_mod +from .embeddings import Embedder, Reranker, SkillIndex +from .problems import INFRA_SIGNALS, Signal +from .registry import Registry, Slot +from .skills import Skill, SkillLibrary + +PromptVariant = Literal["full", "compact", "none"] + +FORMAT_HINT_JSON = "Antworte ausschließlich mit gültigem JSON ohne weiteren Text." + + +class RouteDecision(BaseModel): + status: Literal["routed", "failed", "too_large"] = "routed" + stage: int = 0 + slot: str | None = None + model: str | None = None + skill: str | None = None + skill_version: int | None = None + adapter: str | None = None + prompt_variant: PromptVariant = "none" + similarity: float | None = None + estimated_tokens: int | None = None + max_output_tokens: int | None = None + attempt: int = 0 + tried_slots: list[str] = Field(default_factory=list) + reasons: list[str] = Field(default_factory=list) + signals: list[Signal] = Field(default_factory=list) + + +@dataclass +class RouterConfig: + top_k: int = 5 + conflict_margin: float = 0.02 + max_escalations: int = 3 + default_max_output_tokens: int = 1024 + + +def system_prompt(skill: Skill | None, variant: PromptVariant) -> str: + parts: list[str] = [] + if skill is not None and variant != "none": + parts.append(skill.prompt(variant)) + if skill is not None and skill.meta.output.format == "json": + parts.append(FORMAT_HINT_JSON) + return "\n\n".join(parts) + + +class Router: + def __init__( + self, + registry: Registry, + library: SkillLibrary, + embedder: Embedder, + reranker: Reranker | None = None, + tokenizer: budget_mod.Tokenizer = budget_mod.estimate_tokens, + config: RouterConfig | None = None, + ): + self.registry = registry + self.library = library + self.embedder = embedder + self.reranker = reranker + self.tokenizer = tokenizer + self.config = config or RouterConfig() + self.index = SkillIndex(embedder) + self.refresh() + + def refresh(self) -> None: + """Index neu aufbauen (nach Skill-Änderungen).""" + self.index.rebuild({s.name: s.index_texts() for s in self.library.all()}) + + # --- Routing ---------------------------------------------------------------------------- + def route(self, text: str, hint_skill: str | None = None) -> RouteDecision: + reasons: list[str] = [] + signals: list[Signal] = [] + routable = {s.name for s in self.library.routable()} + + skill: Skill | None = None + similarity: float | None = None + if hint_skill: + if hint_skill not in routable: + raise KeyError(f"Skill {hint_skill} unbekannt oder nicht routbar") + skill = self.library.get(hint_skill) + reasons.append(f"Skill per Hinweis: {hint_skill}") + else: + skill, similarity = self._match(text, routable, reasons, signals) + + if skill is None: + return self._plan_strong(text, None, reasons, signals, similarity=similarity) + + meta = skill.meta + if meta.status == "demoted" or meta.stage == 0: + reasons.append(f"Skill {skill.name} ist auf Stufe 0 ({meta.status})") + return self._plan_strong(text, skill, reasons, signals, similarity=similarity) + + if meta.execution == "classifier" and meta.stage >= 3 and meta.classifier: + return self._decision(stage=3, skill=skill, similarity=similarity, reasons=reasons, + signals=signals) + if meta.execution == "program" and meta.stage >= 2 and meta.program: + return self._decision(stage=2, skill=skill, similarity=similarity, reasons=reasons, + signals=signals) + return self._plan_weak(text, skill, reasons, signals, similarity=similarity) + + def _match(self, text: str, routable: set[str], reasons: list[str], + signals: list[Signal]) -> tuple[Skill | None, float | None]: + if not routable: + reasons.append("Keine routbaren Skills") + signals.append(Signal.NO_SKILL) + return None, None + qvec = self.embedder.embed([text])[0] + hits = self.index.search(qvec, k=self.config.top_k, only=routable) + if not hits: + signals.append(Signal.NO_SKILL) + return None, None + similarity_of = dict(hits) + + ordered = hits + if self.reranker is not None and len(hits) > 1: + docs = [self.library.get(name).description for name, _ in hits] + scores = self.reranker.rerank(text, docs) + ordered = sorted(zip([h[0] for h in hits], scores, strict=True), key=lambda x: -x[1]) + reasons.append("Reranking aktiv") + + best_name = ordered[0][0] + best_sim = similarity_of[best_name] + skill = self.library.get(best_name) + if best_sim < skill.meta.threshold: + reasons.append(f"Beste Ähnlichkeit {best_sim:.3f} < Schwelle {skill.meta.threshold} ({best_name})") + signals.append(Signal.NO_SKILL) + return None, best_sim + + if len(hits) > 1: + second_name, second_sim = hits[1] if hits[0][0] == best_name else hits[0] + if abs(best_sim - second_sim) <= self.config.conflict_margin: + signals.append(Signal.SKILL_CONFLICT) + reasons.append(f"Konflikt {best_name} ({best_sim:.3f}) vs. {second_name} ({second_sim:.3f})") + if self.reranker is None: + return None, best_sim + reasons.append(f"Skill {best_name} (Ähnlichkeit {best_sim:.3f})") + return skill, best_sim + + # --- Planung ---------------------------------------------------------------------------- + def _decision(self, **kw) -> RouteDecision: + skill: Skill | None = kw.pop("skill", None) + if skill is not None: + kw.setdefault("skill", skill.name) + kw.setdefault("skill_version", skill.meta.version) + kw.setdefault("max_output_tokens", skill.meta.max_output_tokens) + return RouteDecision(**kw) + + def _fit(self, text: str, skill: Skill | None, slots: list[Slot] + ) -> tuple[Slot, PromptVariant, budget_mod.Budget] | None: + max_out = skill.meta.max_output_tokens if skill else self.config.default_max_output_tokens + variants: list[PromptVariant] = ["full", "compact"] if skill else ["none"] + for slot in slots: + for variant in variants: + b = budget_mod.compute(system_prompt(skill, variant), text, max_out, + slot.effective_context, self.tokenizer) + if b.fits: + return slot, variant, b + return None + + def _weak_slots(self, skill: Skill, exclude: set[str]) -> tuple[list[Slot], bool]: + names = [skill.meta.slot, *self._fallbacks(skill.meta.slot)] + slots = [self.registry.slot(n) for n in names if self.registry.has_slot(n) and n not in exclude] + primary_ok = bool(slots) and slots[0].name == skill.meta.slot and slots[0].available + return [s for s in slots if s.available], primary_ok + + def _fallbacks(self, name: str) -> list[str]: + if not self.registry.has_slot(name): + return [] + return list(self.registry.slot(name).spec.fallbacks) + + def _plan_weak(self, text: str, skill: Skill, reasons: list[str], signals: list[Signal], *, + similarity: float | None, attempt: int = 0, tried: list[str] | None = None + ) -> RouteDecision: + tried = list(tried or []) + slots, primary_ok = self._weak_slots(skill, set(tried)) + if not primary_ok and skill.meta.slot not in tried: + signals.append(Signal.SLOT_UNAVAILABLE) + reasons.append(f"Slot {skill.meta.slot} nicht verfügbar") + if not slots: + reasons.append("Kein verfügbarer Slot für Stufe 1") + return self._plan_strong(text, skill, reasons, signals, similarity=similarity, + attempt=attempt, tried=tried) + fit = self._fit(text, skill, slots) + if fit is None: + signals.append(Signal.CONTEXT_EXCEEDED) + reasons.append("Token-Budget passt in keinen Stufe-1-Slot") + return self._plan_strong(text, skill, reasons, signals, similarity=similarity, + attempt=attempt, tried=tried) + slot, variant, b = fit + if variant == "compact": + signals.append(Signal.COMPACT_PROMPT) + if slot.name != slots[0].name: + signals.append(Signal.LARGER_SLOT) + reasons.append(f"Ausweichen auf {slot.name} wegen Kontextlänge") + + model, adapter = slot.spec.litellm_model, None + if slot.spec.supports_lora and skill.meta.adapter == "auto": + found = self.registry.active_adapter(slot.base_model, skill.cluster) + if found: + model, adapter = found.name, found.name + reasons.append(f"Adapter {found.name}") + return self._decision(stage=1, slot=slot.name, model=model, adapter=adapter, skill=skill, + prompt_variant=variant, similarity=similarity, + estimated_tokens=b.prompt_tokens, attempt=attempt, + tried_slots=tried + [slot.name], reasons=reasons, signals=signals) + + def _plan_strong(self, text: str, skill: Skill | None, reasons: list[str], signals: list[Signal], *, + similarity: float | None = None, attempt: int = 0, tried: list[str] | None = None + ) -> RouteDecision: + tried = list(tried or []) + strong = [s for s in self.registry.slots() + if s.spec.kind == "chat" and s.spec.tier == 0 and s.name not in tried] + available = [s for s in strong if s.available] or strong + if not available: + reasons.append("Kein Stufe-0-Slot mehr übrig") + return self._decision(status="failed", stage=0, skill=skill, similarity=similarity, + attempt=attempt, tried_slots=tried, reasons=reasons, signals=signals) + fit = self._fit(text, skill, available) + if fit is None and skill is not None: + # notfalls ohne Skill-Text + fit = self._fit(text, None, available) + if fit is None: + signals.append(Signal.NEEDS_SPLIT) + reasons.append("Eingabe passt in keinen Slot – Aufgabe muss geteilt werden") + return self._decision(status="too_large", stage=0, skill=skill, similarity=similarity, + attempt=attempt, tried_slots=tried, reasons=reasons, signals=signals) + slot, variant, b = fit + return self._decision(stage=0, slot=slot.name, model=slot.spec.litellm_model, skill=skill, + prompt_variant=variant, similarity=similarity, + estimated_tokens=b.prompt_tokens, attempt=attempt, + tried_slots=tried + [slot.name], reasons=reasons, signals=signals) + + # --- Eskalation ------------------------------------------------------------------------- + def escalate(self, decision: RouteDecision, signal: Signal, text: str) -> RouteDecision: + attempt = decision.attempt + 1 + reasons = [f"Eskalation {attempt} wegen {signal.value} (Stufe {decision.stage})"] + signals: list[Signal] = [signal] + skill = self.library.get(decision.skill) if decision.skill and self.library.has(decision.skill) else None + + if attempt > self.config.max_escalations: + reasons.append("Maximale Eskalationen erreicht") + return self._decision(status="failed", stage=decision.stage, skill=skill, attempt=attempt, + tried_slots=decision.tried_slots, reasons=reasons, signals=signals) + + infra = signal in INFRA_SIGNALS + if decision.stage == 0: + if infra: + return self._plan_strong(text, skill, reasons, signals, similarity=decision.similarity, + attempt=attempt, tried=decision.tried_slots) + reasons.append("Stufe 0 bereits erreicht") + return self._decision(status="failed", stage=0, skill=skill, attempt=attempt, + tried_slots=decision.tried_slots, reasons=reasons, signals=signals) + + if decision.stage == 1 and skill is not None: + if infra: + return self._plan_weak(text, skill, reasons, signals, similarity=decision.similarity, + attempt=attempt, tried=decision.tried_slots) + return self._plan_strong(text, skill, reasons, signals, similarity=decision.similarity, + attempt=attempt) + + # Stufe 2/3 → Stufe 1 (falls erlaubt) oder 0 + if skill is not None and skill.meta.llm_fallback: + return self._plan_weak(text, skill, reasons, signals, similarity=decision.similarity, + attempt=attempt) + return self._plan_strong(text, skill, reasons, signals, similarity=decision.similarity, + attempt=attempt) diff --git a/core/src/knaster/runner.py b/core/src/knaster/runner.py new file mode 100644 index 0000000..6837cf0 --- /dev/null +++ b/core/src/knaster/runner.py @@ -0,0 +1,157 @@ +"""Die komplette Schleife: Route → Ausführung → Verifier → Eskalation → Fall/Präferenzpaar.""" + +from __future__ import annotations + +from typing import Any + +from pydantic import BaseModel, Field + +from .cases import split_for +from .executor import ExecutionResult, Executor +from .problems import Signal +from .router import RouteDecision, Router +from .skills import SkillLibrary +from .store import Store, new_id +from .verifier import Verifier, VerifyResult + +PREVENTIVE_EVENT_SIGNALS = { + Signal.CONTEXT_EXCEEDED, Signal.COMPACT_PROMPT, Signal.LARGER_SLOT, Signal.NEEDS_SPLIT, + Signal.SKILL_CONFLICT, Signal.SLOT_UNAVAILABLE, +} + + +class Attempt(BaseModel): + decision: RouteDecision + execution: ExecutionResult | None = None + verification: VerifyResult | None = None + + +class RunResult(BaseModel): + run_id: str + status: str # ok | failed | too_large + output: str | None = None + stage: int | None = None + skill: str | None = None + attempts: list[Attempt] = Field(default_factory=list) + + @property + def escalations(self) -> int: + return max(0, len(self.attempts) - 1) + + +class Runner: + def __init__(self, router: Router, executor: Executor, verifier: Verifier, library: SkillLibrary, + store: Store, metrics: Any | None = None): + self.router = router + self.executor = executor + self.verifier = verifier + self.library = library + self.store = store + self.metrics = metrics + + def run(self, text: str, hint_skill: str | None = None) -> RunResult: + run_id = new_id() + decision = self.router.route(text, hint_skill=hint_skill) + attempts: list[Attempt] = [] + failed_weak: list[tuple[Attempt, ExecutionResult]] = [] + + while True: + self._events(decision) + attempt = Attempt(decision=decision) + attempts.append(attempt) + if decision.status != "routed": + return self._finish(run_id, decision.status, attempts, None) + + execution = self.executor.execute(decision, text) + attempt.execution = execution + skill = self.library.get(decision.skill) if decision.skill else None + + signal: Signal | None = None + if execution.signals: + signal = execution.signals[0] + for s in execution.signals: + self.store.add_event(s.value, subject=decision.slot or decision.skill) + else: + verification = self.verifier.verify(skill, text, execution.output) + attempt.verification = verification + if not verification.passed: + signal = Signal.VERIFIER_FAILED + if decision.similarity is not None and decision.stage >= 1: + self.store.add_event(Signal.MISROUTE_SUSPECTED.value, subject=decision.skill) + + infra = signal in {Signal.ENDPOINT_ERROR, Signal.CONTEXT_OVERFLOW} + if not infra: + self._record_case(run_id, text, decision, attempt, skill) + + if signal is None: + self._record_pairs(run_id, text, attempt, failed_weak, skill) + return self._finish(run_id, "ok", attempts, attempt) + if decision.stage == 1 and not infra and execution.output: + failed_weak.append((attempt, execution)) + decision = self.router.escalate(decision, signal, text) + + # --- Hilfen ----------------------------------------------------------------------------- + def _events(self, decision: RouteDecision) -> None: + for s in decision.signals: + if s in PREVENTIVE_EVENT_SIGNALS: + self.store.add_event(s.value, subject=decision.slot or decision.skill) + + def _record_case(self, run_id: str, text: str, decision: RouteDecision, attempt: Attempt, + skill: Any) -> None: + execution = attempt.execution + assert execution is not None + verification = attempt.verification + if verification is None: + passed: bool | None = False + elif verification.inconclusive: + passed = None # nicht unabhängig geprüft → nicht ins Training + else: + passed = verification.passed + self.store.add_case( + run_id=run_id, + input=text, + skill=decision.skill, + skill_version=decision.skill_version, + cluster=skill.cluster if skill else None, + stage=decision.stage, + slot=decision.slot, + model=decision.model, + adapter=decision.adapter, + messages=execution.messages or [{"role": "user", "content": text}], + output=execution.output, + passed=passed, + signals=[s.value for s in execution.signals], + prompt_tokens=decision.estimated_tokens, + split=split_for(text), + origin="route", + ) + + def _record_pairs(self, run_id: str, text: str, winner: Attempt, + failed: list[tuple[Attempt, ExecutionResult]], skill: Any) -> None: + if skill is None or not failed or winner.verification is None or winner.verification.inconclusive: + return + for _loser, execution in failed: + # Prompt im Format der schwachen Stufe: genau das soll das schwache Modell lernen + prompt = execution.messages or [{"role": "user", "content": text}] + self.store.add_preference_pair( + run_id=run_id, + skill=skill.name, + cluster=skill.cluster, + prompt=prompt, + chosen=winner.execution.output if winner.execution else "", + rejected=execution.output, + split=split_for(text), + ) + + def _finish(self, run_id: str, status: str, attempts: list[Attempt], last: Attempt | None) -> RunResult: + result = RunResult( + run_id=run_id, + status=status, + output=last.execution.output if last and last.execution else None, + stage=last.decision.stage if last else attempts[-1].decision.stage, + skill=attempts[0].decision.skill, + attempts=attempts, + ) + if self.metrics is not None: + self.metrics.observe_run(result) + return result diff --git a/core/src/knaster/settings.py b/core/src/knaster/settings.py new file mode 100644 index 0000000..4cebd55 --- /dev/null +++ b/core/src/knaster/settings.py @@ -0,0 +1,64 @@ +"""Konfiguration aus Umgebungsvariablen.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[3] + + +def _bool(value: str | None, default: bool) -> bool: + if value is None: + return default + return value.strip().lower() in {"1", "true", "yes", "ja", "on"} + + +@dataclass +class Settings: + api_token: str = "" + db_url: str = "sqlite:///data/knaster.db" + site_dir: Path = field(default_factory=lambda: REPO_ROOT / "sites" / "knaster") + skills_dir: Path = field(default_factory=lambda: REPO_ROOT / "skills") + config_dir: Path = field(default_factory=lambda: REPO_ROOT / "config") + dry_run: bool = True + litellm_base_url: str = "http://litellm:4000" + litellm_api_key: str = "" + embed_model: str = "embed-default" + rerank_url: str = "" + rerank_model: str = "rerank-default" + portainer_host: str = "" + portainer_key: str = "" + proxmox_url: str = "" + proxmox_token_id: str = "" + proxmox_token_secret: str = "" + notify_webhook_url: str = "" + job_webhook_url: str = "" + request_timeout: float = 120.0 + + @classmethod + def from_env(cls) -> Settings: + env = os.environ + base = cls() + return cls( + api_token=env.get("KNASTER_API_TOKEN", base.api_token), + db_url=env.get("KNASTER_DB_URL", base.db_url), + site_dir=Path(env.get("KNASTER_SITE_DIR", base.site_dir)), + skills_dir=Path(env.get("KNASTER_SKILLS_DIR", base.skills_dir)), + config_dir=Path(env.get("KNASTER_CONFIG_DIR", base.config_dir)), + dry_run=_bool(env.get("KNASTER_DRY_RUN"), True), + litellm_base_url=env.get("LITELLM_BASE_URL", base.litellm_base_url), + litellm_api_key=env.get("LITELLM_API_KEY", base.litellm_api_key), + embed_model=env.get("KNASTER_EMBED_MODEL", base.embed_model), + rerank_url=env.get("RERANK_URL", base.rerank_url), + rerank_model=env.get("KNASTER_RERANK_MODEL", base.rerank_model), + portainer_host=env.get("PORTAINER_HOST", ""), + portainer_key=env.get("PORTAINER_KEY", ""), + proxmox_url=env.get("PROXMOX_URL", ""), + proxmox_token_id=env.get("PROXMOX_TOKEN_ID", ""), + proxmox_token_secret=env.get("PROXMOX_TOKEN_SECRET", ""), + notify_webhook_url=env.get("NOTIFY_WEBHOOK_URL", ""), + job_webhook_url=env.get("KNASTER_JOB_WEBHOOK_URL", ""), + request_timeout=float(env.get("KNASTER_REQUEST_TIMEOUT", base.request_timeout)), + ) diff --git a/core/src/knaster/skills.py b/core/src/knaster/skills.py new file mode 100644 index 0000000..7a30dfc --- /dev/null +++ b/core/src/knaster/skills.py @@ -0,0 +1,161 @@ +"""SKILL.md-Bibliothek (agentskills-/Hermes-kompatibel, Zusatzfelder unter metadata.knaster).""" + +from __future__ import annotations + +import re +from pathlib import Path +from typing import Any, Literal + +import yaml +from pydantic import BaseModel, Field + +from .store import Store + +_FRONTMATTER = re.compile(r"^---\s*\n(.*?)\n---\s*\n?(.*)$", re.DOTALL) +_OPTIONAL_HEADING = re.compile(r"^##\s+Beispiele\b", re.IGNORECASE) + +SkillStatus = Literal["draft", "active", "demoted", "retired"] + + +class OutputSpec(BaseModel): + format: Literal["text", "json"] = "text" + schema_: dict[str, Any] | None = Field(default=None, alias="schema") + + model_config = {"populate_by_name": True} + + +class KnasterMeta(BaseModel): + version: int = 1 + status: SkillStatus = "draft" + stage: int = Field(default=0, ge=0, le=3) + execution: Literal["llm", "program", "classifier"] = "llm" + llm_fallback: bool = True + slot: str = "tier1-general" + cluster: str | None = None + adapter: Literal["auto", "none"] = "auto" + threshold: float = 0.80 + triggers: list[str] = Field(default_factory=list) + output: OutputSpec = Field(default_factory=OutputSpec) + verifier: list[dict[str, Any]] = Field(default_factory=list) + max_output_tokens: int = 1024 + program: str | None = None + classifier: str | None = None + min_confidence: float = 0.6 + + +class Skill(BaseModel): + name: str + description: str + body: str + meta: KnasterMeta + path: str | None = None + + @property + def cluster(self) -> str: + return self.meta.cluster or self.name + + @property + def routable(self) -> bool: + return self.meta.status in {"active", "demoted"} + + def prompt(self, variant: Literal["full", "compact"]) -> str: + if variant == "full": + return self.body.strip() + return compact_body(self.body) + + def index_texts(self) -> list[str]: + return [self.description, *self.meta.triggers] + + +def compact_body(body: str) -> str: + """Entfernt '## Beispiele…'-Abschnitte bis zur nächsten Überschrift gleicher/höherer Ebene.""" + out: list[str] = [] + skipping = False + for line in body.splitlines(): + if _OPTIONAL_HEADING.match(line): + skipping = True + continue + if skipping and re.match(r"^#{1,2}\s", line): + skipping = False + if not skipping: + out.append(line) + return "\n".join(out).strip() + + +def parse_skill(text: str, path: str | None = None) -> Skill: + match = _FRONTMATTER.match(text) + if not match: + raise ValueError(f"{path or 'Skill'}: Frontmatter fehlt") + front = yaml.safe_load(match.group(1)) or {} + for key in ("name", "description"): + if not front.get(key): + raise ValueError(f"{path or 'Skill'}: Feld '{key}' fehlt") + knaster = ((front.get("metadata") or {}).get("knaster")) or {} + return Skill( + name=str(front["name"]), + description=str(front["description"]).strip(), + body=match.group(2), + meta=KnasterMeta(**knaster), + path=path, + ) + + +class SkillLibrary: + """Lädt Skills aus einem Verzeichnis (//SKILL.md) und überlagert Laufzeitzustand.""" + + def __init__(self, skills: list[Skill], store: Store | None = None): + self._skills = {s.name: s for s in skills} + if len(self._skills) != len(skills): + raise ValueError("Doppelte Skill-Namen") + self.store = store + + @classmethod + def from_dir(cls, directory: str | Path, store: Store | None = None) -> SkillLibrary: + skills = [] + for file in sorted(Path(directory).glob("*/SKILL.md")): + skills.append(parse_skill(file.read_text(encoding="utf-8"), str(file))) + return cls(skills, store) + + def _states(self) -> dict[str, dict[str, Any]]: + if not self.store: + return {} + return {row["name"]: row for row in self.store.all_skill_states()} + + @staticmethod + def _apply_state(skill: Skill, states: dict[str, dict[str, Any]]) -> Skill: + row = states.get(skill.name) + if not row: + return skill + meta = skill.meta.model_copy() + if row.get("status"): + meta.status = row["status"] + if row.get("stage") is not None: + meta.stage = row["stage"] + return skill.model_copy(update={"meta": meta}) + + def get(self, name: str) -> Skill: + return self._apply_state(self._skills[name], self._states()) + + def has(self, name: str) -> bool: + return name in self._skills + + def all(self) -> list[Skill]: + states = self._states() + return [self._apply_state(s, states) for s in self._skills.values()] + + def routable(self) -> list[Skill]: + return [s for s in self.all() if s.routable] + + def set_state(self, name: str, *, status: str | None = None, stage: int | None = None, + reason: str = "") -> None: + if name not in self._skills: + raise KeyError(name) + if not self.store: + raise RuntimeError("SkillLibrary ohne Store kann keinen Zustand speichern") + current = self.get(name) + self.store.set_skill_state( + name, + status=status or current.meta.status, + stage=current.meta.stage if stage is None else stage, + reason=reason, + ) diff --git a/core/src/knaster/store.py b/core/src/knaster/store.py new file mode 100644 index 0000000..101189e --- /dev/null +++ b/core/src/knaster/store.py @@ -0,0 +1,348 @@ +"""Persistenz (SQLAlchemy Core). SQLite für Tests/kleine Setups, Postgres im Betrieb.""" + +from __future__ import annotations + +import uuid +from typing import Any + +from sqlalchemy import ( + JSON, + Boolean, + Column, + Float, + Integer, + MetaData, + String, + Table, + Text, + create_engine, + delete, + func, + insert, + select, + update, +) +from sqlalchemy.engine import Engine +from sqlalchemy.pool import StaticPool + +from .util import now + +metadata = MetaData() + +cases = Table( + "cases", + metadata, + Column("id", String(36), primary_key=True), + Column("ts", Float, nullable=False, index=True), + Column("run_id", String(36), index=True), + Column("input", Text, nullable=False), + Column("skill", String(200), index=True), + Column("skill_version", Integer), + Column("cluster", String(200), index=True), + Column("stage", Integer, nullable=False), + Column("slot", String(100)), + Column("model", String(200)), + Column("adapter", String(200)), + Column("messages", JSON), + Column("output", Text), + Column("passed", Boolean), + Column("signals", JSON), + Column("prompt_tokens", Integer), + Column("split", String(10), nullable=False), + Column("origin", String(20), nullable=False), +) + +preference_pairs = Table( + "preference_pairs", + metadata, + Column("id", String(36), primary_key=True), + Column("ts", Float, nullable=False), + Column("run_id", String(36)), + Column("skill", String(200), index=True), + Column("cluster", String(200)), + Column("prompt", JSON), + Column("chosen", Text), + Column("rejected", Text), + Column("split", String(10)), +) + +proposals = Table( + "proposals", + metadata, + Column("seq", Integer, primary_key=True, autoincrement=True), + Column("kind", String(50), nullable=False), + Column("params", JSON), + Column("summary", Text), + Column("reason", Text), + Column("risk", String(10)), + Column("state", String(20), nullable=False, index=True), + Column("source", String(20)), + Column("dedup_key", String(300), index=True), + Column("created_at", Float), + Column("decided_at", Float), + Column("decided_by", String(100)), + Column("result", JSON), +) + +slot_state = Table( + "slot_state", + metadata, + Column("slot", String(100), primary_key=True), + Column("status", String(20)), + Column("effective_context", Integer), + Column("base_model", String(200)), + Column("reason", Text), + Column("set_by", String(50)), + Column("updated_at", Float), +) + +adapter_state = Table( + "adapter_state", + metadata, + Column("name", String(200), primary_key=True), + Column("base_model", String(200)), + Column("cluster", String(200)), + Column("version", Integer), + Column("status", String(20)), + Column("updated_at", Float), +) + +skill_state = Table( + "skill_state", + metadata, + Column("name", String(200), primary_key=True), + Column("status", String(20)), + Column("stage", Integer), + Column("reason", Text), + Column("updated_at", Float), +) + +candidates = Table( + "candidates", + metadata, + Column("id", String(36), primary_key=True), + Column("kind", String(30)), + Column("slot", String(100)), + Column("target", JSON), + Column("baseline", JSON), + Column("state", String(30), index=True), + Column("metrics", JSON), + Column("history", JSON), + Column("snapshot", JSON), + Column("created_at", Float), + Column("updated_at", Float), +) + +healer_state = Table( + "healer_state", + metadata, + Column("key", String(300), primary_key=True), + Column("data", JSON), + Column("updated_at", Float), +) + +events = Table( + "events", + metadata, + Column("id", Integer, primary_key=True, autoincrement=True), + Column("ts", Float, nullable=False, index=True), + Column("kind", String(50), nullable=False, index=True), + Column("subject", String(200), index=True), + Column("data", JSON), +) + + +def new_id() -> str: + return str(uuid.uuid4()) + + +class Store: + """Dünne Repository-Schicht. Jede Methode öffnet ihre eigene Transaktion.""" + + def __init__(self, url: str): + kwargs: dict[str, Any] = {} + if url.startswith("sqlite"): + kwargs["connect_args"] = {"check_same_thread": False} + if url in {"sqlite://", "sqlite:///:memory:"}: + kwargs["poolclass"] = StaticPool + self.engine: Engine = create_engine(url, future=True, **kwargs) + metadata.create_all(self.engine) + + # --- Fälle ----------------------------------------------------------------------------- + def add_case(self, **row: Any) -> str: + row.setdefault("id", new_id()) + row.setdefault("ts", now()) + with self.engine.begin() as conn: + conn.execute(insert(cases).values(**row)) + return row["id"] + + def list_cases( + self, + *, + skill: str | None = None, + cluster: str | None = None, + min_stage: int | None = None, + stage: int | None = None, + split: str | None = None, + passed: bool | None = None, + since: float | None = None, + limit: int | None = None, + ) -> list[dict[str, Any]]: + q = select(cases).order_by(cases.c.ts.desc()) + if skill is not None: + q = q.where(cases.c.skill == skill) + if cluster is not None: + q = q.where(cases.c.cluster == cluster) + if min_stage is not None: + q = q.where(cases.c.stage >= min_stage) + if stage is not None: + q = q.where(cases.c.stage == stage) + if split is not None: + q = q.where(cases.c.split == split) + if passed is not None: + q = q.where(cases.c.passed == passed) + if since is not None: + q = q.where(cases.c.ts >= since) + if limit is not None: + q = q.limit(limit) + with self.engine.connect() as conn: + return [dict(r._mapping) for r in conn.execute(q)] + + def add_preference_pair(self, **row: Any) -> str: + row.setdefault("id", new_id()) + row.setdefault("ts", now()) + with self.engine.begin() as conn: + conn.execute(insert(preference_pairs).values(**row)) + return row["id"] + + def list_preference_pairs(self, *, cluster: str | None = None) -> list[dict[str, Any]]: + q = select(preference_pairs).order_by(preference_pairs.c.ts) + if cluster is not None: + q = q.where(preference_pairs.c.cluster == cluster) + with self.engine.connect() as conn: + return [dict(r._mapping) for r in conn.execute(q)] + + # --- Ereignisse ------------------------------------------------------------------------ + def add_event(self, kind: str, subject: str | None = None, data: dict | None = None) -> None: + with self.engine.begin() as conn: + conn.execute(insert(events).values(ts=now(), kind=kind, subject=subject, data=data or {})) + + def count_events(self, kind: str, *, subject: str | None = None, since: float | None = None) -> int: + q = select(func.count()).select_from(events).where(events.c.kind == kind) + if subject is not None: + q = q.where(events.c.subject == subject) + if since is not None: + q = q.where(events.c.ts >= since) + with self.engine.connect() as conn: + return int(conn.execute(q).scalar_one()) + + def event_subjects(self, kind: str, *, since: float | None = None) -> dict[str, int]: + q = select(events.c.subject, func.count()).where(events.c.kind == kind).group_by(events.c.subject) + if since is not None: + q = q.where(events.c.ts >= since) + with self.engine.connect() as conn: + return {r[0]: int(r[1]) for r in conn.execute(q) if r[0] is not None} + + # --- generische Zustandstabellen ----------------------------------------------------- + def _upsert(self, table: Table, key_col: str, key: str, values: dict[str, Any]) -> None: + values = {**values, "updated_at": now()} + with self.engine.begin() as conn: + exists = conn.execute(select(table).where(table.c[key_col] == key)).first() + if exists: + conn.execute(update(table).where(table.c[key_col] == key).values(**values)) + else: + conn.execute(insert(table).values(**{key_col: key, **values})) + + def _get(self, table: Table, key_col: str, key: str) -> dict[str, Any] | None: + with self.engine.connect() as conn: + row = conn.execute(select(table).where(table.c[key_col] == key)).first() + return dict(row._mapping) if row else None + + def _all(self, table: Table) -> list[dict[str, Any]]: + with self.engine.connect() as conn: + return [dict(r._mapping) for r in conn.execute(select(table))] + + def set_slot_state(self, slot: str, **values: Any) -> None: + self._upsert(slot_state, "slot", slot, values) + + def get_slot_state(self, slot: str) -> dict[str, Any] | None: + return self._get(slot_state, "slot", slot) + + def all_slot_states(self) -> list[dict[str, Any]]: + return self._all(slot_state) + + def set_adapter_state(self, name: str, **values: Any) -> None: + self._upsert(adapter_state, "name", name, values) + + def all_adapter_states(self) -> list[dict[str, Any]]: + return self._all(adapter_state) + + def set_skill_state(self, name: str, **values: Any) -> None: + self._upsert(skill_state, "name", name, values) + + def all_skill_states(self) -> list[dict[str, Any]]: + return self._all(skill_state) + + # --- Vorschläge ------------------------------------------------------------------------ + def add_proposal(self, **row: Any) -> int: + row.setdefault("created_at", now()) + row.setdefault("state", "open") + with self.engine.begin() as conn: + res = conn.execute(insert(proposals).values(**row)) + return int(res.inserted_primary_key[0]) + + def get_proposal(self, seq: int) -> dict[str, Any] | None: + return self._get(proposals, "seq", seq) # type: ignore[arg-type] + + def update_proposal(self, seq: int, **values: Any) -> None: + with self.engine.begin() as conn: + conn.execute(update(proposals).where(proposals.c.seq == seq).values(**values)) + + def list_proposals(self, *, state: str | None = None) -> list[dict[str, Any]]: + q = select(proposals).order_by(proposals.c.seq) + if state is not None: + q = q.where(proposals.c.state == state) + with self.engine.connect() as conn: + return [dict(r._mapping) for r in conn.execute(q)] + + def open_proposal_with_key(self, dedup_key: str) -> dict[str, Any] | None: + q = select(proposals).where(proposals.c.dedup_key == dedup_key, proposals.c.state == "open") + with self.engine.connect() as conn: + row = conn.execute(q).first() + return dict(row._mapping) if row else None + + # --- Kandidaten ------------------------------------------------------------------------ + def add_candidate(self, **row: Any) -> str: + row.setdefault("id", new_id()) + row.setdefault("created_at", now()) + row.setdefault("updated_at", row["created_at"]) + with self.engine.begin() as conn: + conn.execute(insert(candidates).values(**row)) + return row["id"] + + def get_candidate(self, cid: str) -> dict[str, Any] | None: + return self._get(candidates, "id", cid) + + def update_candidate(self, cid: str, **values: Any) -> None: + values["updated_at"] = now() + with self.engine.begin() as conn: + conn.execute(update(candidates).where(candidates.c.id == cid).values(**values)) + + def list_candidates(self, *, state: str | None = None) -> list[dict[str, Any]]: + q = select(candidates).order_by(candidates.c.created_at) + if state is not None: + q = q.where(candidates.c.state == state) + with self.engine.connect() as conn: + return [dict(r._mapping) for r in conn.execute(q)] + + # --- Healer-Zustand -------------------------------------------------------------------- + def get_healer_state(self, key: str) -> dict[str, Any] | None: + row = self._get(healer_state, "key", key) + return row["data"] if row else None + + def set_healer_state(self, key: str, data: dict[str, Any]) -> None: + self._upsert(healer_state, "key", key, {"data": data}) + + def delete_healer_state(self, key: str) -> None: + with self.engine.begin() as conn: + conn.execute(delete(healer_state).where(healer_state.c.key == key)) diff --git a/core/src/knaster/util.py b/core/src/knaster/util.py new file mode 100644 index 0000000..4b5f747 --- /dev/null +++ b/core/src/knaster/util.py @@ -0,0 +1,56 @@ +"""Kleine Hilfsfunktionen ohne Fachlogik.""" + +from __future__ import annotations + +import re +import time +from pathlib import Path +from typing import Any + +import yaml + +_DURATION = re.compile(r"^\s*(\d+(?:\.\d+)?)\s*([smhd]?)\s*$") +_FACTORS = {"": 1, "s": 1, "m": 60, "h": 3600, "d": 86400} +_PLACEHOLDER = re.compile(r"\{([a-zA-Z0-9_]+)\}") + + +def now() -> float: + return time.time() + + +def load_yaml(path: str | Path) -> Any: + with open(path, encoding="utf-8") as fh: + return yaml.safe_load(fh) or {} + + +def parse_duration(value: str | int | float | None) -> float: + """'5m' -> 300.0; Zahlen gelten als Sekunden.""" + if value is None: + return 0.0 + if isinstance(value, int | float): + return float(value) + match = _DURATION.match(value) + if not match: + raise ValueError(f"Ungültige Dauer: {value!r}") + return float(match.group(1)) * _FACTORS[match.group(2)] + + +class MissingLabel(KeyError): + pass + + +def fill_template(value: Any, labels: dict[str, str]) -> Any: + """Ersetzt {label} rekursiv; fehlende Labels lösen MissingLabel aus.""" + if isinstance(value, str): + def repl(m: re.Match[str]) -> str: + key = m.group(1) + if key not in labels: + raise MissingLabel(key) + return str(labels[key]) + + return _PLACEHOLDER.sub(repl, value) + if isinstance(value, dict): + return {k: fill_template(v, labels) for k, v in value.items()} + if isinstance(value, list): + return [fill_template(v, labels) for v in value] + return value diff --git a/core/src/knaster/verifier.py b/core/src/knaster/verifier.py new file mode 100644 index 0000000..15bb585 --- /dev/null +++ b/core/src/knaster/verifier.py @@ -0,0 +1,116 @@ +"""Verifier: deterministische Prüfungen + optionaler LLM-Judge.""" + +from __future__ import annotations + +import re +from typing import Any + +import jsonschema +from pydantic import BaseModel, Field + +from .llm import ChatClient, LLMError +from .problems import extract_json +from .registry import Registry +from .skills import Skill + +JUDGE_SYSTEM = ( + "Du bist ein strenger Prüfer. Bewerte, ob die Antwort die Aufgabe gemäß den Kriterien erfüllt. " + "Antworte mit genau einer Zeile: 'PASS' oder 'FAIL: '." +) + + +class CheckResult(BaseModel): + type: str + passed: bool | None # None = nicht entscheidbar (z. B. Judge nicht erreichbar) + detail: str = "" + + +class VerifyResult(BaseModel): + passed: bool + inconclusive: bool = False + checks: list[CheckResult] = Field(default_factory=list) + + @property + def failed_checks(self) -> list[CheckResult]: + return [c for c in self.checks if c.passed is False] + + +class Verifier: + def __init__(self, registry: Registry | None = None, chat: ChatClient | None = None, + default_judge_slot: str = "verifier-small"): + self.registry = registry + self.chat = chat + self.default_judge_slot = default_judge_slot + + def verify(self, skill: Skill | None, text: str, output: str) -> VerifyResult: + if skill is None: + return VerifyResult(passed=True, checks=[]) + checks: list[CheckResult] = [] + specs = list(skill.meta.verifier) + schema = skill.meta.output.schema_ + if skill.meta.output.format == "json": + checks.append(self._json(output, schema)) + specs = [s for s in specs if s.get("type") != "json_schema"] + for spec in specs: + checks.append(self._run(spec, skill, text, output)) + conclusive = [c for c in checks if c.passed is not None] + return VerifyResult( + passed=all(c.passed for c in conclusive), + inconclusive=len(conclusive) != len(checks), + checks=checks, + ) + + def _run(self, spec: dict[str, Any], skill: Skill, text: str, output: str) -> CheckResult: + kind = spec.get("type") + if kind == "json_schema": + return self._json(output, skill.meta.output.schema_) + if kind == "regex": + ok = re.search(spec["pattern"], output, re.MULTILINE) is not None + return CheckResult(type=kind, passed=ok, detail="" if ok else f"Muster fehlt: {spec['pattern']}") + if kind == "contains": + ok = spec["value"] in output + return CheckResult(type=kind, passed=ok, detail="" if ok else f"fehlt: {spec['value']}") + if kind == "not_contains": + ok = spec["value"] not in output + return CheckResult(type=kind, passed=ok, detail="" if ok else f"enthält: {spec['value']}") + if kind == "max_chars": + ok = len(output) <= int(spec["value"]) + return CheckResult(type=kind, passed=ok, detail=f"{len(output)} Zeichen") + if kind == "judge": + return self._judge(spec, text, output) + return CheckResult(type=str(kind), passed=False, detail="Unbekannter Prüftyp") + + @staticmethod + def _json(output: str, schema: dict[str, Any] | None) -> CheckResult: + try: + data = extract_json(output) + except ValueError as exc: + return CheckResult(type="json", passed=False, detail=f"kein gültiges JSON: {exc}") + if schema: + try: + jsonschema.validate(data, schema) + except jsonschema.ValidationError as exc: + return CheckResult(type="json_schema", passed=False, detail=exc.message) + return CheckResult(type="json_schema" if schema else "json", passed=True) + + def _judge(self, spec: dict[str, Any], text: str, output: str) -> CheckResult: + if self.chat is None or self.registry is None: + return CheckResult(type="judge", passed=None, detail="kein Judge konfiguriert") + slot_name = spec.get("slot", self.default_judge_slot) + slot = self.registry.slot(slot_name) + if not slot.available: + return CheckResult(type="judge", passed=None, detail=f"Judge-Slot {slot_name} nicht verfügbar") + user = f"Aufgabe:\n{text}\n\nKriterien:\n{spec.get('rubric', 'Korrekt und vollständig.')}\n\nAntwort:\n{output}" + try: + res = self.chat.chat(slot.spec.litellm_model, + [{"role": "system", "content": JUDGE_SYSTEM}, {"role": "user", "content": user}], + max_tokens=60, temperature=0) + except LLMError as exc: + return CheckResult(type="judge", passed=None, detail=f"Judge-Fehler: {exc}") + verdict = res.content.strip() + first = verdict.split(maxsplit=1)[0].upper().rstrip(":") if verdict else "" + if first == "PASS": + return CheckResult(type="judge", passed=True) + if first == "FAIL": + return CheckResult(type="judge", passed=False, detail=verdict[:200]) + return CheckResult(type="judge", passed=None, detail=f"Unklare Judge-Antwort: {verdict[:100]}") diff --git a/core/tests/__init__.py b/core/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/core/tests/conftest.py b/core/tests/conftest.py new file mode 100644 index 0000000..01ef6e1 --- /dev/null +++ b/core/tests/conftest.py @@ -0,0 +1,178 @@ +from __future__ import annotations + +import textwrap +from collections.abc import Callable +from pathlib import Path + +import pytest + +from knaster.embeddings import HashingEmbedder +from knaster.llm import ChatResult +from knaster.registry import Registry +from knaster.skills import SkillLibrary +from knaster.store import Store + +REGISTRY = { + "slots": { + "tier0-strong": { + "tier": 0, "device": "external", "engine": "openrouter", "litellm_model": "tier0-strong", + "base_model": "strong-x", "max_context": 128000, "effective_context": 100000, + }, + "tier1-general": { + "tier": 1, "engine": "vllm", "node": "gpu-main", "litellm_model": "tier1-general", + "base_model": "base-14b", "max_context": 8192, "effective_context": 2000, + "supports_lora": True, "fallbacks": ["tier1-flex"], + }, + "tier1-flex": { + "tier": 1, "engine": "vllm", "node": "gpu-flex", "litellm_model": "tier1-flex", + "base_model": "base-14b", "max_context": 32768, "effective_context": 16000, + "supports_lora": True, + }, + "verifier-small": { + "tier": 1, "engine": "llamacpp", "node": "gpu-small", "litellm_model": "verifier-small", + "base_model": "small-4b", "max_context": 8192, + }, + "embed-default": { + "tier": 1, "engine": "llamacpp", "node": "gpu-embed", "litellm_model": "embed-default", + "base_model": "embed-0.6b", "max_context": 8192, "kind": "embedding", + }, + }, + "adapters": [ + {"name": "rechnungen-v1", "base_model": "base-14b", "cluster": "rechnungen", "version": 1}, + ], +} + +SKILLS = { + "rechnung-extrahieren": """\ + --- + name: rechnung-extrahieren + description: Rechnung auslesen und Rechnungsnummer Betrag Datum als JSON extrahieren + metadata: + knaster: + status: active + stage: 1 + cluster: rechnungen + threshold: 0.3 + triggers: ["Rechnungsdaten aus einer Rechnung extrahieren"] + max_output_tokens: 300 + output: + format: json + schema: + type: object + required: [nummer, betrag] + properties: + nummer: {type: string} + betrag: {type: number} + --- + # Rechnung extrahieren + Lies Nummer und Betrag aus. + + ## Beispiele + Eingabe: Rechnung 7 über 10 EUR -> {"nummer": "7", "betrag": 10} + """, + "ticket-kategorisieren": """\ + --- + name: ticket-kategorisieren + description: Support Ticket einer Kategorie zuordnen Drucker Netzwerk Passwort + metadata: + knaster: + status: active + stage: 3 + execution: classifier + classifier: ticket-kat + threshold: 0.3 + min_confidence: 0.6 + --- + Ordne das Ticket einer Kategorie zu: drucker, netzwerk, passwort. + """, + "text-gross": """\ + --- + name: text-gross + description: Text in Großbuchstaben umwandeln uppercase + metadata: + knaster: + status: active + stage: 2 + execution: program + program: echo-upper + threshold: 0.3 + --- + Wandle den Text in Großbuchstaben um. + """, + "entwurf": """\ + --- + name: entwurf + description: Gedicht über Rechnung schreiben + metadata: + knaster: + status: draft + --- + Entwurf. + """, + "vertrag-pruefen": """\ + --- + name: vertrag-pruefen + description: Vertrag prüfen Kündigungsfrist Laufzeit Klauseln + metadata: + knaster: + status: demoted + stage: 1 + threshold: 0.3 + verifier: + - {type: contains, value: "Kündigungsfrist"} + --- + Prüfe den Vertrag. + """, +} + + +@pytest.fixture +def store() -> Store: + return Store("sqlite://") + + +@pytest.fixture +def registry(store: Store) -> Registry: + return Registry.from_dict(REGISTRY, store) + + +@pytest.fixture +def skills_dir(tmp_path: Path) -> Path: + for name, text in SKILLS.items(): + d = tmp_path / "skills" / name + d.mkdir(parents=True) + (d / "SKILL.md").write_text(textwrap.dedent(text), encoding="utf-8") + return tmp_path / "skills" + + +@pytest.fixture +def library(skills_dir: Path, store: Store) -> SkillLibrary: + return SkillLibrary.from_dir(skills_dir, store) + + +@pytest.fixture +def embedder() -> HashingEmbedder: + return HashingEmbedder() + + +class FakeChat: + """Antworten pro Modellname; ein Handler darf ChatResult liefern oder eine Exception werfen.""" + + def __init__(self, handlers: dict[str, Callable[[list[dict[str, str]]], ChatResult | str]] | None = None): + self.handlers = handlers or {} + self.calls: list[tuple[str, list[dict[str, str]]]] = [] + + def chat(self, model, messages, *, max_tokens=None, temperature=None): + self.calls.append((model, messages)) + handler = self.handlers.get(model) + if handler is None: + return ChatResult(content="ok", finish_reason="stop") + result = handler(messages) + if isinstance(result, str): + return ChatResult(content=result, finish_reason="stop") + return result + + +@pytest.fixture +def fake_chat() -> FakeChat: + return FakeChat() diff --git a/core/tests/test_execution.py b/core/tests/test_execution.py new file mode 100644 index 0000000..516c97d --- /dev/null +++ b/core/tests/test_execution.py @@ -0,0 +1,222 @@ +import json + +import pytest + +from knaster.calibration import effective_context +from knaster.cases import export_preferences, export_sft, split_for +from knaster.classifier import EmbeddingClassifier +from knaster.executor import Executor +from knaster.llm import ChatResult, ContextOverflowError, EndpointError +from knaster.problems import Signal, extract_json, is_degenerate, repetition_ratio +from knaster.router import Router +from knaster.runner import Runner +from knaster.verifier import Verifier + +from .test_router import RECHNUNG + +GOOD = json.dumps({"nummer": "4711", "betrag": 99}) + + +@pytest.fixture +def classifiers(embedder): + examples = [] + for label, texts in { + "drucker": ["Drucker druckt nicht", "Drucker Papierstau", "Drucker offline"], + "netzwerk": ["WLAN geht nicht", "Netzwerk langsam", "kein Internet"], + }.items(): + examples += [(v, label) for v in embedder.embed(texts)] + return {"ticket-kat": EmbeddingClassifier.train(examples)} + + +@pytest.fixture +def make_runner(registry, library, embedder, store, fake_chat, classifiers): + def make(**handlers): + fake_chat.handlers.update(handlers) + router = Router(registry, library, embedder) + executor = Executor(library, fake_chat, embedder, classifiers) + verifier = Verifier(registry, fake_chat) + return Runner(router, executor, verifier, library, store) + + return make + + +def test_weak_success_records_case(make_runner, store): + runner = make_runner(**{"rechnungen-v1": lambda m: GOOD}) + res = runner.run(RECHNUNG) + assert res.status == "ok" and res.stage == 1 and res.escalations == 0 + cases = store.list_cases() + assert len(cases) == 1 and cases[0]["passed"] is True and cases[0]["cluster"] == "rechnungen" + assert cases[0]["messages"][0]["role"] == "system" + + +def test_schema_failure_escalates_and_creates_pair(make_runner, store): + runner = make_runner(**{ + "rechnungen-v1": lambda m: json.dumps({"nummer": "4711"}), # betrag fehlt + "tier0-strong": lambda m: GOOD, + }) + res = runner.run(RECHNUNG) + assert res.status == "ok" and res.stage == 0 and res.escalations == 1 + pairs = store.list_preference_pairs() + assert len(pairs) == 1 + assert pairs[0]["chosen"] == GOOD and "betrag" not in pairs[0]["rejected"] + assert {c["stage"]: c["passed"] for c in store.list_cases()} == {1: False, 0: True} + + +def test_invalid_json_signal(make_runner, store): + runner = make_runner(**{"rechnungen-v1": lambda m: "kein json", "tier0-strong": lambda m: GOOD}) + res = runner.run(RECHNUNG) + assert Signal.INVALID_FORMAT in res.attempts[0].execution.signals + assert res.stage == 0 + assert store.count_events("invalid_format") == 1 + + +def test_endpoint_error_uses_fallback_slot_without_case(make_runner, store): + def boom(m): + raise EndpointError("down") + + runner = make_runner(**{"rechnungen-v1": boom}) + # Fallback-Slot tier1-flex hat dieselbe Basis → gleicher Adapter-Name; daher Fehler nur einmal + calls = {"n": 0} + + def flaky(m): + calls["n"] += 1 + if calls["n"] == 1: + raise EndpointError("down") + return GOOD + + runner.executor.chat.handlers["rechnungen-v1"] = flaky + res = runner.run(RECHNUNG) + assert res.status == "ok" + assert [a.decision.slot for a in res.attempts] == ["tier1-general", "tier1-flex"] + assert len(store.list_cases()) == 1 # Infrastrukturfehler erzeugt keinen Fall + assert store.list_preference_pairs() == [] + + +def test_context_overflow_detected(make_runner): + def overflow(m): + raise ContextOverflowError("maximum context length") + + runner = make_runner(**{"rechnungen-v1": overflow, "tier0-strong": lambda m: GOOD}) + res = runner.run(RECHNUNG) + assert res.attempts[0].execution.signals == [Signal.CONTEXT_OVERFLOW] + assert res.status == "ok" + + +def test_truncated_and_degenerate(make_runner): + loop = " ".join(["immer wieder das gleiche"] * 30) + runner = make_runner(**{ + "rechnungen-v1": lambda m: ChatResult(content=loop, finish_reason="length"), + "tier0-strong": lambda m: GOOD, + }) + res = runner.run(RECHNUNG) + sig = res.attempts[0].execution.signals + assert Signal.OUTPUT_TRUNCATED in sig and Signal.DEGENERATE_OUTPUT in sig + + +def test_total_failure(make_runner): + runner = make_runner(**{"rechnungen-v1": lambda m: "x", "tier0-strong": lambda m: "y"}) + res = runner.run(RECHNUNG) + assert res.status == "failed" + + +def test_classifier_stage(make_runner): + runner = make_runner() + res = runner.run("Ticket: Drucker druckt nicht Papierstau") + assert res.status == "ok" and res.stage == 3 and res.output == "drucker" + + +def test_classifier_low_confidence_escalates(make_runner): + runner = make_runner(**{"tier1-general": lambda m: "drucker"}) + res = runner.run("Ticket einer Kategorie zuordnen: Monitor flackert") + assert res.attempts[0].execution.signals == [Signal.CLASSIFIER_LOW_CONFIDENCE] + assert res.attempts[1].decision.stage == 1 + + +def test_program_stage(make_runner): + res = make_runner().run("Text in Großbuchstaben: hallo welt") + assert res.output == "TEXT IN GROSSBUCHSTABEN: HALLO WELT" + + +def test_judge_check(registry, library, fake_chat): + skill = library.get("vertrag-pruefen").model_copy(deep=True) + skill.meta.verifier = [{"type": "judge", "rubric": "muss Frist nennen"}] + v = Verifier(registry, fake_chat) + fake_chat.handlers["verifier-small"] = lambda m: "FAIL: Frist fehlt" + r = v.verify(skill, "Vertrag", "Antwort") + assert not r.passed and r.failed_checks[0].detail.startswith("FAIL") + fake_chat.handlers["verifier-small"] = lambda m: "PASS" + assert v.verify(skill, "Vertrag", "Antwort").passed + + +def test_judge_unavailable_is_inconclusive(registry, library, fake_chat): + skill = library.get("vertrag-pruefen").model_copy(deep=True) + skill.meta.verifier = [{"type": "judge"}, {"type": "contains", "value": "Frist"}] + registry.set_status("verifier-small", "unavailable") + r = Verifier(registry, fake_chat).verify(skill, "Vertrag", "Frist 3 Monate") + assert r.passed and r.inconclusive + + +def test_inconclusive_case_not_used_for_training(make_runner, registry, library, store): + library.set_state("vertrag-pruefen", status="active") + registry.set_status("verifier-small", "unavailable") + skill_file = library._skills["vertrag-pruefen"] + skill_file.meta.verifier = [{"type": "judge"}] + runner = make_runner(**{"tier1-general": lambda m: "Kündigungsfrist: 3 Monate"}) + res = runner.run("Vertrag prüfen auf Kündigungsfrist") + assert res.status == "ok" + assert store.list_cases()[0]["passed"] is None + + +def test_verifier_rules(): + from knaster.skills import KnasterMeta, Skill + + skill = Skill(name="s", description="d", body="b", meta=KnasterMeta(verifier=[ + {"type": "regex", "pattern": r"^\d+$"}, {"type": "not_contains", "value": "Fehler"}, + {"type": "max_chars", "value": 5}, {"type": "unbekannt"}])) + r = Verifier().verify(skill, "t", "123") + assert [c.passed for c in r.checks] == [True, True, True, False] + + +def test_problem_helpers(): + assert extract_json('```json\n{"a": 1}\n```') == {"a": 1} + with pytest.raises(ValueError): + extract_json("{a:") + assert repetition_ratio("a b c d e f") == 0 + assert is_degenerate(" ".join(["la le lu"] * 20)) + assert not is_degenerate("kurz kurz kurz kurz") + + +def test_split_is_deterministic(): + assert split_for("abc") == split_for("abc") + splits = {split_for(f"text {i}") for i in range(100)} + assert splits == {"train", "eval"} + + +def test_exports(make_runner, store): + runner = make_runner(**{"rechnungen-v1": lambda m: '{"nummer": "1"}', "tier0-strong": lambda m: GOOD}) + for i in range(30): + runner.run(f"{RECHNUNG} Nr {i}") + train = list(export_sft(store, cluster="rechnungen", split="train")) + evals = list(export_sft(store, cluster="rechnungen", split="eval")) + assert len(train) + len(evals) == 30 + row = json.loads(train[0]) + assert row["messages"][-1] == {"role": "assistant", "content": GOOD} + assert row["meta"]["stage"] == 0 + prefs = list(export_preferences(store, cluster="rechnungen")) + assert prefs and json.loads(prefs[0])["chosen"] == GOOD + + +def test_effective_context_calibration(): + obs = [(1000, True)] * 30 + [(3000, True)] * 28 + [(3000, False)] * 2 + [(6000, True)] * 15 + \ + [(6000, False)] * 10 + [(12000, True)] * 5 + assert effective_context(obs, target=0.9, min_n=20) == 4096 + assert effective_context([(1000, True)] * 5) is None + # Buckets ohne genug Daten unterbrechen nicht + assert effective_context([(1000, True)] * 25 + [(20000, True)] * 25, min_n=20) == 32768 + + +def test_classifier_persistence(tmp_path, classifiers, embedder): + clf = classifiers["ticket-kat"] + clf.save(tmp_path / "c.json") + loaded = EmbeddingClassifier.load(tmp_path / "c.json") + assert loaded.predict(embedder.embed(["Drucker"])[0])[0] == "drucker" diff --git a/core/tests/test_registry_skills.py b/core/tests/test_registry_skills.py new file mode 100644 index 0000000..b42a5bd --- /dev/null +++ b/core/tests/test_registry_skills.py @@ -0,0 +1,78 @@ +from knaster.registry import AdapterSpec, Registry +from knaster.skills import compact_body, parse_skill + + +def test_slot_effective_context_prefers_calibration(registry): + slot = registry.slot("tier1-general") + assert slot.effective_context == 2000 + registry.set_calibrated_context("tier1-general", 1500) + assert registry.slot("tier1-general").effective_context == 1500 + + +def test_effective_context_never_exceeds_max(registry): + registry.set_calibrated_context("tier1-general", 999999) + assert registry.slot("tier1-general").effective_context == 8192 + + +def test_status_and_strong_slot(registry): + registry.set_status("tier1-general", "degraded", reason="test", set_by="healer") + slot = registry.slot("tier1-general") + assert not slot.available and slot.set_by == "healer" + assert registry.strong_slot().name == "tier0-strong" + + +def test_unknown_fallback_rejected(store): + import pytest + + with pytest.raises(ValueError): + Registry.from_dict({"slots": {"a": {"tier": 1, "litellm_model": "a", "base_model": "b", + "max_context": 10, "fallbacks": ["x"]}}}, store) + + +def test_adapter_overlay_and_active(registry): + assert registry.active_adapter("base-14b", "rechnungen").name == "rechnungen-v1" + registry.set_adapter(AdapterSpec(name="rechnungen-v2", base_model="base-14b", cluster="rechnungen", + version=2, status="active")) + assert registry.active_adapter("base-14b", "rechnungen").name == "rechnungen-v2" + registry.set_adapter(AdapterSpec(name="rechnungen-v2", base_model="base-14b", cluster="rechnungen", + version=2, status="retired")) + assert registry.active_adapter("base-14b", "rechnungen").name == "rechnungen-v1" + assert registry.active_adapter("other-base", "rechnungen") is None + + +def test_base_model_override_and_export(registry): + registry.set_base_model("tier1-general", "new-base") + assert registry.slot("tier1-general").base_model == "new-base" + exported = registry.export() + assert exported["slots"]["tier1-general"]["base_model"] == "new-base" + + +def test_parse_skill_and_compact(library): + skill = library.get("rechnung-extrahieren") + assert skill.cluster == "rechnungen" + assert skill.meta.output.format == "json" + assert "Beispiele" in skill.prompt("full") + assert "Beispiele" not in skill.prompt("compact") + assert "Lies Nummer" in skill.prompt("compact") + + +def test_compact_body_keeps_following_sections(): + body = "# T\nA\n## Beispiele\nx\ny\n## Regeln\nR\n" + assert compact_body(body) == "# T\nA\n## Regeln\nR" + + +def test_parse_skill_requires_fields(): + import pytest + + with pytest.raises(ValueError): + parse_skill("---\nname: x\n---\nbody") + with pytest.raises(ValueError): + parse_skill("kein frontmatter") + + +def test_routable_and_state_overlay(library): + names = {s.name for s in library.routable()} + assert "entwurf" not in names and "vertrag-pruefen" in names + library.set_state("rechnung-extrahieren", status="demoted", reason="drift") + assert library.get("rechnung-extrahieren").meta.status == "demoted" + assert library.get("rechnung-extrahieren").meta.stage == 1 diff --git a/core/tests/test_router.py b/core/tests/test_router.py new file mode 100644 index 0000000..a546998 --- /dev/null +++ b/core/tests/test_router.py @@ -0,0 +1,133 @@ +import pytest + +from knaster.problems import Signal +from knaster.router import Router, RouterConfig + +RECHNUNG = "Rechnung auslesen: Rechnungsnummer und Betrag extrahieren. Rechnung 4711 über 99 EUR" + + +@pytest.fixture +def router(registry, library, embedder): + return Router(registry, library, embedder) + + +def test_routes_to_weak_slot_with_adapter(router): + d = router.route(RECHNUNG) + assert d.status == "routed" + assert (d.stage, d.slot, d.skill) == (1, "tier1-general", "rechnung-extrahieren") + assert d.model == d.adapter == "rechnungen-v1" + assert d.prompt_variant == "full" + + +def test_unknown_request_goes_to_strong(router): + d = router.route("Wie ist das Wetter morgen in Berlin?") + assert (d.stage, d.slot, d.skill) == (0, "tier0-strong", None) + assert Signal.NO_SKILL in d.signals + + +def test_draft_skill_not_routed(router): + d = router.route("Gedicht über Rechnung schreiben") + assert d.skill != "entwurf" + + +def test_demoted_skill_runs_on_strong_with_hint(router): + d = router.route("Vertrag prüfen auf Kündigungsfrist") + assert (d.stage, d.skill, d.prompt_variant) == (0, "vertrag-pruefen", "full") + + +def test_classifier_and_program_stages(router): + assert router.route("Ticket: Drucker druckt nicht").stage == 3 + d = router.route("Text in Großbuchstaben: hallo welt") + assert (d.stage, d.slot) == (2, None) + + +def test_unavailable_slot_uses_fallback(router, registry): + registry.set_status("tier1-general", "degraded") + d = router.route(RECHNUNG) + assert d.slot == "tier1-flex" + assert Signal.SLOT_UNAVAILABLE in d.signals + + +def test_all_weak_slots_unavailable_goes_strong(router, registry): + registry.set_status("tier1-general", "unavailable") + registry.set_status("tier1-flex", "unavailable") + d = router.route(RECHNUNG) + assert d.stage == 0 and d.skill == "rechnung-extrahieren" + + +def test_long_input_uses_compact_then_larger_slot(router): + # Eingabe ~1585 Tokens: mit Beispielen (64) zu groß für 2000, kompakt (38) passt es + filler = "x" * (3 * 1585 - len(RECHNUNG) - 1) + d = router.route(RECHNUNG + " " + filler) + assert d.slot == "tier1-general" and d.prompt_variant == "compact" + assert Signal.COMPACT_PROMPT in d.signals + + d = router.route(RECHNUNG + " " + "x" * (3 * 5000)) + assert d.slot == "tier1-flex" + assert Signal.LARGER_SLOT in d.signals + + +def test_too_large_for_everything(router): + d = router.route(RECHNUNG + " " + "x" * (3 * 120000)) + assert d.status == "too_large" + assert Signal.NEEDS_SPLIT in d.signals + + +def test_context_exceeded_goes_to_strong(router): + d = router.route(RECHNUNG + " " + "x" * (3 * 30000)) + assert d.stage == 0 and Signal.CONTEXT_EXCEEDED in d.signals + + +def test_hint_skill(router): + d = router.route("irgendwas", hint_skill="rechnung-extrahieren") + assert d.skill == "rechnung-extrahieren" and d.similarity is None + with pytest.raises(KeyError): + router.route("x", hint_skill="entwurf") + + +def test_escalation_quality_goes_to_strong(router): + d = router.route(RECHNUNG) + e = router.escalate(d, Signal.VERIFIER_FAILED, RECHNUNG) + assert (e.stage, e.slot, e.attempt) == (0, "tier0-strong", 1) + f = router.escalate(e, Signal.VERIFIER_FAILED, RECHNUNG) + assert f.status == "failed" + + +def test_escalation_infra_uses_fallback_slot(router): + d = router.route(RECHNUNG) + e = router.escalate(d, Signal.ENDPOINT_ERROR, RECHNUNG) + assert (e.stage, e.slot) == (1, "tier1-flex") + f = router.escalate(e, Signal.ENDPOINT_ERROR, RECHNUNG) + assert f.stage == 0 + + +def test_escalation_from_program_to_weak(router): + d = router.route("Text in Großbuchstaben: hallo welt") + e = router.escalate(d, Signal.PROGRAM_ERROR, "Text in Großbuchstaben: hallo welt") + assert e.stage == 1 + + +def test_max_escalations(registry, library, embedder): + r = Router(registry, library, embedder, config=RouterConfig(max_escalations=1)) + d = r.route(RECHNUNG) + e = r.escalate(d, Signal.ENDPOINT_ERROR, RECHNUNG) + assert r.escalate(e, Signal.ENDPOINT_ERROR, RECHNUNG).status == "failed" + + +class PreferRechnung: + def rerank(self, query, documents): + return [1.0 if "Rechnung" in d else 0.0 for d in documents] + + +def test_conflict_without_reranker_goes_strong(registry, library, embedder): + r = Router(registry, library, embedder, config=RouterConfig(conflict_margin=1.0)) + d = r.route(RECHNUNG) + assert d.stage == 0 and Signal.SKILL_CONFLICT in d.signals + + +def test_conflict_with_reranker_trusts_reranker(registry, library, embedder): + r = Router(registry, library, embedder, reranker=PreferRechnung(), config=RouterConfig(conflict_margin=1.0)) + d = r.route("Vertrag prüfen Kündigungsfrist. Rechnung auslesen Rechnungsnummer Betrag") + assert Signal.SKILL_CONFLICT in d.signals + assert "Reranking aktiv" in d.reasons + assert d.skill == "rechnung-extrahieren" and d.stage == 1