FP-Patterns für Agent-Systeme – Agent Fundamentals

Komposition, Pipelines und automatische Code-Validierung – wie man mit FP und Kategorientheorie konkrete Agent-Systeme baut.

In den bisherigen drei Artikeln haben wir das Fundament gelegt: Pure Functions und Immutability, Kategorien und Funktoren, Monaden als Protokoll. In diesem abschließenden Artikel setzen wir alles zusammen.

1. Komposition als Agent-Architektur

Kleine Funktionen sind für KI-Agents zuverlässiger generierbar als große. Ein Agent kann strip_whitespace oder to_lowercase mit hoher Sicherheit korrekt erzeugen – eine 100-Zeilen-process_order-Funktion eher nicht.

Die Lösung: Komposition. Kleine Bausteine, die der Agent einzeln generiert und dann verbindet.

"Composition is the essence of all things categorical. The key insight is that the sum of two parts is no more complex than the parts themselves." — Boris Marinov: Category Theory Illustrated, S. 29

Komposition in Python

def compose(f, g):
    """(g ∘ f)(x) = g(f(x))"""
    return lambda x: g(f(x))

def compose_all(*fns):
    """Komponiert beliebig viele Funktionen von links nach rechts."""
    def composed(x):
        result = x
        for fn in fns:
            result = fn(result)
        return result
    return composed

Die kategorischen Gesetze

Assoziativität: Egal wie der Agent die Schritte gruppiert, das Ergebnis ist gleich:

# (h ∘ g) ∘ f == h ∘ (g ∘ f)
assert compose(compose(f, g), h)(x) == compose(f, compose(g, h))(x)

Identität: Es gibt eine „Nichts-tun"-Funktion, die neutral ist:

def identity(x): return x
assert compose(f, identity)(x) == f(x) == compose(identity, f)(x)

Für den Agent bedeutet das: Er kann Schritte frei umgruppieren und einen leeren Schritt einfügen, ohne das Ergebnis zu verändern.

Agent-Bausteine als komponierbare Funktionen

def strip_whitespace(s: str) -> str: return s.strip()
def to_lowercase(s: str) -> str: return s.lower()
def remove_special_chars(s: str) -> str:
    return "".join(c for c in s if c.isalnum() or c.isspace())

def truncate(max_len: int) -> Callable[[str], str]:
    """Higher-Order Function: Konfiguration einmal übergeben."""
    return lambda s: s[:max_len]

# Der Agent komponiert:
normalize = compose_all(strip_whitespace, to_lowercase, remove_special_chars)
sanitize = compose_all(normalize, truncate(100))

Benannte Transformationsketten

@dataclass(frozen=True)
class TransformStep:
    name: str
    transform: Callable

steps = (
    TransformStep("strip", strip_whitespace),
    TransformStep("lower", to_lowercase),
    TransformStep("clean", remove_special_chars),
)

chain = create_transform_chain(steps)  # → komponierte Funktion
describe_chain(steps)  # → ("strip", "lower", "clean")

Die Schritte sind Daten, nicht Code – der Agent kann sie dynamisch zusammenstellen.

→ Vollständiger Code: src/agent_komposition.py

2. Pipelines: Railway Oriented Programming für Agents

Die Komposition aus Abschnitt 1 verbindet reine Funktionen A → B. Aber Agent-Workflows können fehlschlagen. Hier kommt die Result-Monade ins Spiel:

Konfigurierbare Pipelines

@dataclass(frozen=True)
class PipelineStep:
    name: str
    execute: Callable  # → Result

@dataclass(frozen=True)
class PipelineConfig:
    name: str
    steps: tuple[PipelineStep, ...] = ()

    def add_step(self, step): ...   # → neue Config
    def remove_step(self, name): ...  # → neue Config

Die Pipeline ist eine immutable Datenstruktur – der Agent kann Schritte hinzufügen und entfernen, ohne die Original-Config zu verändern.

Pipeline-Ausführung

def run_pipeline(config, initial):
    """Railway-Stil: Beim ersten Err stoppt die Pipeline."""
    result = Ok(initial)
    log, completed = [], 0
    for step in config.steps:
        result = result.bind(step.execute)
        if result.is_ok:
            log.append(f"✓ {step.name}")
            completed += 1
        else:
            log.append(f"✗ {step.name}: {result.error}")
            return PipelineResult(success=False, error=result.error, ...)
    return PipelineResult(success=True, value=result.value, ...)

Vordefinierte Bausteine

# Validierung: Fehlende Felder?
make_validate_step(("name", "email"))

# Transformation: Funktion in Result wrappen
make_transform_step("uppercase", str.upper)

# Anreicherung: Zusätzliche Daten
make_enrich_step({"version": "1.0", "processed": True})

Pipeline-Komposition

Pipelines selbst sind komponierbar:

p1 = create_data_processing_pipeline(required_fields=("name",))
p2 = PipelineConfig("enrichment").add_step(make_enrich_step({"v": "2"}))
combined = chain_pipelines(p1, p2)  # → "data-processing → enrichment"

"Composing a list of functions together gives us a new function that invokes the first function, passes its output to the next, which passes it to the next, and so on until a result is returned." — Stuart Sierra & Luke VanderHart: Functional Patterns, S. 189

→ Vollständiger Code: src/agent_pipeline.py

3. Tool-Registry: Funktionen als Daten

FP-Funktionen haben alles in der Signatur: Name, Parameter, Rückgabetyp. Das macht sie ideal als Agent-Tools – die Signatur A → B ist gleichzeitig Dokumentation, Vertrag und Tool-Definition.

"Understanding and using type signatures is a critical part of coding." — Scott Wlaschin: Domain Modeling Made Functional, S. 72

Aus einer FP-Funktion wie:

def validate_email(email: str) -> str:
    """Validiert eine E-Mail-Adresse und gibt sie normalisiert zurück."""
    ...

kann man automatisch eine Tool-Definition generieren, die direkt im JSON-Schema-Format für OpenAI Function Calling oder andere Agent-Frameworks nutzbar ist. Die Signatur lügt nicht – und der Agent kann sich darauf verlassen.

4. Automatische Code-Validierung

FP-Regeln sind maschinenprüfbar. Mit Python's ast-Modul können wir generierten Code automatisch auf FP-Konformität prüfen:

Die Checks

check_type_hints(source)        # Alle Funktionen haben Return-Type-Hints?
check_frozen_dataclasses(source) # Alle Dataclasses sind frozen=True?
check_no_global_state(source)    # Keine globalen mutierbaren Variablen?
check_no_print_statements(source) # Keine print()-Seiteneffekte?
check_docstrings(source)         # Alle Funktionen haben Docstrings?

Ergebnis als immutable Datenstruktur

@dataclass(frozen=True)
class CodeCheck:
    rule: str
    passed: bool
    message: str

@dataclass(frozen=True)
class CheckReport:
    checks: tuple[CodeCheck, ...]

    @property
    def all_passed(self) -> bool: ...
    @property
    def failed(self) -> tuple[CodeCheck, ...]: ...
    def summary(self) -> str: ...

Gesamt-Check

report = run_all_checks(source_code)
report.summary()  # → "5/5 Checks bestanden"

# Oder selektiv:
report = run_selected_checks(source, (check_type_hints, check_docstrings))

Beispiel: Guter vs. schlechter Code

# ✅ FP-konform
good = '''
@dataclass(frozen=True)
class User:
    name: str

def greet(user: User) -> str:
    """Begrüßt einen User."""
    return f"Hallo {user.name}"
'''
run_all_checks(good).all_passed  # → True

# ❌ Nicht FP-konform
bad = '''
cache = {}
def process(data):
    print("processing...")
    cache[data] = True
'''
run_all_checks(bad).all_passed  # → False
# Fehler: globaler State, fehlende Type Hints, print(), fehlender Docstring

Das ist die Quintessenz: FP-Prinzipien sind nicht nur guter Stil – sie sind maschinenprüfbar. Der Agent kann seinen eigenen generierten Code validieren, bevor er ihn ausführt.

→ Vollständiger Code: src/agent_code_checks.py

5. Der FP-Systemprompt für KI-Agents

Zum Abschluss destillieren wir alles in 7 Regeln, die du direkt als Systemprompt in deinen KI-Workflow einbauen kannst:

## FP-Regeln für Code-Generierung

### Regel 1: Schreibe Pure Functions
Gleicher Input → gleicher Output, keine Seiteneffekte.
Alle Abhängigkeiten als Parameter.

### Regel 2: Nutze immutable Datenstrukturen
@dataclass(frozen=True), tuple statt list.
Neue Objekte statt Mutation.

### Regel 3: Nutze sprechende Type Signatures
Type Hints für alle Parameter und Rückgabewerte.
A → Result[B] dokumentiert: kann fehlschlagen.

### Regel 4: Komponiere kleine Funktionen
Zerlege in unabhängig testbare Bausteine.
Verbinde mit bind, map, Pipelines.

### Regel 5: Verwende Result statt Exceptions
Ok/Err als Rückgabetyp für erwartbare Fehler.
Exceptions nur für Programmfehler.

### Regel 6: Deklarativ statt imperativ
map, filter, Comprehensions, Pattern Matching.
Was berechnet werden soll, nicht wie.

### Regel 7: Higher-Order Functions für Konfiguration
Konfigurierte Funktionen mit Closures.
Reduziert Kopplung, erhöht Wiederverwendbarkeit.

Selbst ausprobieren

Genau Abschnitt 1 in klein, direkt im Browser: kleine Pure-String-Transformationen, ein generisches compose_all, eine benannte Schritt-Kette als frozen-Dataclass und eine truncate(max_len)-Higher-Order-Function. Der Agent kann diese Schritte als Daten zusammenstellen, der Code bleibt eine reine Datentransformation.

from dataclasses import dataclass
from typing import Callable

def strip_whitespace(s: str) -> str: return s.strip()
def to_lowercase(s: str) -> str:     return s.lower()
def remove_special(s: str) -> str:
    return "".join(c for c in s if c.isalnum() or c.isspace())

def truncate(max_len: int) -> Callable[[str], str]:
    """Higher-Order: Konfiguration einmal binden, wiederverwenden."""
    return lambda s: s[:max_len]

def compose_all(*fns):
    def run(x):
        for f in fns:
            x = f(x)
        return x
    return run

@dataclass(frozen=True)
class TransformStep:
    name: str
    transform: Callable

def create_chain(steps):
    """Aus benannten Schritten eine komponierte Funktion bauen."""
    return compose_all(*(s.transform for s in steps))

def describe(steps):
    return tuple(s.name for s in steps)

steps = (
    TransformStep("strip",    strip_whitespace),
    TransformStep("lower",    to_lowercase),
    TransformStep("clean",    remove_special),
    TransformStep("truncate", truncate(20)),
)

chain = create_chain(steps)

for raw in [
    "   Hallo, WELT!!  ",
    "  Functional Programming für KI-Agents – yay!  ",
    "kurz",
]:
    print(f"{raw!r:55} → {chain(raw)!r}")

print("Pipeline-Schritte (als Daten):", describe(steps))

Erweiterungs-Idee: Ergänze einen Schritt prefix(p) (analog zu truncate) und schreibe swap(steps, i, j), das die Reihenfolge zweier Schritte tauscht – die Assoziativität der Komposition garantiert, dass das Ergebnis nur von der neuen Reihenfolge abhängt, nicht von der ursprünglichen Gruppierung.

6. Zusammenfassung der Artikelserie

Was wir aufgebaut haben

Artikel	Konzept	Agent-Nutzen
1 – Warum FP für KI	Pure Functions, Immutability, Signaturen	Vorhersagbare, sichere Bausteine
2 – Kategorien-Basics	Kategorie, Funktor, Natürliche Transformation	Formale Garantien für Transformationen
3 – Monaden als Protokoll	Result, Maybe, Validation	Universelles Steuerungsprotokoll
4 – FP-Patterns	Komposition, Pipelines, Code-Checks	Konkrete Agent-Architektur

Die zentrale Erkenntnis

Verlagere Komplexität in Datenstrukturen, nicht in Kontrollfluss.

Statt if/else-Ketten: Result-Monade. Statt try/except: Railway Oriented Programming. Statt monolithischer Funktionen: Komposition. Statt informeller Konventionen: Kategorische Gesetze.

Ein KI-Agent ist nondeterministisch. Der Code, den er aufruft und generiert, sollte es nicht sein. Functional Programming und Kategorientheorie liefern genau das: vorhersagbare Bausteine mit formalen Garantien, die ein Agent zuverlässig kombinieren kann.

Quellen

Boris Marinov: Category Theory Illustrated, S. 29
Stuart Sierra & Luke VanderHart: Functional Patterns, S. 189
Scott Wlaschin: Domain Modeling Made Functional, S. 72
Robert C. Martin: Clean Code, S. 134
Jay Alammar & Maarten Grootendorst: Common Sense Guide to AI Engineering, S. 34

Code-Beispiele

src/agent_komposition.py – Funktionskomposition, benannte Ketten, numerische Komposition
src/agent_pipeline.py – Konfigurierbare Pipelines mit Result-Monade
src/agent_code_checks.py – AST-basierte Validierung von FP-Regeln