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:

```mermaid
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.

```mermaid
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).

```mermaid
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

```mermaid
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:

```json
{
  "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:

```json
{
  "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

```yaml
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

```json
{
  "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:

```json
{
  "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)](canonical-control-library.md) — Domains, Datenmodell, Lizenz-Gate
- [Deduplizierungs-Engine](dedup-engine.md) — Multi-Parent-Linking, Review-Queue
- [Obligations v2 (CP-OBL)](obligations.md) — Obligation-Datenmodell
- [Normative Verbindlichkeit](normative-verbindlichkeit.md) — Dreistufenmodell (Pflicht/Empfehlung/Kann)
- [Evidence Type](evidence-type.md) — Code vs. Prozess Evidence