breakpilot-compliance/docs-src/services/sdk-modules/control-generator-pipeline.md

Control Generator Pipeline

Automatische Generierung, Zerlegung und Bewertung von Compliance Controls aus dem RAG-Korpus (~105.000 Chunks aus EU/DE-Gesetzen, Verordnungen und Standards).

!!! info "Pipeline-Standort (seit April 2026)" Die gesamte Pipeline liegt im breakpilot-core Repository unter control-pipeline/. Port: 8098 · Container: bp-core-control-pipeline

---

Gesamtueberblick

Die Pipeline verwandelt Rohtexte aus Regulierungen in pruefbare, atomare Compliance Controls mit Abhaengigkeitslogik. Sie besteht aus 6 Hauptphasen:

flowchart TD
    A["Phase 1: RAG Ingestion"] -->|105k Chunks| B["Phase 2: Control Generation (7 Stufen)"]
    B -->|~33k Rich Controls| C["Phase 3: Obligation Extraction (Pass 0a)"]
    C -->|~181k Obligations| D["Phase 4: Atomic Composition (Pass 0b)"]
    D -->|~15-25k Atomic Controls| E["Phase 5: Deduplizierung"]
    E -->|~15-25k Unique| F["Phase 6: Dependency Engine + Evaluation"]

Phase	Name	Input	Output	Methode
1	RAG Ingestion	PDF/HTML Dokumente	105k Chunks in Qdrant	Embedding + Chunking
2	Control Generation	RAG Chunks	~33k Rich Controls	Claude Sonnet + Lizenz-Gate
3	Pass 0a	Rich Controls	~181k Obligations	Claude Haiku (Batch API)
4	Pass 0b	Obligations	Atomic Controls	Claude Sonnet (Batch API) + Pre-LLM Filter
5	Deduplizierung	Atomic Controls	Unique Master Controls	Embedding + LLM Dedup
6	Dependencies	Master Controls	Evaluierte Controls	Ontology + Pattern + Domain Packs

---

Phase 1: RAG Document Ingestion

Dokumente werden in den RAG-Service (Qdrant) ingestiert und in Chunks zerlegt.

Qdrant Collections

Collection	Inhalte	Lizenzregel	Chunks
bp_compliance_gesetze	Deutsche Gesetze (BDSG, TTDSG, TKG)	Rule 1	~45k
bp_compliance_datenschutz	Datenschutz-Leitlinien, EU-Verordnungen	Rule 1/2	~25k
bp_compliance_ce	CE/Sicherheitsstandards, Maschinenverordnung	Rule 1/2/3	~20k
bp_dsfa_corpus	DSFA-spezifische Dokumente	Rule 1/2	~10k
bp_legal_templates	Rechtsvorlagen	Rule 1	~5k

Chunk-Metadaten

Jeder Chunk in Qdrant hat:

• regulation_id — Eindeutige Regulations-ID
• regulation_name_de — Deutscher Name
• article, paragraph — Strukturreferenz
• chunk_text — Der eigentliche Text
• download_url — Quelldokument-Link

---

Phase 2: Control Generation (7-Stufen-Pipeline)

Datei: control-pipeline/services/control_generator.py

Aus RAG-Chunks werden eigenstaendige "Rich Controls" erzeugt. Dies ist die aelteste Pipeline-Stufe.

flowchart TD
    A[1. RAG Scan] -->|Alle Chunks laden| B[2. License Classify]
    B -->|Rule 1/2| C[3a. Structure Batch]
    B -->|Rule 3| D[3b. Reform Batch]
    C --> E[4. Harmonize]
    D --> E
    E -->|Duplikat| F[Als Duplikat markieren]
    E -->|Neu| G[5. Anchor Search]
    G --> H[6. Store Control]
    H --> I[7. Mark Processed]

Stufe 1: RAG Scan

Laedt unverarbeitete Chunks aus Qdrant via Scroll-API. Filtert per SHA-256-Hash gegen canonical_processed_chunks (keine Doppelverarbeitung).

Stufe 2: License Classify (3-Regel-System)

Regel	Typ	Original erlaubt?	Beispiele
Rule 1 (free_use)	EU-Gesetze, NIST, DE-Gesetze	Ja	DSGVO, BDSG, NIS2, AI Act
Rule 2 (citation_required)	CC-BY, CC-BY-SA	Ja, mit Zitation	OWASP ASVS, OWASP SAMM
Rule 3 (restricted)	Proprietaer	Nein, volle Reformulierung	BSI TR-03161, ISO 27001

!!! warning "Keine Normtexte reproduzieren" Rule 3 Quellen werden VOLLSTAENDIG reformuliert. Keine Originaltexte, keine Quellennamen im Output.

Stufe 3a/3b: Strukturierung / Reformulierung

• Rule 1+2 → _structure_batch(): Originaltext als Control strukturieren
• Rule 3 → _reformulate_batch(): Vollstaendig reformulieren

Batch Processing: 5 Chunks pro Claude API-Call. Ergebnis: JSON-Array mit null fuer irrelevante Chunks.

Stufe 4: Harmonize (Embedding-Dedup)

Cosine Similarity > 0.85 via Qdrant → Duplikat. Zwei-Tier:

• Score >= 0.92: Automatisch als Duplikat markiert
• Score 0.85-0.92: Lokales LLM (qwen3.5) verifiziert

Stufe 5: Anchor Search

Findet Open-Source-Referenzen (OWASP, NIST, ENISA) via Qdrant-Vektorsuche gegen offene Collections.

Stufe 6+7: Store + Mark Processed

Persistiert Controls in canonical_controls mit allen Metadaten. Markiert jeden Chunk als verarbeitet.

Aktueller Stand

Metrik	Wert
RAG-Chunks gesamt	~105.000
Generierte Rich Controls	~32.793 (draft)
Pipeline-Version	3 (mit Scoped Applicability)

---

Phase 3: Pass 0a — Obligation Extraction

Datei: control-pipeline/services/decomposition_pass.py Funktion: DecompositionPass.submit_batch_pass0a()

Extrahiert individuelle normative Pflichten aus Rich Controls via Claude Haiku (Anthropic Batch API, 50% Kostenreduktion).

flowchart LR
    A[Rich Control] -->|Claude Haiku| B[Obligations]
    B --> C{Quality Gate}
    C -->|Pass| D[validated]
    C -->|Fail| E[rejected]

3-Tier Klassifikation

Typ	Signal	Beispiel
Pflicht	muss, ist verpflichtet, hat sicherzustellen	"Der Verantwortliche MUSS ein Verzeichnis fuehren"
Empfehlung	soll, sollte, wird empfohlen	"Es SOLLTE eine Risikobewertung durchgefuehrt werden"
Kann	kann, darf, ist berechtigt	"Die Aufsichtsbehoerde KANN Geldbussen verhaengen"

Quality Gate (6 Regeln)

1. Nur normative Aussagen (muss, sicherzustellen, verpflichtet)
2. Ein Hauptverb pro Obligation
3. Test-Obligations separat von operativen
4. Reporting-Obligations separat
5. Nicht auf Evidence-Ebene splitten
6. Parent-Link immer erhalten

Obligation-Felder

Feld	Beschreibung
obligation_text	Vollstaendiger Pflichttext
action	Hauptverb (z.B. "implementieren")
object	Gegenstand (z.B. "Rate-Limiting")
condition	Bedingung (optional)
normative_strength	must / should / may
is_test_obligation	Testpflicht?
is_reporting_obligation	Meldepflicht?
trigger_type	event / periodic / continuous

Aktueller Stand

Metrik	Wert
Validierte Obligations	~180.827
Obligations pro Rich Control	~5-10

---

Phase 4: Pass 0b — Atomic Control Composition

Datei: control-pipeline/services/decomposition_pass.py Funktion: DecompositionPass.submit_batch_pass0b()

Verwandelt jede validierte Obligation in ein eigenstaendiges atomares Control via Claude Sonnet (Batch API).

Pre-LLM Filter (Ontology)

Datei: control-pipeline/services/control_ontology.py

Vor dem LLM-Call werden Obligations klassifiziert um Kosten zu sparen:

Routing	Beschreibung	Aktion
atomic	Echte atomare Pflicht	An LLM senden
composite	Container-Obligation (z.B. "Sichere Sitzungsverwaltung umsetzen")	Ueberspringen
evidence	Nachweis-Pflicht (z.B. "Screenshot der MFA-Konfiguration")	Ueberspringen
framework_container	Framework-Referenz (z.B. "OWASP ASVS V3 umsetzen")	Ueberspringen

Control Ontology (26 Action Types)

Action Type	Phase	Deutsche Aliases
define	definition	definieren, festlegen, bestimmen
implement	implementation	implementieren, umsetzen, einfuehren
configure	configuration	konfigurieren, einstellen
enforce	implementation	durchsetzen, erzwingen
monitor	monitoring	ueberwachen, beobachten
test	testing	testen, Test durchfuehren
review	review	ueberpruefen, pruefen
assess	assessment	bewerten, beurteilen
train	training	schulen, unterweisen
report	reporting	melden, berichten
prevent	implementation	verhindern, unterbinden
exclude	implementation	ausschliessen, verbieten
...	...	(26 Typen gesamt, siehe control_ontology.py)

MCP-taugliche Output-Felder

Jedes atomare Control hat Felder die MCP-Scanner verwenden koennen:

Feld	Beschreibung	Beispiel
assertion	Pruefbare Aussage	"Alle API-Endpunkte haben Rate-Limiting konfiguriert"
pass_criteria	Wann erfuellt?	["Rate-Limit < 100 req/min", "429 Response bei Ueberschreitung"]
fail_criteria	Wann nicht erfuellt?	["Kein Rate-Limiting konfiguriert", "Unbegrenzter Durchsatz"]
check_type	Pruefmethode	technical_config_check / document_clause_check / code_pattern_check
merge_key	Dedup-Key	"implement:api_rate_limiting:implementation"
dependency_hints	Abhaengigkeits-Hinweise	["prerequisite:define:api_security_policy"]
lifecycle_phase_order	Lebenszyklusphase 1-13	4 (implementation)

Canonical Key Format

action_type:normalized_object:control_phase[:asset_scope]

Beispiel: implement:api_rate_limiting:implementation:api_endpoints

Aktueller Stand

Metrik	Wert
Verarbeitete Obligations (Test)	~1.000
Erzeugte atomare Controls	562
Merge-Key Coverage	99.8%
MCP-Feld Coverage	100% (neue Batches)

---

Phase 5: Deduplizierung

Datei: control-pipeline/services/control_generator.py (Harmonization)

3-Stufen Entscheidungsbaum

flowchart TD
    A[Atomic Control] --> B{Merge-Key gleich?}
    B -->|Ja| C{Embedding Similarity}
    B -->|Nein| D[NEW CONTROL]
    C -->|>= 0.92| E[AUTO-LINK]
    C -->|0.85 - 0.92| F{LLM Verify}
    C -->|< 0.85| D
    F -->|Duplikat| E
    F -->|Verschieden| D

1. Merge-Key Grouping: Controls mit gleichem action_type:normalized_object werden verglichen
2. Embedding Similarity: bge-m3 Embeddings, Cosine Distance
3. LLM Verification: Lokales qwen3.5 (think:false) fuer Grenzfaelle (0.85-0.92)

Multi-Parent-Linking

Ein atomares Control kann aus mehreren Regulierungen stammen:

{
  "control_id": "AUTH-1072",
  "title": "MFA fuer privilegierte Accounts aktiviert",
  "parent_links": [
    {"source": "NIST IA-02(01)", "regulation": "nist_sp_800_53"},
    {"source": "NIS2 Art. 21", "regulation": "eu_2022_2555"},
    {"source": "DSGVO Art. 32", "regulation": "eu_2016_679"}
  ]
}

---

Phase 6: Dependency Engine + Evaluation

Datei: control-pipeline/services/dependency_engine.py Generator: control-pipeline/services/dependency_generator.py Domain Packs: control-pipeline/data/domain_packs/*.yaml

Controls sind nicht unabhaengig — sie haben logische Abhaengigkeiten. Die Dependency Engine modelliert diese generisch.

5 Dependency-Typen

Typ	Prioritaet	Beschreibung	Beispiel
supersedes	10 (staerkste)	A ersetzt B vollstaendig	GHV-Klausel im Vertrag → Schulung nicht noetig
scope_exclusion	20	A schliesst B aus dem Scope aus	"Kein KI-Einsatz" → KI-Controls not_applicable
prerequisite	50	A muss vor B existieren	Policy definieren → Policy implementieren
conditional_requirement	70	B nur unter Bedingung X	Enhanced Logging nur fuer grosse Unternehmen
compensating_control	80 (schwaechste)	A mildert B-Failure	ISO-Zertifizierung kompensiert Einzelcontrols

Evaluation-Status

Status	Bedeutung
pass	Control erfuellt
fail	Control nicht erfuellt
not_applicable	Nicht anwendbar (durch Dependency oder Scope)
partially_satisfied	Teilweise erfuellt
compensated_fail	Nicht erfuellt, aber kompensiert
review_required	Manuelle Pruefung noetig (z.B. bei Zyklen)

Generische Condition Language

Dependencies verwenden eine regelbasierte Sprache:

{
  "operator": "AND",
  "clauses": [
    {"field": "source.status", "op": "==", "value": "pass"},
    {"field": "context.company_size", "op": "in", "value": ["large", "enterprise"]}
  ]
}

Felder: source.status, target.status, context.industry, context.company_size, context.scope_signals Operatoren: ==, !=, in, not_in, >, <, >=, <=, AND, OR, NOT

Evaluation-Algorithmus

1. Adjacency-List bauen (target → Dependencies)
2. Zyklen erkennen (DFS) → betroffene Controls = review_required
3. Topologisch sortieren (Kahn's Algorithmus)
4. Fuer jedes Control: Conditions pruefen, Effects sammeln, Priority-Konflikt loesen
5. Dependency-Trace fuer MCP-Output aufzeichnen

Automatische Dependency-Generierung

3 Strategien:

1. Ontology-basiert: Controls mit gleichem normalized_object und aufeinanderfolgenden Phasen → prerequisite
2. Pattern-basiert: Regeln wie "define → implement", "implement → monitor"
3. Domain Packs (YAML): DSGVO, AI Act, CRA, Security, Arbeitsrecht

Beispiel: GHV-Szenario

MC-001: GHV-Klausel im Arbeitsvertrag vorhanden
MC-002: Vertraulichkeitsschulung durchgefuehrt
MC-003: Jaehrliche Nachschulung durchgefuehrt

Dependencies:
  MC-001 supersedes MC-002 (wenn MC-001 = pass)
  MC-002 prerequisite MC-003

Fall 1: Vertrag hat GHV → MC-002 + MC-003 = not_applicable (kein unnoeiger Aufwand) Fall 2: Vertrag fehlt, Schulung vorhanden → MC-002 = pass, MC-003 erforderlich

MCP-Output mit Dependency-Trace

{
  "control_id": "MC-002",
  "raw_status": "fail",
  "resolved_status": "not_applicable",
  "dependency_resolution": [
    {
      "dependency_type": "supersedes",
      "source_control_id": "MC-001",
      "source_status": "pass",
      "condition_met": true,
      "effect_applied": {"set_status": "not_applicable"},
      "priority": 10
    }
  ]
}

---

Phase C2: Applicability Engine

Datei: control-pipeline/services/applicability_engine.py

Filtert Controls basierend auf dem Unternehmensprofil (Branche, Groesse, Scope-Signale).

Zwei-Schicht-Filterung

1. Deterministischer Filter (schnell): SQL + Python Pre-Filter auf applicable_industries, applicable_company_size, scope_conditions
2. Assessment Layer: Confidence-Scoring, Eskalationserkennung, Reasoning

Scope-Signale

Controls koennen scope_conditions haben, die bestimmte Unternehmenssignale erfordern:

{
  "scope_conditions": {
    "requires_any": ["has_ai_system", "uses_third_party_ai"]
  }
}

Wenn das Unternehmen keines dieser Signale hat → Control nicht anwendbar.

---

Document Compliance Engine

Datei: control-pipeline/services/document_scope_resolver.py

Bestimmt welche rechtlichen Dokumente ein Unternehmen braucht, basierend auf Business-Flags:

Flag	Erforderliches Dokument
has_website	Impressum, Datenschutzerklaerung
uses_cookies	Cookie-Banner, Cookie-Richtlinie
sells_online	AGB, Widerrufsbelehrung, Preisangabenverordnung
processes_personal_data	DSGVO Verarbeitungsverzeichnis
uses_ai_system	KI-Transparenzhinweis
sells_electronics	WEEE-Entsorgungshinweis

---

Qualitaetsmetriken

Endpoint: GET /v1/canonical/generate/quality-metrics

Metrik	Zielwert	Aktuell
Controls pro Obligation	0.2-0.4	0.56 (vor Dedup)
Duplikat-Merge-Key-Rate	< 5%	0.2%
Evidence-Leak-Rate	< 2%	0.0%
Truncated-Title-Rate	0%	0.0%
Merge-Key Coverage	100%	99.8%
MCP-Feld Coverage	100%	100% (neue Batches)

---

API Endpoints

Alle Endpoints unter dem Control Pipeline Service (Port 8098).

Control Generation

Methode	Pfad	Beschreibung
POST	/v1/canonical/generate	Generierungs-Job starten
GET	/v1/canonical/generate/status/{job_id}	Job-Status abfragen
GET	/v1/canonical/generate/quality-metrics	Qualitaetsmetriken
POST	/v1/canonical/generate/submit-pass0b	Pass 0b Batch einreichen
POST	/v1/canonical/generate/process-batch	Batch-Ergebnisse verarbeiten
GET	/v1/canonical/generate/batch-api-status/{id}	Anthropic Batch Status

Controls CRUD

Methode	Pfad	Beschreibung
GET	/v1/canonical/controls	Alle Controls (filterbar)
GET	/v1/canonical/controls/{id}	Einzelnes Control
POST	/v1/canonical/controls/applicable	Applicability-Filter
GET	/v1/canonical/controls/{id}/traceability	Traceability-Chain

Dependencies

Methode	Pfad	Beschreibung
GET	/v1/dependencies	Dependencies auflisten
POST	/v1/dependencies	Dependency erstellen
POST	/v1/dependencies/generate	Auto-Generierung
POST	/v1/dependencies/evaluate	Controls evaluieren
POST	/v1/dependencies/validate	Zyklen pruefen
GET	/v1/dependencies/graph	Graph fuer Visualisierung

Document Compliance

Methode	Pfad	Beschreibung
POST	/v1/document-compliance/required	Erforderliche Dokumente

---

Datenbanktabellen

Tabelle	Schema	Beschreibung
canonical_controls	compliance	Alle Controls (rich + atomic)
obligation_candidates	compliance	Extrahierte Obligations (Pass 0a)
control_parent_links	compliance	M:N Parent-Child-Beziehungen
control_dependencies	compliance	Abhaengigkeiten zwischen Controls
control_evaluation_results	compliance	Evaluierungsergebnisse mit Traces
canonical_processed_chunks	compliance	Verarbeitete RAG-Chunks
canonical_generation_jobs	compliance	Generierungs-Jobs

---

Quelldateien

Datei	Repo	Beschreibung
services/control_generator.py	core	7-Stufen Control Generation Pipeline
services/decomposition_pass.py	core	Pass 0a + 0b (Obligation → Atomic Control)
services/control_ontology.py	core	26 Action Types, Phase Ordering, Pre-LLM Filter
services/dependency_engine.py	core	Dependency Evaluation (5 Typen, Cycle Detection)
services/dependency_generator.py	core	Auto-Generierung (Ontology + Pattern + Domain Packs)
services/applicability_engine.py	core	Scope-basierte Applicability-Filterung
services/document_scope_resolver.py	core	28 Document Compliance Rules
services/anchor_finder.py	core	Open-Source Anchor Search (Qdrant)
data/domain_packs/*.yaml	core	5 Domain Packs (GDPR, AI Act, CRA, Security, Arbeitsrecht)
tests/test_dependency_engine.py	core	39 Tests (Dependency Engine)
tests/test_dependency_generator.py	core	14 Tests (Auto-Generierung)
tests/test_golden_controls.py	core	37 Tests (Golden Test Suite, alle 8 Kategorien)
tests/test_control_ontology.py	core	36 Tests (Ontology-Klassifikation)

Tests gesamt: 126+ (alle bestanden)

---

Verwandte Dokumentation

• Canonical Control Library (CP-CLIB) — Domains, Datenmodell, Lizenz-Gate
• Deduplizierungs-Engine — Multi-Parent-Linking, Review-Queue
• Obligations v2 (CP-OBL) — Obligation-Datenmodell
• Normative Verbindlichkeit — Dreistufenmodell (Pflicht/Empfehlung/Kann)
• Evidence Type — Code vs. Prozess Evidence