Home › Fp For Ai Agents › Artikel 3
Teil 3 AI-Agenten

Monaden als Agent-Protokoll

Bild: KI-generiert

Monaden als Agent-Protokoll

Warum Result, Maybe und Validation das universelle Steuerungsprotokoll für KI-Agent-Workflows sind.

Im vorherigen Artikel haben wir Funktoren kennengelernt: Sie transformieren Werte innerhalb eines Kontexts. Aber was, wenn die Transformation selbst einen neuen Kontext erzeugt?

def parse_int(s: str) -> Maybe:
    ...  # Gibt Just(42) oder Nothing() zurück

Just("42").map(parse_int)  # → Just(Just(42)) – verschachtelt!

Das Problem: map verpackt das Ergebnis nochmal. Wir brauchen eine Operation, die den Kontext flachklopft: bind (auch flatMap oder >>= genannt). Und genau das macht eine Monade.

1. Q: Was ist eine Monade?

A: Ein Monoid in der Kategorie der Endofunktoren.

Diese Aussage sollte man zum späteren Gebrauch im Hinterkopf behalten, aber jetzt erstmal weiter mit der praktischen Definition:

Eine Monade ist ein Funktor mit einer zusätzlichen Operation, genannt bind oder flatMap:

Dazu drei Gesetze:

  1. Linke Identität: return(a).bind(f) == f(a)
  2. Rechte Identität: m.bind(return) == m
  3. Assoziativität: m.bind(f).bind(g) == m.bind(λx. f(x).bind(g))

Das klingt sehr abstrakt, aber die Intuition ist einfach: Eine Monade ist ein Wert im Kontext mit der Fähigkeit, Operationen zu verketten, wobei der Kontext automatisch verwaltet wird.

"Monads let us compose functions that have effects. The key insight is that the plumbing is hidden — you only see the happy path." — Bartosz Milewski: Category Theory for Programmers, S. 317

"A monad is just a monoid in the category of endofunctors." — Saunders Mac Lane: Categories for the Working Mathematician

2. Result als Agent-Protokoll

Die Result-Monade ist das zentrale Protokoll für Agent-Workflows:

@dataclass(frozen=True)
class Ok:
    value: any
    def map(self, f):  return Ok(f(self.value))
    def bind(self, f): return f(self.value)

@dataclass(frozen=True)
class Err:
    error: str
    def map(self, f):  return self  # Fehler durchreichen
    def bind(self, f): return self  # Fehler durchreichen

Railway Oriented Programming

Das Geniale: Bei Err wird jeder weitere Schritt übersprungen. Das ist Railway Oriented Programming – der Fehler fließt automatisch durch:

def validate_input(data: dict) -> Result:
    if "action" not in data:
        return Err("Feld 'action' fehlt")
    return Ok(data)

def normalize_data(data: dict) -> Result:
    return Ok({k.lower().strip(): v for k, v in data.items()})

def enrich_with_metadata(data: dict) -> Result:
    return Ok({**data, "processed": True})

# Pipeline: validate → normalize → enrich
pipeline = pipeline(validate_input, normalize_data, enrich_with_metadata)

pipeline({"action": "search"})  # → Ok({...})
pipeline({"no_action": True})   # → Err("Feld 'action' fehlt")

Für den KI-Agent ist das ideal: Er muss sich nicht um Fehlerbehandlung kümmern. Der erste Fehler stoppt die Pipeline, und er bekommt eine klare Fehlermeldung.

Monaden-Gesetze verifizieren

# Linke Identität: Ok(5).bind(f) == f(5)
f = lambda x: Ok(x * 2)
assert Ok(5).bind(f) == f(5)  # Ok(10) == Ok(10) ✓

# Rechte Identität: m.bind(Ok) == m
assert Ok(5).bind(Ok) == Ok(5)  # ✓
assert Err("e").bind(Ok) == Err("e")  # ✓

# Assoziativität: m.bind(f).bind(g) == m.bind(λx. f(x).bind(g))
g = lambda x: Ok(x + 1)
assert Ok(5).bind(f).bind(g) == Ok(5).bind(lambda x: f(x).bind(g))  # ✓

Agent-Workflow-Orchestrierung

def run_workflow(steps, initial):
    """Führt benannte Schritte aus – stoppt beim ersten Fehler."""
    result = Ok(initial)
    log = []
    for name, fn in steps:
        result = result.bind(fn)
        if result.is_ok:
            log.append(f"✓ {name}")
        else:
            log.append(f"✗ {name}: {result.error}")
            break
    return WorkflowResult(steps_completed=..., final_result=result, log=tuple(log))

→ Vollständiger Code: src/agent_result.py

3. Maybe für optionale Daten

Die Maybe-Monade modelliert optionale Werte – ohne None/Null-Chaos:

@dataclass(frozen=True)
class Just:
    value: any
    def bind(self, f): return f(self.value)
    def or_else(self, default): return self.value

@dataclass(frozen=True)
class Nothing:
    def bind(self, f): return self
    def or_else(self, default): return default

Agent-Use-Cases

Sicherer Dictionary-Zugriff:

def from_dict(d: dict, key: str) -> Maybe:
    if key in d:
        return Just(d[key])
    return Nothing()

# Kein KeyError, kein try/except
from_dict(config, "model").or_else("gpt-4")

Sicheres Parsing:

def safe_parse_int(s: str) -> Maybe:
    try:
        return Just(int(s))
    except (ValueError, TypeError):
        return Nothing()

safe_parse_int("42")   # → Just(42)
safe_parse_int("abc")  # → Nothing()

Fallback-Ketten:

def chain_lookups(*lookups):
    """Gibt das erste Just zurück."""
    for lookup in lookups:
        result = lookup()
        if isinstance(result, Just):
            return result
    return Nothing()

# Agent sucht: erst Cache, dann DB, dann Default
chain_lookups(
    lambda: from_dict(cache, "key"),
    lambda: from_dict(database, "key"),
    lambda: Just("default"),
)

Immutable Konfiguration:

@dataclass(frozen=True)
class AgentConfig:
    settings: tuple[tuple[str, str], ...] = ()

    def get(self, key: str) -> Maybe:
        for k, v in self.settings:
            if k == key: return Just(v)
        return Nothing()

    def with_setting(self, key: str, value: str) -> AgentConfig:
        # Gibt neue Config zurück – Original bleibt unverändert
        ...

→ Vollständiger Code: src/agent_maybe.py

4. Validation für parallele Fehlersammlung

Result stoppt beim ersten Fehler. Aber manchmal will man alle Fehler auf einmal:

@dataclass(frozen=True)
class Valid:
    value: any

@dataclass(frozen=True)
class Invalid:
    errors: tuple[str, ...]

Applicative Kombination

def combine(*validations):
    """Alle Valid → Valid(tuple), mindestens ein Invalid → Invalid(alle Fehler)."""
    errors, values = [], []
    for v in validations:
        match v:
            case Valid(value): values.append(value)
            case Invalid(errs): errors.extend(errs)
    if errors:
        return Invalid(tuple(errors))
    return Valid(tuple(values))

Agent-Use-Case: Tool-Eingabe validieren

def validate_tool_input(data: dict) -> Validation:
    action_v = validate_not_empty("action", data.get("action", ""))
    target_v = validate_not_empty("target", data.get("target", ""))
    return combine_with(
        lambda a, t: ToolInput(action=a, target=t),
        action_v, target_v,
    )

# Alle Fehler auf einmal:
validate_tool_input({})
# → Invalid(("action darf nicht leer sein", "target darf nicht leer sein"))

Für den KI-Agent ist das besonders wertvoll: Statt nach jedem Fehler einzeln zu korrigieren, bekommt er alle Probleme auf einmal und kann sie in einem Schritt beheben.

→ Vollständiger Code: src/agent_validation.py

5. Monaden als universelles Agent-Interface

Alle drei Monaden folgen demselben Interface:

Monade map bind Kontext
Result Transformiert Ok-Wert Verkettet Schritte, Err stoppt Erfolg/Fehler
Maybe Transformiert Just-Wert Verkettet Lookups, Nothing stoppt Vorhanden/Nicht vorhanden
Validation Transformiert Valid-Wert — (Applicative statt Monade) Gültig/Alle Fehler

Der KI-Agent muss nur ein Pattern kennen:

step1(input).bind(step2).bind(step3)

Und drei Entscheidungen treffen:

  1. Result → wenn Fehler den Workflow stoppen sollen
  2. Maybe → wenn fehlende Werte ok sind (mit Fallback)
  3. Validation → wenn alle Fehler gesammelt werden sollen

Die zentrale Erkenntnis

Monade = Protokoll für „was passiert bei Erfolg/Fehler".

Die Semantik steckt in der Datenstruktur, nicht im Kontrollfluss. Der Agent braucht keine if/else-Ketten, keine try/except-Blöcke – die Monade regelt das automatisch.

Selbst ausprobieren

Genau das Railway-Oriented-Programming-Muster aus Abschnitt 2 lässt sich in wenigen Zeilen direkt ausprobieren: Ok/Err mit map/bind, drei winzige Pipeline-Schritte, und eine generische pipeline-Funktion, die sie verkettet. Beachte: Im Pipeline-Code gibt es keinen einzigen if/else-Fehlerzweig – Err reicht sich selbst durch.

from dataclasses import dataclass

@dataclass(frozen=True)
class Ok:
    value: object
    def map(self, f):  return Ok(f(self.value))
    def bind(self, f): return f(self.value)

@dataclass(frozen=True)
class Err:
    error: str
    def map(self, f):  return self
    def bind(self, f): return self

def validate_input(data):
    if "action" not in data:
        return Err("Feld 'action' fehlt")
    return Ok(data)

def normalize_data(data):
    return Ok({k.lower().strip(): v for k, v in data.items()})

def enrich_with_metadata(data):
    return Ok({**data, "processed": True})

def pipeline(*steps):
    """Verkettet Schritte vom Typ A -> Result[B]."""
    def run(initial):
        result = Ok(initial)
        for step in steps:
            result = result.bind(step)
        return result
    return run

run = pipeline(validate_input, normalize_data, enrich_with_metadata)

for case in [
    {"action": "Search", "Query": "Funktor"},
    {"no_action": True},
    {"action": "SUMMARIZE", "TARGET": "doc.pdf"},
]:
    print(f"{case!r:55} → {run(case)}")

Erweiterungs-Idee: Ergänze einen Schritt log_result(name), der via map einen Eintrag in processed_log schreibt – map bleibt rein, weil der ganze State immutabel weitergereicht wird, und die Pipeline ändert sich nicht.

Ausblick

Jetzt haben wir die Bausteine: Pure Functions, Immutability, Funktoren, Monaden. Im nächsten Artikel setzen wir alles zusammen: Komposition, konfigurierbare Pipelines, Tool-Registry und automatische Code-Validierung.

Quellen

Code-Beispiele

📄 Als PDF