Component Model¶

Every G6 component is built on three primitives: AIBlock for computation, Protocol contracts for capabilities, and Result[T] for error handling.

AIBlock[Input, Output, State]¶

The universal building block. Every component extends this generic dataclass.

from dataclasses import dataclass
from mvp.core import AIBlock, Result

@dataclass
class MyBlock(AIBlock[MyInput, MyOutput, dict]):
    name: str = "my_component"
    state: dict = None

    def __post_init__(self):
        if self.state is None:
            self.state = {}

Methods¶

Method	Signature	Purpose
`infer`	`(input: I) -> Result[O]`	Run inference -- the primary operation
`learn`	`(data: list[I]) -> Result[S]`	Update internal state from examples
`save`	`(path: str) -> Result[str]`	Persist state to disk
`load`	`(path: str) -> Result[str]`	Restore state from disk
`>>`	`(other: AIBlock) -> PipelineBlock`	Compose into a pipeline

Fields¶

Field	Type	Purpose
`name`	`str`	Component identifier used by the registry
`state`	`S`	Mutable internal state (type varies per component)

Result[T] -- railway-oriented error handling¶

Every operation returns a Result instead of raising exceptions. This enables railway-oriented programming where errors short-circuit through the pipeline.

from mvp.core import Result

# Create results
success = Result.ok(42)
failure = Result.fail("something went wrong")

# Check status
success.is_ok()    # True
failure.is_fail()  # True

# Access value
success.value    # 42
failure.error    # "something went wrong"

Chaining operations¶

result = (
    Result.ok(input_data)
    .map(transform)           # Transform value if ok
    .flat_map(validate)       # Chain operation that returns Result
    .or_else(handle_error)    # Handle error, return new Result
)

Method	Signature	Behavior
`.ok(value)`	`T -> Result[T]`	Create a success result
`.fail(error)`	`str -> Result[T]`	Create a failure result
`.map(fn)`	`(T -> U) -> Result[U]`	Transform value if ok, pass through error
`.flat_map(fn)`	`(T -> Result[U]) -> Result[U]`	Chain operations that return Result
`.or_else(fn)`	`(str -> Result[T]) -> Result[T]`	Handle error, attempt recovery
`.is_ok()`	`-> bool`	True if success
`.is_fail()`	`-> bool`	True if failure

Railway pattern¶

flowchart LR
    I[Input] --> A[Block A]
    A -->|ok| B[Block B]
    A -->|fail| E[Error]
    B -->|ok| C[Block C]
    B -->|fail| E
    C -->|ok| O[Output]
    C -->|fail| E

On the ok track, values flow forward through each block. On the fail track, errors short-circuit past remaining blocks. No exceptions are thrown.

PipelineBlock and the >> operator¶

Components compose using the >> operator:

from mvp.core import PipelineBlock

pipeline = block_a >> block_b >> block_c
result = pipeline.process(input_data)
# Result.ok(output) or Result.fail(error)

PipelineBlock uses regular __init__

PipelineBlock uses a regular __init__ method, not dataclass inheritance. This avoids Python's restriction on dataclass inheritance when parent classes have fields with defaults.

PipelineBlock chains the infer methods:

Call block_a.infer(input) -- get Result[A_Output]
If ok, call block_b.infer(a_output) -- get Result[B_Output]
If ok, call block_c.infer(b_output) -- get Result[C_Output]
If any step fails, the error propagates immediately

Protocol contracts¶

Protocols declare what a component can do using Python's structural typing (typing.Protocol):

Protocol	Method	Description
Ingestible	`ingest(raw) -> Result`	Parse raw input into structured form
Emittable	`emit(data) -> Result`	Format structured data for output
Storable	`save(path) -> Result` / `load(path) -> Result`	Persist and restore state
Inferrable	`infer(input) -> Result`	Core inference operation
Learnable	`learn(data) -> Result`	Online learning from examples
InductiveBias	`bias() -> dict`	Declare assumptions and priors

A component satisfies a protocol by implementing the required methods -- no explicit inheritance needed:

# This component satisfies Inferrable and Storable
@dataclass
class MyBlock(AIBlock[str, str, dict]):
    name: str = "my_block"

    def infer(self, input: str) -> Result[str]:  # Inferrable
        return Result.ok(input.upper())

    def save(self, path: str) -> Result[str]:    # Storable
        return Result.ok(path)

    def load(self, path: str) -> Result[str]:    # Storable
        return Result.ok(path)

How components compose¶

A typical G6 workflow composes multiple components:

flowchart LR
    GE[goal_engine<br/>Decompose] --> CS[ctx_search<br/>Retrieve]
    CS --> RAG[ctx_rag<br/>Ground]
    RAG --> AC[agent_claude<br/>Synthesize]
    AC --> FM[formal_methods<br/>Verify]
    FM --> AE[align_evals<br/>Evaluate]

Each arrow is a Result[T] -- if any step fails, the pipeline short-circuits.

Component conventions¶

All G6 components follow these conventions:

Schema file (schema.py) -- Pydantic models for Input/Output
Block file (block.py) -- AIBlock subclass with infer method
Init file (__init__.py) -- public exports via __all__
Optional MCP sub-package -- <name>_mcp/ directory with tool definitions
Optional skill directory -- skill/ with plugin.json, agents, commands

See Pipelines for the execution engine that orchestrates component composition.