"""Offline teaching fixtures; Python 3.12+, standard library only. These fabricated documents and scores illustrate contracts, not model quality. Run: python contract_lab.py """ from dataclasses import dataclass from hashlib import sha256 import asyncio import json import math import re DOCUMENTS = ( {"id": "manual-e17-v1", "text": "E17 냉각수 유량 저하. 첫 점검은 필터 막힘 확인.", "group": "equipment"}, {"id": "manual-e18-v1", "text": "E18 센서 통신 오류. 첫 점검은 케이블 연결 확인.", "group": "equipment"}, {"id": "fruit-v1", "text": "사과 보관 안내.", "group": "public"}, {"id": "restricted-v1", "text": "E17 비공개 정비 기록.", "group": "restricted"}, ) def terms(text): # Lexical toy analyzer, NOT a model tokenizer or Korean morphological analyzer. return re.findall(r"[a-z0-9]+|[가-힣]+", text.lower()) def search(query, *, groups, k=2): if k < 1: raise ValueError("k must be positive") query_terms = set(terms(query)) candidates = [] for document in DOCUMENTS: if document["group"] not in groups: continue score = len(query_terms & set(terms(document["text"]))) if score: candidates.append({**document, "score": score}) return sorted(candidates, key=lambda item: (-item["score"], item["id"]))[:k] class ExtractiveFakeGenerator: """Deterministic adapter: copies first authorized hit; does not call an LLM.""" def generate(self, hits): if not hits: return {"status": "insufficient_evidence", "answer": "", "citations": []} return {"status": "answered", "answer": hits[0]["text"], "citations": [hits[0]["id"]]} def validate_answer(answer, hits): allowed = {hit["id"] for hit in hits} citations = set(answer["citations"]) if not citations <= allowed: raise ValueError("unknown citation") if answer["status"] == "answered" and not citations: raise ValueError("answered needs a citation") # This checks reference identity, not claim entailment. def rag(query): hits = search(query, groups={"equipment", "public"}) answer = ExtractiveFakeGenerator().generate(hits) validate_answer(answer, hits) return {"hit_ids": [hit["id"] for hit in hits], **answer} def hit_and_recall(retrieved, relevant): if not relevant: raise ValueError("unanswerable is a separate task") found = set(retrieved) & set(relevant) return {"hit": int(bool(found)), "recall": len(found) / len(set(relevant))} def index_recall(exact, approximate): if not exact or len(exact) != len(approximate): raise ValueError("equal non-empty rankings required") if any(i < 0 for i in exact) or len(set(exact)) != len(exact): raise ValueError("exact ranking must contain k distinct valid IDs") valid_approximate = {i for i in approximate if i >= 0} return len(set(exact) & valid_approximate) / len(exact) def mmr(order, relevance, similarity, weight, count=2): selected = [] remaining = list(order) while remaining and len(selected) < count: def key(item): redundancy = max((similarity[item][old] for old in selected), default=0.0) return (weight * relevance[item] - (1 - weight) * redundancy, relevance[item], -order.index(item)) chosen = max(remaining, key=key) selected.append(chosen) remaining.remove(chosen) return selected def plan_order(dependencies): unknown = set().union(*map(set, dependencies.values())) - set(dependencies) if unknown: raise ValueError("unknown dependency") pending = dict(dependencies) done = [] while pending: ready = sorted(key for key, needs in pending.items() if set(needs) <= set(done)) if not ready: raise ValueError("cycle") for key in ready: done.append(key) del pending[key] return done @dataclass(frozen=True) class Deadline: expires: float def require_remaining(self, *, now): if now >= self.expires: raise TimeoutError("deadline expired") class FakeRetrieval: def __init__(self, *, now=0.0): self.now = now self.calls = [] async def search(self, request, *, deadline): deadline.require_remaining(now=self.now) self.calls.append(request) return search(request["query"], groups=request["groups"]) async def run_read_proposals(proposals, registry, *, deadline, now, budget): for name, arguments in proposals: deadline.require_remaining(now=now) if name not in registry: return "invalid_proposal" if budget <= 0: return "tool_budget_exhausted" budget -= 1 await registry[name](arguments, deadline=deadline) return "completed" def mask_toy_conversation(): # Fabricated token vocabulary: this is not any provider's chat template. tokens = ["", "질문", "", "", "답", ""] labels = [-100, -100, -100, -100, 4, 5] return [{"position": i, "token": token, "label": label} for i, (token, label) in enumerate(zip(tokens, labels))] def training_example(): logits = [2.0, 1.0, 0.0] # Positive first; two deliberately fabricated negatives. probability = math.exp(logits[0]) / sum(math.exp(x) for x in logits) gradient = [math.exp(x) / sum(math.exp(y) for y in logits) - int(i == 0) for i, x in enumerate(logits)] updated = [x - 0.1 * g for x, g in zip(logits, gradient)] updated_probability = math.exp(updated[0]) / sum(math.exp(x) for x in updated) return {"positive_probability": round(probability, 6), "loss_before": round(-math.log(probability), 6), "loss_after": round(-math.log(updated_probability), 6)} def outputs(): relevance = {"A": 0.95, "B": 0.90, "C": 0.75, "D": 0.20} similarity = { "A": {"A": 1, "B": .95, "C": .10, "D": .05}, "B": {"A": .95, "B": 1, "C": .10, "D": .05}, "C": {"A": .10, "B": .10, "C": 1, "D": .05}, "D": {"A": .05, "B": .05, "C": .05, "D": 1}, } return { "kind": "fabricated-contract-fixture-not-a-benchmark", "rag": rag("E17 원인"), "no_answer": rag("망고 재배"), "metrics": hit_and_recall(["g1"], {"g1", "g2", "g3"}), "mmr_lambda_1": mmr(list(relevance), relevance, similarity, 1), "mmr_lambda_05": mmr(list(relevance), relevance, similarity, .5), "dag": plan_order({"lookup_a": [], "lookup_b": [], "compare": ["lookup_a", "lookup_b"]}), "contrastive_logit_step": training_example(), "assistant_mask": mask_toy_conversation(), "lora_shape_count": {"target": "toy.q_proj", "out": 8, "in": 4, "rank": 2, "frozen_base_parameters": 8 * 4, "trainable_adapter_parameters": 2 * (8 + 4)}, "fixture_sha256": sha256(json.dumps(DOCUMENTS, ensure_ascii=False, sort_keys=True).encode()).hexdigest(), } if __name__ == "__main__": print(json.dumps(outputs(), ensure_ascii=False, indent=2))