Pattern Composition

Two primitives for building multi-step agentic pipelines out of simpler patterns.

PatternChain — sequential piping

from brain.patterns import PatternChain, RAGAgent, ReflexionAgent

chain = PatternChain([
    RAGAgent(retriever=my_retriever),          # step 1: retrieve context
    ReflexionAgent(max_iterations=2),          # step 2: self-critique
])
result = chain.run("What is our auth strategy?")
print(result.answer)

# Inspect per-step results
for step in result.metadata["chain_steps"]:
    print(f"[{step['pattern']}] {step['answer'][:80]}")

Custom transform

By default, the previous pattern's answer becomes the next task. Override with transform_fn:

chain = PatternChain(
    [rag_agent, reflexion_agent],
    transform_fn=lambda result, i: f"Improve this answer: {result.answer}",
)

Failure handling

# Stop on first failure (default)
chain = PatternChain([p1, p2], stop_on_failure=True)

# Continue even if a step fails
chain = PatternChain([p1, p2], stop_on_failure=False)

PatternRouter — conditional dispatch

from brain.patterns import PatternRouter, RAGAgent, ReActAgent

router = PatternRouter(
    routes={
        "sql":      RAGAgent(retriever=db_retriever),
        "research": ReActAgent(tools={"search": web_search}),
    },
    route_fn=lambda task: "sql" if "SELECT" in task.upper() else "research",
    default="research",
)

result = router.run("How many events were logged last week?")
print(result.metadata["route"])    # "sql"
print(result.metadata["pattern"]) # "RAGAgent"

API Reference

brain.patterns.compose.PatternChain

PatternChain(patterns: list[BasePattern], transform_fn: TransformFn | None = None, stop_on_failure: bool = True)

Bases: BasePattern

Run patterns sequentially, piping output to input.

Each pattern receives the previous pattern's answer as its task (or the result of a custom transform_fn). The final PatternResult contains the last pattern's answer; full per-step detail is in result.metadata["chain_steps"].

If any step fails and stop_on_failure=True (default), the chain returns immediately with ok=False. If stop_on_failure=False, the raw answer from the failed step is forwarded and the chain continues.

Parameters:

Name	Type	Description	Default
patterns	list[BasePattern]	Ordered list of BasePattern instances.	required
transform_fn	TransformFn | None	Called as transform_fn(result, step_index) -> str to produce the next task. Defaults to passing result.answer through.	None
stop_on_failure	bool	Stop and return if any pattern returns ok=False. Default: True.	True

Source code in brain/patterns/compose.py

def __init__(
    self,
    patterns: list[BasePattern],
    transform_fn: TransformFn | None = None,
    stop_on_failure: bool = True,
) -> None:
    if not patterns:
        raise ValueError("PatternChain requires at least one pattern")
    self.patterns = patterns
    self.transform_fn: TransformFn = transform_fn or _default_transform
    self.stop_on_failure = stop_on_failure

run

run(task: str, **kwargs: Any) -> PatternResult

Execute the chain.

Parameters:

Name	Type	Description	Default
task	str	Initial input passed to the first pattern.	required

Returns:

Type	Description
PatternResult	PatternResult with the final pattern's answer. metadata contains:
PatternResult	chain_steps: list of ChainStepRecord dicts
PatternResult	chain_length: number of patterns in the chain

Source code in brain/patterns/compose.py

def run(self, task: str, **kwargs: Any) -> PatternResult:
    """Execute the chain.

    Args:
        task: Initial input passed to the first pattern.

    Returns:
        PatternResult with the final pattern's answer. ``metadata`` contains:

        - ``chain_steps``: list of ``ChainStepRecord`` dicts
        - ``chain_length``: number of patterns in the chain
    """
    chain_steps: list[ChainStepRecord] = []
    all_steps: list[Step] = []
    current_task = task
    total_iterations = 0

    for i, pattern in enumerate(self.patterns):
        pattern_name = type(pattern).__name__
        logger.debug("PatternChain: step %d/%d — %s", i + 1, len(self.patterns), pattern_name)

        try:
            result = pattern.run(current_task, **kwargs)
        except Exception as exc:  # noqa: BLE001
            logger.warning("PatternChain: step %d (%s) raised: %s", i, pattern_name, exc)
            record = ChainStepRecord(
                step_index=i,
                pattern_name=pattern_name,
                task=current_task,
                answer="",
                iterations=0,
                ok=False,
                error=str(exc),
            )
            chain_steps.append(record)
            if self.stop_on_failure:
                return PatternResult(
                    answer="",
                    steps=all_steps,
                    iterations=total_iterations,
                    ok=False,
                    error=f"Step {i} ({pattern_name}) raised: {exc}",
                    metadata={
                        "chain_steps": [_step_to_dict(s) for s in chain_steps],
                        "chain_length": len(self.patterns),
                    },
                )
            # Continue with empty answer
            current_task = ""
            continue

        record = ChainStepRecord(
            step_index=i,
            pattern_name=pattern_name,
            task=current_task,
            answer=result.answer,
            iterations=result.iterations,
            ok=result.ok,
            error=result.error,
        )
        chain_steps.append(record)

        # Offset step indices so they don't collide across patterns
        for step in result.steps:
            all_steps.append(
                Step(
                    index=total_iterations + step.index,
                    thought=step.thought,
                    action=step.action,
                    action_input=step.action_input,
                    observation=step.observation,
                    text=step.text,
                )
            )
        total_iterations += result.iterations

        if not result.ok and self.stop_on_failure:
            return PatternResult(
                answer=result.answer,
                steps=all_steps,
                iterations=total_iterations,
                ok=False,
                error=f"Step {i} ({pattern_name}) failed: {result.error}",
                metadata={
                    "chain_steps": [_step_to_dict(s) for s in chain_steps],
                    "chain_length": len(self.patterns),
                },
            )

        # Prepare next task (skip transform for last step)
        if i < len(self.patterns) - 1:
            try:
                current_task = self.transform_fn(result, i)
            except Exception as exc:  # noqa: BLE001
                logger.warning("PatternChain: transform_fn raised at step %d: %s", i, exc)
                current_task = result.answer

    final = chain_steps[-1] if chain_steps else None
    return PatternResult(
        answer=final.answer if final else "",
        steps=all_steps,
        iterations=total_iterations,
        ok=all(s.ok for s in chain_steps),
        metadata={
            "chain_steps": [_step_to_dict(s) for s in chain_steps],
            "chain_length": len(self.patterns),
        },
    )

brain.patterns.compose.PatternRouter

PatternRouter(routes: dict[str, BasePattern], route_fn: RouteFn, default: str | None = None)

Bases: BasePattern

Route a task to one of several patterns based on a routing function.

Parameters:

Name	Type	Description	Default
routes	dict[str, BasePattern]	Dict mapping route label → BasePattern.	required
route_fn	RouteFn	Callable (task: str) -> label that returns a key from routes.	required
default	str | None	Fallback label if route_fn returns an unknown key. If not set and the label is unknown, returns an error result.	None

Source code in brain/patterns/compose.py

def __init__(
    self,
    routes: dict[str, BasePattern],
    route_fn: RouteFn,
    default: str | None = None,
) -> None:
    if not routes:
        raise ValueError("PatternRouter requires at least one route")
    self.routes = routes
    self.route_fn = route_fn
    self.default = default

run

run(task: str, **kwargs: Any) -> PatternResult

Route task to the appropriate pattern and run it.

Parameters:

Name	Type	Description	Default
task	str	The question or instruction for the agent.	required

Returns:

Type	Description
PatternResult	PatternResult from the selected pattern. metadata contains:
PatternResult	route: the label returned by route_fn
PatternResult	pattern: class name of the pattern that ran
PatternResult	available_routes: list of all configured route labels

Source code in brain/patterns/compose.py

def run(self, task: str, **kwargs: Any) -> PatternResult:
    """Route task to the appropriate pattern and run it.

    Args:
        task: The question or instruction for the agent.

    Returns:
        PatternResult from the selected pattern. ``metadata`` contains:

        - ``route``: the label returned by ``route_fn``
        - ``pattern``: class name of the pattern that ran
        - ``available_routes``: list of all configured route labels
    """
    try:
        label = self.route_fn(task)
    except Exception as exc:  # noqa: BLE001
        logger.warning("PatternRouter: route_fn raised: %s", exc)
        label = self.default or ""

    pattern = self.routes.get(label)
    if pattern is None:
        if self.default is not None:
            logger.debug(
                "PatternRouter: unknown route %r — falling back to %r", label, self.default
            )
            label = self.default
            pattern = self.routes.get(self.default)

    if pattern is None:
        available = list(self.routes.keys())
        return PatternResult(
            answer="",
            ok=False,
            error=(f"PatternRouter: unknown route {label!r}. Available: {available}"),
            metadata={
                "route": label,
                "pattern": None,
                "available_routes": available,
            },
        )

    logger.debug("PatternRouter: routing %r to %s", label, type(pattern).__name__)
    try:
        result = pattern.run(task, **kwargs)
    except Exception as exc:  # noqa: BLE001
        return PatternResult(
            answer="",
            ok=False,
            error=f"PatternRouter: pattern {type(pattern).__name__} raised: {exc}",
            metadata={
                "route": label,
                "pattern": type(pattern).__name__,
                "available_routes": list(self.routes.keys()),
            },
        )

    # Merge routing metadata into the result
    result.metadata["route"] = label
    result.metadata["pattern"] = type(pattern).__name__
    result.metadata["available_routes"] = list(self.routes.keys())
    return result

brain.patterns.compose.ChainStepRecord dataclass

ChainStepRecord(step_index: int, pattern_name: str, task: str, answer: str, iterations: int, ok: bool, error: str | None = None)

Metadata for one pattern's execution within a PatternChain run.

Composition patterns

Pipeline	When to use
RAG → Reflexion	Retrieve context, then self-critique the answer
PlanExecute → RAG	Plan steps, then retrieve evidence for each
Router → Chain	Route to a multi-step pipeline by query type
Chain → HITL	Run steps, then require human approval of the final action