Pipeline Architecture¶

G6's pipeline system orchestrates component execution using declarative step definitions, trio-powered async concurrency, and functional programming bridges.

PipelineStep¶

A PipelineStep is a declarative description of a single operation:

from mvp.core.pipeline import PipelineStep

step = PipelineStep(
    component="ctx_search",     # Component name (from registry)
    operation="search",         # Operation to invoke
    params={"query": "memory leaks", "engine": "duckduckgo"},
)

Field	Type	Purpose
`component`	`str`	Component name as registered in the `ComponentRegistry`
`operation`	`str`	Operation name to invoke on the component
`params`	`dict`	Parameters to pass to the operation

Execution functions¶

Sequential execution¶

from mvp.core.pipeline import run_pipeline, PipelineStep

steps = [
    PipelineStep(component="ctx_search", operation="search", params={"query": "safety"}),
    PipelineStep(component="ctx_rag", operation="retrieve", params={"top_k": 5}),
    PipelineStep(component="formal_methods", operation="verify", params={"strategy": "dpll"}),
]

result = run_pipeline(steps)
# Result.ok([result_a, result_b, result_c]) or Result.fail(error)

run_pipeline executes steps in order and stops on first failure. If all steps succeed, it returns Result.ok with a list of all results.

Parallel execution¶

from mvp.core.pipeline import run_parallel, PipelineStep

steps = [
    PipelineStep(component="ctx_search", operation="search", params={"query": "safety"}),
    PipelineStep(component="ctx_elastic", operation="search", params={"query": "safety"}),
    PipelineStep(component="ctx_rag", operation="retrieve", params={"query": "safety"}),
]

result = run_parallel(steps)
# All three run concurrently via trio nursery

run_parallel uses a trio nursery to execute all steps concurrently. Results are collected after all steps complete.

flowchart TD
    S[Start] --> N[trio nursery]
    N --> A[ctx_search]
    N --> B[ctx_elastic]
    N --> C[ctx_rag]
    A --> R[Collect Results]
    B --> R
    C --> R
    R --> E[Result.ok or Result.fail]

fp_bridge.py -- functional programming bridge¶

The fp_bridge module bridges G6's Result[T] type with the returns library for advanced functional composition.

Core functions¶

Function	Signature	Purpose
`to_core`	`returns.Result -> mvp.core.Result`	Convert `returns` Result to G6 Result
`from_core`	`mvp.core.Result -> returns.Result`	Convert G6 Result to `returns` Result
`safe_invoke`	`(fn, *args) -> Result`	Call a function and catch exceptions into Result
`flow_blocks`	`(blocks, input) -> Result`	Chain multiple AIBlocks with railway semantics

Usage examples¶

from mvp.core.fp_bridge import safe_invoke, flow_blocks

# Safe function invocation (catches exceptions)
result = safe_invoke(risky_function, arg1, arg2)
# Result.ok(return_value) or Result.fail(str(exception))

# Chain multiple blocks
result = flow_blocks([block_a, block_b, block_c], initial_input)
# Equivalent to: block_a >> block_b >> block_c, but with explicit input

ComponentRegistry¶

The ComponentRegistry auto-discovers all components at import time by scanning components/mvp/.

from mvp.core.registry import get_registry

registry = get_registry()

# List all registered components
components = registry.list_all()  # list[ComponentMeta]

# Invoke a component operation
result = registry.invoke("ctx_search", "search", {"query": "example"})
# Result.ok(output) or Result.fail(error)

ComponentMeta¶

Each registered component exposes metadata:

Field	Type	Description
`name`	`str`	Component identifier
`module`	`str`	Python module path
`operations`	`list[str]`	Available operations
`has_mcp`	`bool`	Whether it has an MCP sub-package

Discovery process¶

flowchart LR
    S[Import registry] --> D[Scan components/mvp/]
    D --> F[Find __init__.py files]
    F --> I[Import each module]
    I --> R[Register AIBlock subclasses]
    R --> C[ComponentRegistry ready<br/>270 registry components]

The registry discovers 270 registry components including 39 job agent components and the job framework.

Example pipeline flow¶

A research pipeline that decomposes a goal, retrieves context, and verifies results:

flowchart TD
    subgraph Sequential
        A[goal_engine<br/>decompose] --> B[ctx_search<br/>web search]
        B --> C[ctx_rag<br/>retrieve + ground]
        C --> D[agent_claude<br/>synthesize]
        D --> E[formal_methods<br/>verify]
    end

    subgraph Parallel["Parallel Retrieval (alternative)"]
        P1[ctx_search]
        P2[ctx_elastic]
        P3[ctx_rag]
    end

As code¶

from mvp.core.pipeline import run_pipeline, PipelineStep

research_pipeline = [
    PipelineStep("goal_engine", "decompose", {"goal": "Analyze memory safety"}),
    PipelineStep("ctx_search", "search", {"query": "memory safety patterns"}),
    PipelineStep("ctx_rag", "retrieve", {"top_k": 10}),
    PipelineStep("agent_claude", "infer", {"prompt": "Synthesize findings"}),
    PipelineStep("formal_methods", "verify", {"strategy": "dpll"}),
]

result = run_pipeline(research_pipeline)

Base integration¶

All 14 bases integrate with the pipeline system through three shared helpers:

# Available in every base
_list_components()                    # -> list[ComponentMeta]
_invoke_component(name, op, params)   # -> Result
_run_pipeline(steps)                  # -> Result[list]

This means pipelines work identically whether triggered via MCP, REST, CLI, gRPC, SOAP, or Erlang.

See Component Model for the AIBlock and Result[T] types that pipelines orchestrate.