EMIL Hexagonal Architecture & Inferenz-Interaktion

1. Die Hexagonale Architektur (Ports & Adapters)

EMIL trennt die reine neuronale Kognition strikt von allen externen Protokollen. Dadurch bleibt der Denkapparat 100 % unabhängig davon, ob Daten über WhatsApp, SAP, Telegram oder eine OPC-UA SPS einfließen.

Abb. 1: 3D-Struktur der Hexagonalen Ports & Adapters Architektur von EMIL AI

2. Ports vs. Adapters: Der fundamentale Unterschied

In der hexagonalen Architektur (Clean Architecture) ist die klare Rollenverteilung zwischen abstrakten Verträgen (Ports) und konkreten Implementierungen (Adapters) entscheidend:

3. Multi-Container Parallelität (0 MB VRAM)

Alle montierten AI Container laufen 100 % parallel im neuronalen Kortex! EMIL führt AI Container nicht nacheinander aus. Stattdessen projiziert der Prismatic Router den eingehenden Kognitions-Tensor zeitgleich in alle aktiven AI Container Module:

Abb. 2: Parallele Multi-Vault Inferenz & Vektor-Fusion im 512D Kognitionsraum

4. Wie verhalten sich Adapter während der Inferenz?

Es gibt genau zwei fundamentale Interaktionsmuster zwischen EMILs Inferenz und den Adaptern:

5. Das selektive Pausieren (Cognitive Fakten-Yield)

Die neuronale Logik rechnet weiter – nur der Satz-Schreiber wartet auf die Zahl!

Die Aussage „die Inferenz pausiert“ bezieht sich ausschließlich auf den finalen Token-Emitter (den Ausgabegenerator) an der Stelle, an der ein externer Echtzeit-Fakt benötigt wird:

Abb. 3: Selektiver 5-Stufen-Zyklus des Cognitive Fakten-Yields & Fakten-Injektion

6. Kann ich innerhalb von Python-Code auf die Inferenz zugreifen?

Ja, uneingeschränkt! EMIL bietet eine First-Class Python API. Sie können Inferenz sowohl auf Text-Ebene als auch direkt auf Vektor-Ebene im 512D Kognitionsraum ausführen:

A. Einfache Text-Kognition (`think`)

from modules.eml_pipeline import emil_engine

# Führt einen vollen kognitiven Denkprozess mit Z3-Prüfung aus
antwort = emil_engine.think(
    prompt="Prüfe Kündigungsfrist für 3 Jahre Betriebszugehörigkeit nach § 622 BGB",
    temperature=0.1
)
print(antwort)

B. Reine Vektor-Inferenz im Kognitionsraum (Latenz < 2 ms)

from modules.eml_pipeline import emil_engine

# Projiziert Text direkt in 512D Kognitions-Vektoren
v_gesetz = emil_engine.embed("KSchG § 1 Wartezeit 6 Monate")
v_anfrage = emil_engine.embed("Gilt Kündigungsschutz nach 7 Monaten Probezeit?")

# Berechnet die semantische Distanz
dist = emil_engine.calculate_distance(v_gesetz, v_anfrage)
if dist < 0.15:
    print("🟢 Relevanz mathematisch bewiesen (Semantischer Treffer)!")

C. Event-Driven Hook im AI Container

from modules.adapters import whatsapp_adapter

# Callback, der von der WhatsApp-Bridge getriggert wird
def on_whatsapp_message(sender_id: str, message_text: str, emil):
    # 1. EMIL denkt mit AI Container Wissen
    gutachten = emil.think(f"Arbeitsrechtliche Prüfung: {message_text}")
    
    # 2. Direkt über den WhatsApp-Adapter mit Buttons antworten
    whatsapp_adapter.send_interactive_buttons(
        recipient=sender_id,
        body=gutachten,
        buttons=[
            {"id": "btn_frist", "title": "📅 Frist berechnen"},
            {"id": "btn_anwalt", "title": "📞 Rückruf anfordern"}
        ]
    )

7. Deterministische Regelprüfung: Das `ProofChain` SDK

Mit der ProofChain-Engine können Entwickler in jedem Adapter, Container-Skript oder Workflow formale, unanfechtbare Prüfungen direkt in Python-if-Bedingungen verpacken. Die Engine verifiziert die Einhaltung aller definierten Geschäfts- und Sicherheitsinvarianten, führt einen mathematischen Z3-SMT Logik-Beweis durch und signiert das Ergebnis bei Bedarf mit einem revisionssicheren NIST FIPS 204 (ML-DSA-65) Quantensiegel.

A. Python-Code: ProofChain in der Praxis

from emil.core.proof import ProofChain, ProofDepth

def verify_pharma_inbound(nfc_data, camera_frame, sap_po):
    # 1. Beweiskette initialisieren
    chain = ProofChain(context="Sana GDP Pharma Inbound", depth=ProofDepth.PQC_SEALED)

    # 2. Container-Fakten verankern (Soll-Grenzwerte)
    f_idx = chain.add_vault_fact("GDP_SOP_8492", temp_min=2.0, temp_max=8.0, min_mhd_days=180)

    # 3. Reale Sensordaten einspeisen
    o1 = chain.add_observation("NFC_Logger", measured_temp=nfc_data.temp_max, measured_mhd=nfc_data.mhd_days)
    o2 = chain.add_observation("Vision_Camera", damage_score=camera_frame.damage_score, seal_intact=True)

    # 4. Kausale Verifikationsschritte
    chain.add_step("Temperatur lückenlos im Sollkorridor (+2°C bis +8°C)", from_node=f_idx, to_node=o1)
    chain.add_step("Plombe und Kartonage unversehrt", from_node=o1, to_node=o2)

    # 5. Invarianten deklarieren
    chain.require_invariants(["temperature_range", "expiry_date", "seal_integrity"])

    # ══════════════════════════════════════════════════════════════════════
    # 6. DIE IF-BEDINGUNG: Formale Regelprüfung vor Ausführung
    # ══════════════════════════════════════════════════════════════════════
    if chain.is_proven():
        # Vollautonome SAP-Buchung & Relais-Entriegelung
        pqc_signature = chain.sign_pqc()  # Erzeugt NIST FIPS 204 (ML-DSA-65) Siegel
        sap_doc = sap_actuator.book_goods_receipt(sap_po, pqc_seal=pqc_signature)
        smart_lock.unlock_gate(gate_id="TOR_04")
        return {"status": "APPROVED", "pqc": pqc_signature, "sap_doc": sap_doc}
    else:
        # Veto & Quarantäne-Alarm
        smart_lock.lock_down()
        return {"status": "REJECTED", "reason": chain.get_failure_cause()}

B. Die 4-Stufige Prüftiefen-Pyramide (`ProofDepth`)

8. Vault-Steuerung via JSON & Native Inferenz im Python SDK

Die saubere Trennung zwischen deklarativem Datenwissen und programmatischer Kognition:

In EMILs Zero-Trust-Architektur ist strikt geregelt, welche Aufgaben deklarativ im Daten-Tresor (AI Container) liegen und welche programmatisch über das Python SDK ausgeführt werden:

A. Für Vault-Ersteller: Grenzwerte & Tabellen im Vault steuern

Jeder Fach-Vault kann bestehende Normgrenzen oder Wissensgraphen dynamisch überschreiben, indem ein deklarativer JSON-Block (oder eine CSV-Tabelle) in die Vault-Quelldatei eingebettet wird:

# In der Markdown- oder EMP-Quelldatei des Fach-Vaults:
[AKTIVER AI CONTAINER: SONDERMONTAGE_WERK_STUTTGART]

# Deklaratives JSON-Steuerobjekt für Grenzwert-Overlays:
{
  "tolerances": {
    "Welle_D50_h6": [-0.016, 0.0],
    "Bohrung_D50_H7": [0.0, 0.025]
  }
}

Sobald dieser Vault montiert wird, liest die Inferenz-Pipeline (_apply_vault_tables_to_graph) den JSON-Block aus, identifiziert über topologische Token-Resonanz die betroffenen IntervalNode-Knoten und passt deren upper_bound dynamisch an – ohne dass der Python-Kern neu gestartet oder kompiliert werden muss.

B. Für Python-Entwickler: Native Inferenz im Python SDK nutzen

Das Python SDK ermöglicht den direkten, CPU-autarken Zugriff auf alle 4 Säulen der nativen Inferenz-Kaskade (0 MB VRAM, pure Mathematik):

from modules.nlp import get_native_processor
from modules.nlp.knowledge_graph import KnowledgeGraph, IntervalNode

processor = get_native_processor()

# 1. Säule: Safe AST Math & physikalische Einheiten (ohne unsicheres eval())
res_math = processor.synthesize("10 bar in psi")
print(res_math)
# ➔ "10 bar entsprechen 145,037738 psi."

# 2. Säule: Astronomischer Temporal-Kernel (Gauß-Osterformel & BGB-Werktagsverschiebung)
res_cal = processor.synthesize("12.09.2026 + 21 Werktage")
print(res_cal)
# ➔ Berechnet Feiertag (03.10. Tag d. Dt. Einheit / Sa) und verschiebt transparent auf Mo, 05.10.2026

# 3. Säule: Normen-agnostische Intervallprüfung mit Vault-Overlay
vault_context = '''
[AKTIVER AI CONTAINER: SPEZIAL_ENGINEERING]
{"tolerances": {"Welle_D50_h6": [-0.016, 0.0]}}
'''
reply = processor.synthesize(
    "Fertigungsmaß Welle liegt bei 49.992 mm. Liegt das im Toleranzband?",
    vault_text=vault_context
)
print(reply)
# ➔ "Das Fertigungsmaß von 49.992 mm liegt innerhalb des ISO-Toleranzfeldes (-0.016 mm bis 0.0 mm). Sicherheitsreserve von 0.008 mm."

C. System-Schichten & Signalfluss im Enterprise-Einsatz

9. 10-Minuten Tutorial: Eigenen Inhouse-Adapter schreiben

Jeder Entwickler kann in unter 10 Minuten einen neuen Adapter für jedes Altsystem (z. B. Hochregallager, AS/400, Oracle SQL) schreiben. Er erbt automatisch Circuit Breaking, Post-Quantum Audit-Logging und 4-Augen-Sperre:

# inhouse_warehouse_adapter.py
from modules.adapters import BaseAdapter

class InhouseWarehouseAdapter(BaseAdapter):
    def __init__(self):
        super().__init__(
            adapter_id="adapter.warehouse.custom_rack_system",
            name="Inhouse Hochregallager TCP Adapter",
            system_type="WAREHOUSE_TCP_SOCKET",
            read_only_mode=True # Standardmäßig schreibgeschützt
        )

    def get_rack_status(self, rack_id: str) -> dict:
        self._check_circuit_breaker() # Schutz vor Server-Überlastung
        try:
            # 1. Verbindung zum hauseigenen TCP-Socket / Datenbank
            status = {"rack_id": rack_id, "capacity_percent": 84.5, "temperature": 18.2}
            
            # 2. Revisionssicherer Audit-Eintrag (SHA-256 + ML-DSA-65)
            self.record_audit_entry(
                action="GET_RACK_STATUS", direction="OUTBOUND",
                payload_summary=f"Regal {rack_id} abgefragt: {status['capacity_percent']}% belegt",
                status="SUCCESS"
            )
            self._record_success()
            return status
        except Exception as e:
            self._record_failure(str(e))
            raise

10. AI Container Anbindung & Adapter-Deklaration

Ein autarker AI Container (.emp Container) deklariert in seinem Manifest (vault_manifest.json), welche Adapter er benötigt. Sobald EMIL den AI Container mountet (Hot-Swap in 1,8 ms), bindet der Kortex die entsprechenden Adapter automatisch:

// In vault_manifest.json des AI Container:
{
  "vault_id": "vault.legal.labor.kschg_arbzg",
  "title": "Arbeitsrecht & Kündigungsschutz Vault",
  "required_adapters": [
    "adapter.channel.chat.whatsapp_meta",
    "adapter.channel.chat.telegram_bot"
  ],
  "entrypoint_script": "scripts/on_client_message.py"
}

11. IT-Sicherheit, 4-Augen-Prinzip & ISO 27001

12. Cognitive Harness: Steuerung & Governance von LLM-Agenten (LangGraph, CrewAI, AutoGen)

In modernen Unternehmen entstehen heterogene Multi-Agenten-Ökosysteme. Doch stochastische LLM-Agenten neigen zu Halluzinationen, unkontrollierten Tool-Calls und Endlosschleifen. EMIL fungiert als übergeordneter Cognitive Harness & Metakognitiver Supervisor nach EU AI Act Art. 14: Er zerlegt komplexe Aufgaben, übergibt Teilaufgaben per MCP/REST an externe Agenten und sichert jede resultierende Transaktion mathematisch ab.

# cognitive_agent_harness_example.py
from emil.core.harness import CognitiveAgentHarness
from emil.core.veto import Z3ConstraintGuard

# 1. Harness initialisieren & Z3-Sicherheitsschranken definieren
guard = Z3ConstraintGuard(
    max_transaction_eur=5000.00,
    require_fido2_above=1000.00,
    restricted_tables=["core_payroll", "customer_master_gdpr"]
)
harness = CognitiveAgentHarness(constraint_guard=guard)

# 2. Externe Agenten-Flotte registrieren (CrewAI, LangGraph, n8n-Webhook)
harness.register_subagent(
    agent_id="agent.research.crewai",
    protocol="MCP_CLIENT",
    endpoint="http://localhost:8000/mcp/v1"
)
harness.register_subagent(
    agent_id="agent.workflow.n8n",
    protocol="REST_WEBHOOK",
    endpoint="https://n8n.internal/webhook/order-dispatch"
)

# 3. Kontrollierte Delegation: EMIL steuert, validiert und vollstreckt
def handle_customer_claim(claim_text: str):
    # EMIL analysiert Absicht semantisch & füttert Fakten aus .emp Containern
    intent = harness.semantic_parse(claim_text)
    
    # Rohentwurf an externen CrewAI-Agenten delegieren
    draft = harness.delegate_task(
        target_agent="agent.research.crewai",
        prompt=f"Extrahiere Reklamationsposten: {claim_text}"
    )
    
    # Veto-Prüfung: Stoppt Halluzinationen oder unautorisierte Buchungsvorschläge
    verification = harness.verify_intent_with_z3(draft)
    if verification.passed:
        # Erst nach mathematischer Freigabe triggert EMIL den n8n-Workflow
        harness.trigger_workflow("agent.workflow.n8n", payload=draft.verified_data)
    else:
        harness.escalate_to_human(reason=verification.violation_details)

13. Domain Codecs & Ingestion SDK: Symmetrische Datenverarbeitung (Decode & Encode)

In industriellen Kognitions-Ökosystemen genügt es nicht, Dokumente nur passiv einzulesen (Ingestion). Was ein kognitives Organ imstande ist zu verstehen (Decode / Sensorik), muss es auch deterministisch und normkonform in der physischen oder digitalen Welt erzeugen können (Encode / Lokale Synthese).

Das BaseDomainCodec-Paradigma realisiert diesen geschlossenen Regelkreis: Strukturierte Industriestandards (z. B. XRechnung EN 16931 in UBL 2.1 und UN/CEFACT CII, DIN-Normen, STEP CAD, G-Code) werden verlustfrei in topologische Wissensgraphen mit 8D-Okta-Simplex-Semantik (S₈) überführt und können umgekehrt aus verifizierten Graphenknoten deterministisch synthetisiert werden – mit 0 MB VRAM und unter strikter Wahrung des Zero Network Egress (Zero-Leak Local Loop).

A. Implementierung eines Custom Domain Codecs

Jeder Domain Codec erbt von BaseDomainCodec und überschreibt die beiden Kernmethoden decode() und encode():

# custom_domain_codec.py
from typing import Any, Dict, Optional, Union
from modules.nlp.base_domain_codec import BaseDomainCodec
from modules.knowledge_graph import KnowledgeGraph, KnowledgeNode

class CustomTelemetryCodec(BaseDomainCodec):
    FORMAT_ID = "telemetry.industrial.v1"
    SUPPORTED_STANDARDS = ["OPC-UA", "MQTT-JSON"]

    @classmethod
    def decode(cls, raw_content: Union[str, bytes], kg: KnowledgeGraph, **kwargs) -> Dict[str, Any]:
        # 1. Parse Rohdaten verlustfrei
        data = parse_payload(raw_content)
        
        # 2. Integriere Entitäten in den KnowledgeGraph mit 8D-Koordinaten
        node_id = f"telemetry_{data['sensor_id']}"
        kg.add_node(KnowledgeNode(
            id=node_id,
            label=f"Sensor {data['sensor_id']}",
            attributes=data,
            coordinates_8d=[0.1, 0.8, 0.0, 0.0, 0.0, 0.0, 0.0, 0.1]
        ))
        return {"root_id": node_id, "status": "success"}

    @classmethod
    def encode(cls, kg: KnowledgeGraph, entity_id: Optional[str] = None, options: Optional[Dict[str, Any]] = None, **kwargs) -> Union[str, bytes]:
        # 3. Deterministische Synthese des Zielformats (z. B. normierter Steuerbefehl)
        node = kg.get_node(entity_id)
        return format_outbound_payload(node.attributes)

B. Praxis-Beispiel: Symmetrischer XRechnung-Codec

Der mitgelieferte XrechnungCodec unterstützt sowohl das Einlesen als auch das Generieren von Rechnungen nach dem EU-Standard EN 16931 in den beiden offiziellen Syntaxen UBL 2.1 (Universal Business Language) und UN/CEFACT CII (Cross Industry Invoice):

# xrechnung_roundtrip_example.py
from modules.knowledge_graph import KnowledgeGraph
from modules.nlp.finance import XrechnungCodec

kg = KnowledgeGraph()

# 1. Decode: Eingehende XRechnung (UBL oder CII XML) in den Graphen parsen
with open("incoming_invoice.xml", "r", encoding="utf-8") as f:
    xml_data = f.read()

result = XrechnungCodec.decode(xml_data, kg)
inv_node_id = result["invoice_id"]
print(f"Rechnung {inv_node_id} mit Betrag {result['payable_amount']} EUR im Graph verankert.")

# 2. Inferenz & Z3-Regelprüfung: Rabattierung / Skonto deterministisch anpassen
node = kg.get_node(inv_node_id)
node.attributes["payable_amount"] = 11400.00  # Skontoabzug

# 3. Encode: Neue, EN 16931-konforme XRechnung (UBL 2.1) synthetisieren
outbound_ubl_xml = XrechnungCodec.encode(kg, entity_id=inv_node_id, syntax="UBL_2.1")

# 4. Lokale Speicherung (Zero Network Egress)
with open("approved_outgoing_invoice.xml", "w", encoding="utf-8") as f:
    f.write(outbound_ubl_xml)
print("XRechnung erfolgreich lokal generiert – 0 MB VRAM, 0 Bytes Egress.")

C. Deklaration im AI Container Manifest (v2.0)

Damit der AI Container Store und der Dependency Resolver die Codecs automatisch erkennen und Schnittstellen auflösen, werden sie im manifest.json deklariert:

{
  "manifest_version": "2.0",
  "id": "vault.finance.invoicing",
  "dependencies": {
    "suite_id": "suite.enterprise.finance",
    "suite_name": "Enterprise Finance Suite",
    "required_vaults": [
      { "vault_id": "vault.core.compliance_z3", "min_version": "1.0.0" }
    ]
  },
  "capabilities": {
    "codecs": [
      {
        "format_id": "invoice.xrechnung.v3",
        "standards": ["EN 16931", "UBL 2.1", "UN/CEFACT CII"],
        "can_decode": true,
        "can_encode": true
      }
    ],
    "invariants": ["tax_subtotals_sum_equals_tax_total", "line_extension_sum_equals_net"]
  }
}