Home › Fp For Ai Agents › Article 4
Part 4 AI Agents

FP Patterns for Agent Systems

Image: AI-generated

FP Patterns for Agent Systems

Composition, pipelines, and automatic code validation — how to build concrete agent systems with FP and category theory.

In the previous three articles, we laid the foundation: Pure functions and immutability, categories and functors, monads as protocol. In this concluding article, we put it all together.

1. Composition as Agent Architecture

Small functions are more reliably generated by AI agents than large ones. An agent can produce strip_whitespace or to_lowercase with high confidence — a 100-line process_order function, not so much.

The solution: Composition. Small building blocks that the agent generates individually and then connects.

"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, p. 29

Composition in Python

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

def compose_all(*fns):
    """Composes any number of functions from left to right."""
    def composed(x):
        result = x
        for fn in fns:
            result = fn(result)
        return result
    return composed

The Categorical Laws

Associativity: No matter how the agent groups the steps, the result is the same:

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

Identity: There is a "do nothing" function that is neutral:

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

For the agent, this means: It can freely regroup steps and insert an empty step without changing the result.

Agent Building Blocks as Composable Functions

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: Pass configuration once."""
    return lambda s: s[:max_len]

# The agent composes:
normalize = compose_all(strip_whitespace, to_lowercase, remove_special_chars)
sanitize = compose_all(normalize, truncate(100))

Named Transformation Chains

@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)  # → composed function
describe_chain(steps)  # → ("strip", "lower", "clean")

The steps are data, not code — the agent can assemble them dynamically.

→ Full code: src/agent_komposition.py

2. Pipelines: Railway Oriented Programming for Agents

The composition from section 1 connects pure functions A → B. But agent workflows can fail. This is where the Result monad comes in:

Configurable 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): ...   # → new config
    def remove_step(self, name): ...  # → new config

The pipeline is an immutable data structure — the agent can add and remove steps without modifying the original config.

Pipeline Execution

def run_pipeline(config, initial):
    """Railway style: Pipeline stops at first Err."""
    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, ...)

Pre-built Building Blocks

# Validation: Missing fields?
make_validate_step(("name", "email"))

# Transformation: Wrap function in Result
make_transform_step("uppercase", str.upper)

# Enrichment: Additional data
make_enrich_step({"version": "1.0", "processed": True})

Pipeline Composition

Pipelines themselves are composable:

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, p. 189

→ Full code: src/agent_pipeline.py

3. Tool Registry: Functions as Data

FP functions carry everything in the signature: name, parameters, return type. This makes them ideal as agent tools — the signature A → B serves simultaneously as documentation, contract, and tool definition.

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

From an FP function like:

def validate_email(email: str) -> str:
    """Validates an email address and returns it normalized."""
    ...

you can automatically generate a tool definition that is directly usable in JSON Schema format for OpenAI Function Calling or other agent frameworks. The signature doesn't lie — and the agent can rely on it.

4. Automatic Code Validation

FP rules are machine-checkable. Using Python's ast module, we can automatically check generated code for FP conformity:

The Checks

check_type_hints(source)        # All functions have return type hints?
check_frozen_dataclasses(source) # All dataclasses are frozen=True?
check_no_global_state(source)    # No global mutable variables?
check_no_print_statements(source) # No print() side effects?
check_docstrings(source)         # All functions have docstrings?

Results as Immutable Data Structure

@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: ...

Complete Check

report = run_all_checks(source_code)
report.summary()  # → "5/5 checks passed"

# Or selectively:
report = run_selected_checks(source, (check_type_hints, check_docstrings))

Example: Good vs. Bad Code

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

def greet(user: User) -> str:
    """Greets a user."""
    return f"Hello {user.name}"
'''
run_all_checks(good).all_passed  # → True

# ❌ Not FP-conformant
bad = '''
cache = {}
def process(data):
    print("processing...")
    cache[data] = True
'''
run_all_checks(bad).all_passed  # → False
# Errors: global state, missing type hints, print(), missing docstring

This is the quintessence: FP principles are not just good style — they are machine-checkable. The agent can validate its own generated code before executing it.

→ Full code: src/agent_code_checks.py

5. The FP System Prompt for AI Agents

To conclude, we distill everything into 7 rules that you can use directly as a system prompt in your AI workflow:

## FP Rules for Code Generation

### Rule 1: Write Pure Functions
Same input → same output, no side effects.
All dependencies as parameters.

### Rule 2: Use Immutable Data Structures
@dataclass(frozen=True), tuple instead of list.
New objects instead of mutation.

### Rule 3: Use Descriptive Type Signatures
Type hints for all parameters and return values.
A → Result[B] documents: can fail.

### Rule 4: Compose Small Functions
Decompose into independently testable building blocks.
Connect with bind, map, pipelines.

### Rule 5: Use Result Instead of Exceptions
Ok/Err as return type for expected errors.
Exceptions only for programming errors.

### Rule 6: Declarative Over Imperative
map, filter, comprehensions, pattern matching.
What should be computed, not how.

### Rule 7: Higher-Order Functions for Configuration
Configured functions with closures.
Reduces coupling, increases reusability.

Try It Yourself

Section 1 in miniature, right in the browser: tiny pure string transforms, a generic compose_all, a named step chain as a frozen dataclass, and a truncate(max_len) higher-order function. The agent can assemble these steps as data; the code stays a pure data transformation.

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: bind configuration once, reuse the function."""
    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):
    """Build a composed function from named steps."""
    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 [
    "   Hello, WORLD!!  ",
    "  Functional Programming for AI agents – yay!  ",
    "short",
]:
    print(f"{raw!r:55} → {chain(raw)!r}")

print("Pipeline steps (as data):", describe(steps))

Extension idea: add a prefix(p) step (analogous to truncate) and write swap(steps, i, j) that swaps two steps – the associativity of composition guarantees that the result depends only on the new order, not on the original grouping.

6. Article Series Summary

What We Built

Article Concept Agent Benefit
1 – Why FP for AI Pure Functions, Immutability, Signatures Predictable, safe building blocks
2 – Category Basics Category, Functor, Natural Transformation Formal guarantees for transformations
3 – Monads as Protocol Result, Maybe, Validation Universal control protocol
4 – FP Patterns Composition, Pipelines, Code Checks Concrete agent architecture

The Central Insight

Shift complexity into data structures, not control flow.

Instead of if/else chains: Result monad. Instead of try/except: Railway Oriented Programming. Instead of monolithic functions: Composition. Instead of informal conventions: Categorical laws.

An AI agent is nondeterministic. The code it calls and generates should not be. Functional Programming and category theory deliver exactly that: predictable building blocks with formal guarantees that an agent can reliably compose.

Sources

Code Examples

📄 As PDF