"""Transparent baseline pipeline for the generated AI project.""" from __future__ import annotations import json import math import re from pathlib import Path def tokenize(value: str) -> list[str]: return re.findall(r"[a-z0-9]+", value.lower()) class Pipeline: def __init__(self, model: dict): self.model = model @classmethod def from_file(cls, path: str | Path) -> "Pipeline": return cls(json.loads(Path(path).read_text(encoding="utf-8"))) def classify(self, text: str) -> tuple[str, float]: tokens = tokenize(text) scores = { label: sum(weights.get(token, 0) for token in tokens) for label, weights in self.model["prototypes"].items() } ranked = sorted(scores.items(), key=lambda item: (-item[1], item[0])) label, best = ranked[0] total = sum(max(score, 0) for _, score in ranked) or 1 return label, best / total def search(self, query: str, limit: int = 3) -> list[dict]: query_tokens = set(tokenize(query)) ranked = [] for document in self.model["documents"]: document_tokens = set(tokenize(document["text"])) lexical = sum( self.model["idf"].get(token, 1.0) for token in query_tokens & document_tokens ) ranked.append({**document, "score": round(lexical, 6)}) return sorted(ranked, key=lambda item: (-item["score"], item["id"]))[:limit] def graph_evidence(self, text: str) -> list[dict]: tokens = set(tokenize(text)) matches = [] for subject, relation, target in self.model.get("graph_edges", []): edge_tokens = set(tokenize(f"{subject} {relation} {target}")) overlap = len(tokens & edge_tokens) if overlap: matches.append( { "subject": subject, "relation": relation, "target": target, "overlap": overlap, } ) return sorted(matches, key=lambda item: -item["overlap"]) def run(self, text: str) -> dict: label, confidence = self.classify(text) evidence = self.search(text) result = { "prediction": label, "confidence": round(confidence, 4), "requires_review": confidence < self.model["confidence_threshold"], "evidence": evidence, } if self.model["mode"] == "graph": result["graph_evidence"] = self.graph_evidence(text) if self.model["mode"] == "agent": result["proposed_tool"] = label result["approval_required"] = label in { "request-approval", "request-human-help", } return result