Orchestration API

orchestration

Orchestration primitives for composing agents.

Provides Pipeline (sequential), Parallel (concurrent), Router (conditional), and RefinementLoop (iterative) — plus the Step protocol for defining executable units.

Evaluator module-attribute

Evaluator = Callable[..., Any] | BaseAgent

An evaluator can be a sync/async callable or a BaseAgent.

StepLike module-attribute

StepLike = Step | BaseAgent | Callable[..., Any]

Union type accepted by Pipeline — auto-coerced to Step.

EvalResult dataclass

EvalResult(
    passed: bool,
    feedback: str = "",
    score: float | None = None,
)

Outcome of evaluating a single iteration's output.

IterationRecord dataclass

IterationRecord(
    iteration: int,
    eval_result: EvalResult,
    duration_ms: float,
)

Record of a single refinement iteration.

LoopResult dataclass

LoopResult(
    output: OutputT,
    iterations: int,
    exit_reason: str,
    history: list[IterationRecord] = list(),
    best_iteration: int | None = None,
    best_score: float | None = None,
)

Final result of a RefinementLoop run.

output is the iteration that the loop is returning to the caller. With keep_best=True (the default), this may be an earlier iteration's output than the final one — specifically the iteration that achieved the highest scored :class:EvalResult. best_iteration and best_score record which iteration was kept and at what score; both are None when no iteration produced a numeric score.

RefinementLoop

RefinementLoop(
    step: StepLike,
    evaluator: Evaluator,
    *,
    state_updater: Callable[..., Any],
    max_iterations: int = 5,
    timeout: float | None = None,
    convergence_window: int | None = None,
    convergence_threshold: float = 0.01,
    confidence_threshold: float | None = None,
    agent_self_eval: BaseAgent | None = None,
    keep_best: bool = True,
)

Iterate a step with evaluation feedback until convergence or limits.

The loop runs the step, evaluates the output, and uses the state_updater to produce the next input. It terminates when the evaluator passes, max iterations are reached, the timeout expires, or scores converge.

Configure the refinement loop.

Parameters:

Name	Type	Description	Default
step	StepLike	The step to iterate.	required
evaluator	Evaluator	Callable or BaseAgent that produces EvalResult.	required
state_updater	Callable[..., Any]	(current_input, output, eval_result) -> next_input.	required
max_iterations	int	Must be >= 1.	5
timeout	float | None	Optional wall-clock timeout in seconds.	None
convergence_window	int | None	Number of recent scores to check for convergence.	None
convergence_threshold	float	Max score variance within the window.	0.01
confidence_threshold	float | None	When set AND EvalResult.score reaches this value, the loop exits with exit_reason="confident". Pairs naturally with agent_self_eval.	None
agent_self_eval	BaseAgent | None	Optional :class:BaseAgent whose run_enriched is used to produce the per-iteration score (the agent's own confidence). When set, confidence_threshold is required. Takes precedence over evaluator for scoring: evaluator is still a required positional argument (so callers explicitly acknowledge the loop has a "fail by default" path), but it is not invoked while agent_self_eval is active. A synthetic EvalResult(passed=False, score=enriched.confidence) is produced each iteration, so the loop only exits early via confidence_threshold — never via the implicit passed=True short-circuit.	None
keep_best	bool	When True (the default), track the iteration with the highest :attr:EvalResult.score and return THAT iteration's output if the final iteration regressed. Protects self-improving loops from "fixer breaks code that already worked" failure modes on imperfect models. The passed=True early exit always wins over keep-best (a passing iteration is the explicit success bar). When the evaluator never returns a numeric score, keep-best is a no-op and the legacy "return latest output" behavior holds. Set keep_best=False to restore strict last-iteration semantics.	True

Source code in orqest/orchestration/loop.py

def __init__(
    self,
    step: StepLike,
    evaluator: Evaluator,
    *,
    state_updater: Callable[..., Any],
    max_iterations: int = 5,
    timeout: float | None = None,
    convergence_window: int | None = None,
    convergence_threshold: float = 0.01,
    confidence_threshold: float | None = None,
    agent_self_eval: BaseAgent | None = None,
    keep_best: bool = True,
) -> None:
    """Configure the refinement loop.

    Args:
        step: The step to iterate.
        evaluator: Callable or BaseAgent that produces EvalResult.
        state_updater: (current_input, output, eval_result) -> next_input.
        max_iterations: Must be >= 1.
        timeout: Optional wall-clock timeout in seconds.
        convergence_window: Number of recent scores to check for convergence.
        convergence_threshold: Max score variance within the window.
        confidence_threshold: When set AND ``EvalResult.score`` reaches
            this value, the loop exits with ``exit_reason="confident"``.
            Pairs naturally with ``agent_self_eval``.
        agent_self_eval: Optional :class:`BaseAgent` whose
            ``run_enriched`` is used to produce the per-iteration
            score (the agent's *own* confidence). When set,
            ``confidence_threshold`` is required. Takes precedence
            over ``evaluator`` for scoring: ``evaluator`` is still a
            required positional argument (so callers explicitly
            acknowledge the loop has a "fail by default" path), but
            it is *not invoked* while ``agent_self_eval`` is active.
            A synthetic ``EvalResult(passed=False,
            score=enriched.confidence)`` is produced each iteration,
            so the loop only exits early via ``confidence_threshold``
            — never via the implicit ``passed=True`` short-circuit.
        keep_best: When ``True`` (the default), track the iteration with
            the highest :attr:`EvalResult.score` and return THAT
            iteration's output if the final iteration regressed. Protects
            self-improving loops from "fixer breaks code that already
            worked" failure modes on imperfect models. The ``passed=True``
            early exit always wins over keep-best (a passing iteration is
            the explicit success bar). When the evaluator never returns a
            numeric ``score``, keep-best is a no-op and the legacy
            "return latest output" behavior holds. Set ``keep_best=False``
            to restore strict last-iteration semantics.

    """
    if max_iterations < 1:
        raise ValueError("max_iterations must be >= 1")
    if agent_self_eval is not None and confidence_threshold is None:
        raise ValueError(
            "agent_self_eval requires confidence_threshold to be set; "
            "self-eval scores are meaningless without a passing bar."
        )
    if (
        agent_self_eval is not None
        and getattr(agent_self_eval, "confidence_protocol", None) is None
    ):
        raise ValueError(
            "agent_self_eval requires the agent to be constructed with a "
            "confidence_protocol; without one run_enriched yields "
            "confidence=None and the loop can never exit via 'confident'."
        )

    self._step = _coerce_step(step)
    self._evaluator = evaluator
    self._state_updater = state_updater
    self._max_iterations = max_iterations
    self._timeout = timeout
    self._convergence_window = convergence_window
    self._convergence_threshold = convergence_threshold
    self._confidence_threshold = confidence_threshold
    self._agent_self_eval = agent_self_eval
    self._keep_best = keep_best

run async

run(initial_input: StateT) -> LoopResult[OutputT]

Execute the refinement loop starting from initial_input.

With keep_best=True (the default), tracks the highest-scoring iteration and returns its output on any non-passed exit if the final iteration regressed. The passed=True early exit always returns the passing iteration's output regardless of score.

Source code in orqest/orchestration/loop.py

async def run(self, initial_input: StateT) -> LoopResult[OutputT]:
    """Execute the refinement loop starting from *initial_input*.

    With ``keep_best=True`` (the default), tracks the highest-scoring
    iteration and returns *its* output on any non-``passed`` exit if the
    final iteration regressed. The ``passed=True`` early exit always
    returns the passing iteration's output regardless of score.
    """
    current_input: Any = initial_input
    history: list[IterationRecord] = []
    scores: list[float] = []
    output: Any = None
    start = time.monotonic()

    # Best-so-far tracking (only meaningful when keep_best=True and the
    # evaluator returns numeric scores). Always populated when a score
    # is seen so LoopResult.best_iteration/best_score are accurate.
    best_output: Any = None
    best_score: float | None = None
    best_iteration: int | None = None
    last_eval_score: float | None = None

    for i in range(1, self._max_iterations + 1):
        if self._timeout is not None:
            elapsed = time.monotonic() - start
            if elapsed >= self._timeout:
                chosen, chosen_iter = self._resolve_kept_output(
                    latest_output=output,
                    latest_score=last_eval_score,
                    latest_iter=i - 1,
                    best_output=best_output,
                    best_score=best_score,
                    best_iteration=best_iteration,
                )
                return LoopResult(
                    output=chosen,
                    iterations=i - 1,
                    exit_reason="timeout",
                    history=history,
                    best_iteration=best_iteration,
                    best_score=best_score,
                )

        iter_start = time.monotonic()
        output = await self._step.execute(current_input)
        eval_result = await self._call_evaluator(output)
        iter_ms = (time.monotonic() - iter_start) * 1000

        record = IterationRecord(
            iteration=i,
            eval_result=eval_result,
            duration_ms=iter_ms,
        )
        history.append(record)
        last_eval_score = eval_result.score

        # Track best-so-far on STRICT improvement (ties keep the earlier
        # one in best_score memory; the resolver still prefers the latest
        # on tied final score so we don't gratuitously walk back).
        if eval_result.score is not None and (
            best_score is None or eval_result.score > best_score
        ):
            best_output = output
            best_score = eval_result.score
            best_iteration = i

        if eval_result.passed:
            # passed is the explicit success bar — keep_best does NOT
            # override it. Also clears best_iteration/score to reflect
            # that the returned output is the passing iteration's.
            return LoopResult(
                output=output,
                iterations=i,
                exit_reason="passed",
                history=history,
                best_iteration=i,
                best_score=eval_result.score,
            )

        if eval_result.score is not None:
            scores.append(eval_result.score)

        # Confidence-driven exit: agent self-rated above the bar.
        if (
            self._confidence_threshold is not None
            and eval_result.score is not None
            and eval_result.score >= self._confidence_threshold
        ):
            return LoopResult(
                output=output,
                iterations=i,
                exit_reason="confident",
                history=history,
                best_iteration=i,
                best_score=eval_result.score,
            )

        if self._check_convergence(scores):
            chosen, chosen_iter = self._resolve_kept_output(
                latest_output=output,
                latest_score=eval_result.score,
                latest_iter=i,
                best_output=best_output,
                best_score=best_score,
                best_iteration=best_iteration,
            )
            return LoopResult(
                output=chosen,
                iterations=i,
                exit_reason="converged",
                history=history,
                best_iteration=best_iteration,
                best_score=best_score,
            )

        current_input = self._state_updater(current_input, output, eval_result)

    chosen, chosen_iter = self._resolve_kept_output(
        latest_output=output,
        latest_score=last_eval_score,
        latest_iter=self._max_iterations,
        best_output=best_output,
        best_score=best_score,
        best_iteration=best_iteration,
    )
    return LoopResult(
        output=chosen,
        iterations=self._max_iterations,
        exit_reason="max_iterations",
        history=history,
        best_iteration=best_iteration,
        best_score=best_score,
    )

MergeStrategy

Built-in merge strategies for combining parallel outputs.

collect_all staticmethod

collect_all(results: list[Any]) -> list[Any]

Return all successful results as a list (default strategy).

Source code in orqest/orchestration/parallel.py

@staticmethod
def collect_all(results: list[Any]) -> list[Any]:
    """Return all successful results as a list (default strategy)."""
    return list(results)

first_wins staticmethod

first_wins(results: list[Any]) -> Any

Return the first successful result, or None if empty.

Source code in orqest/orchestration/parallel.py

@staticmethod
def first_wins(results: list[Any]) -> Any:
    """Return the first successful result, or None if empty."""
    return results[0] if results else None

Parallel

Parallel(
    steps: list[Step | Any],
    *,
    merge: Callable[
        [list[Any]], Any
    ] = MergeStrategy.collect_all,
    timeout: float | None = None,
    name: str = "parallel",
)

Run multiple steps concurrently and merge their results.

Each step is launched as an independent asyncio task. Timed-out tasks are cancelled and recorded as TimeoutError. The merge strategy receives only the successful (non-None) outputs.

Initialize with steps, merge strategy, and optional timeout.

Parameters:

Name	Type	Description	Default
steps	list[Step | Any]	StepLike objects to execute concurrently.	required
merge	Callable[[list[Any]], Any]	Callable that combines successful outputs into a final value.	collect_all
timeout	float | None	Maximum seconds to wait for all steps. None means no limit.	None
name	str	Identifier for logging and events.	'parallel'

Raises:

Type	Description
ValueError	If steps is empty.

Source code in orqest/orchestration/parallel.py

def __init__(
    self,
    steps: list[Step | Any],
    *,
    merge: Callable[[list[Any]], Any] = MergeStrategy.collect_all,
    timeout: float | None = None,
    name: str = "parallel",
) -> None:
    """Initialize with steps, merge strategy, and optional timeout.

    Args:
        steps: StepLike objects to execute concurrently.
        merge: Callable that combines successful outputs into a final value.
        timeout: Maximum seconds to wait for all steps. None means no limit.
        name: Identifier for logging and events.

    Raises:
        ValueError: If steps is empty.

    """
    if not steps:
        raise ValueError("Parallel requires at least one step.")
    self._steps: list[Step] = [_coerce_step(s) for s in steps]
    self._merge = merge
    self._timeout = timeout
    self._name = name

run async

run(input_data: Any) -> ParallelResult[OutputT]

Execute all steps concurrently and return aggregated results.

Source code in orqest/orchestration/parallel.py

async def run(self, input_data: Any) -> ParallelResult[OutputT]:
    """Execute all steps concurrently and return aggregated results."""
    tasks = [
        asyncio.create_task(step.execute(input_data)) for step in self._steps
    ]

    outputs: list[OutputT | None] = [None] * len(tasks)
    errors: list[Exception | None] = [None] * len(tasks)

    done, pending = await asyncio.wait(
        tasks,
        timeout=self._timeout,
        return_when=asyncio.ALL_COMPLETED,
    )

    # Cancel and await pending (timed-out) tasks.
    for task in pending:
        task.cancel()
    if pending:
        await asyncio.wait(pending)

    for i, task in enumerate(tasks):
        if task in pending:
            errors[i] = TimeoutError(
                f"Step '{self._steps[i].step_name}' timed out"
            )
        elif task.exception() is not None:
            errors[i] = task.exception()
        else:
            outputs[i] = task.result()

    successful = [o for o in outputs if o is not None]
    merged = self._merge(successful)

    return ParallelResult(outputs=outputs, errors=errors, merged=merged)

ParallelResult dataclass

ParallelResult(
    outputs: list[OutputT | None],
    errors: list[Exception | None],
    merged: Any,
)

Outcome of a parallel execution.

Attributes:

Name	Type	Description
outputs	list[OutputT | None]	Per-step results; None for steps that failed or timed out.
errors	list[Exception | None]	Per-step exceptions; None for steps that succeeded.
merged	Any	Result of applying the merge strategy to successful outputs.

Pipeline

Pipeline(
    steps: list[StepLike | tuple[StepLike, StepConfig]],
    *,
    name: str = "pipeline",
)

Run a sequence of steps, feeding each output into the next input.

Each entry in steps is either a StepLike value or a (StepLike, StepConfig) tuple. StepLike values are auto-coerced via _coerce_step.

Validate and coerce the step list.

Raises ValueError if steps is empty.

Source code in orqest/orchestration/pipeline.py

def __init__(
    self,
    steps: list[StepLike | tuple[StepLike, StepConfig]],
    *,
    name: str = "pipeline",
) -> None:
    """Validate and coerce the step list.

    Raises ValueError if *steps* is empty.
    """
    if not steps:
        raise ValueError("Pipeline requires at least one step")

    self.name = name
    self._steps: list[tuple[Step, StepConfig]] = []
    for entry in steps:
        if isinstance(entry, tuple):
            raw_step, config = entry
            step = _coerce_step(raw_step)
            self._steps.append((step, config))
        else:
            step = _coerce_step(entry)
            config = StepConfig(name=step.step_name)
            self._steps.append((step, config))

run async

run(input_data: InputT) -> OutputT

Execute all steps sequentially, returning the final output.

Source code in orqest/orchestration/pipeline.py

async def run(self, input_data: InputT) -> OutputT:
    """Execute all steps sequentially, returning the final output."""
    data: Any = input_data
    async for event in self.run_stream(input_data):
        if event.event_type == "pipeline_error":
            raise event.error  # type: ignore[misc]
        if event.event_type == "pipeline_complete":
            data = event.data.get("output", data)
    return data  # type: ignore[return-value]

run_stream async

run_stream(
    input_data: InputT,
) -> AsyncIterator[PipelineEvent]

Execute steps and yield PipelineEvent at each lifecycle point.

Source code in orqest/orchestration/pipeline.py

async def run_stream(self, input_data: InputT) -> AsyncIterator[PipelineEvent]:
    """Execute steps and yield PipelineEvent at each lifecycle point."""
    yield PipelineEvent(
        event_type="pipeline_start",
        pipeline_name=self.name,
    )

    data: Any = input_data
    for index, (step, config) in enumerate(self._steps):
        step_name = config.name or step.step_name
        yield PipelineEvent(
            event_type="step_start",
            pipeline_name=self.name,
            step_name=step_name,
            step_index=index,
        )
        try:
            data = await self._execute_step(step, config, data, index)
            yield PipelineEvent(
                event_type="step_complete",
                pipeline_name=self.name,
                step_name=step_name,
                step_index=index,
                data={"output": data},
            )
        except StepSkipped:
            yield PipelineEvent(
                event_type="step_skip",
                pipeline_name=self.name,
                step_name=step_name,
                step_index=index,
            )
        except PipelineStepError as exc:
            yield PipelineEvent(
                event_type="pipeline_error",
                pipeline_name=self.name,
                step_name=step_name,
                step_index=index,
                error=exc,
            )
            raise

    yield PipelineEvent(
        event_type="pipeline_complete",
        pipeline_name=self.name,
        data={"output": data},
    )

PipelineStepError

PipelineStepError(
    message: str, *, step_name: str, step_index: int
)

Raised when a step fails with ErrorStrategy.STOP.

Store the failing step's name and index alongside the message.

Source code in orqest/orchestration/pipeline.py

def __init__(self, message: str, *, step_name: str, step_index: int) -> None:
    """Store the failing step's name and index alongside the message."""
    super().__init__(message)
    self.step_name = step_name
    self.step_index = step_index

Route

Route(
    name: str,
    step: Step | BaseAgent | Callable[..., Any],
    *,
    condition: Callable[[Any], bool] | None = None,
)

A named step with an optional condition for rule-based routing.

Initialize a route.

Parameters:

Name	Type	Description	Default
name	str	Identifier used for classifier-based selection.	required
step	Step | BaseAgent | Callable[..., Any]	The StepLike to execute when this route is selected.	required
condition	Callable[[Any], bool] | None	Predicate evaluated in rule-based mode. None means this route is only reachable via classifier.	None

Source code in orqest/orchestration/router.py

def __init__(
    self,
    name: str,
    step: Step | BaseAgent | Callable[..., Any],
    *,
    condition: Callable[[Any], bool] | None = None,
) -> None:
    """Initialize a route.

    Args:
        name: Identifier used for classifier-based selection.
        step: The StepLike to execute when this route is selected.
        condition: Predicate evaluated in rule-based mode. None means
            this route is only reachable via classifier.

    """
    self.name = name
    self._step: Step = _coerce_step(step)
    self.condition = condition

step property

step: Step

The coerced Step for this route.

Router

Router(
    routes: list[Route],
    *,
    classifier: BaseAgent
    | Callable[..., Any]
    | None = None,
    fallback: Step
    | BaseAgent
    | Callable[..., Any]
    | None = None,
    name: str = "router",
)

Select and execute a single route based on input.

In rule-based mode, routes are evaluated in order and the first whose condition returns True is selected. In classifier mode, an agent or callable picks the route by name.

Initialize the router.

Parameters:

Name	Type	Description	Default
routes	list[Route]	Ordered list of Route objects.	required
classifier	BaseAgent | Callable[..., Any] | None	Agent or callable that returns a route name.	None
fallback	Step | BaseAgent | Callable[..., Any] | None	Step executed when no route matches.	None
name	str	Identifier for logging and events.	'router'

Raises:

Type	Description
ValueError	If routes is empty, or if no classifier is provided and none of the routes have conditions.

Source code in orqest/orchestration/router.py

def __init__(
    self,
    routes: list[Route],
    *,
    classifier: BaseAgent | Callable[..., Any] | None = None,
    fallback: Step | BaseAgent | Callable[..., Any] | None = None,
    name: str = "router",
) -> None:
    """Initialize the router.

    Args:
        routes: Ordered list of Route objects.
        classifier: Agent or callable that returns a route name.
        fallback: Step executed when no route matches.
        name: Identifier for logging and events.

    Raises:
        ValueError: If routes is empty, or if no classifier is provided
            and none of the routes have conditions.

    """
    if not routes:
        raise ValueError("Router requires at least one route.")
    has_conditions = any(r.condition is not None for r in routes)
    if not has_conditions and classifier is None:
        raise ValueError(
            "Router needs either route conditions or a classifier."
        )
    self._routes = list(routes)
    self._classifier = classifier
    self._fallback: Step | None = _coerce_step(fallback) if fallback else None
    self._name = name
    self._route_map: dict[str, Route] = {r.name: r for r in routes}

run async

run(input_data: InputT) -> OutputT

Route input_data to the selected step and return its output.

Raises:

Type	Description
RouterError	When no route matches and no fallback exists.

Source code in orqest/orchestration/router.py

async def run(self, input_data: InputT) -> OutputT:
    """Route input_data to the selected step and return its output.

    Raises:
        RouterError: When no route matches and no fallback exists.

    """
    route_name = await self._select_route(input_data)
    if route_name is not None and route_name in self._route_map:
        return await self._route_map[route_name].step.execute(input_data)
    if self._fallback is not None:
        return await self._fallback.execute(input_data)
    raise RouterError(
        f"No route matched for input and no fallback configured "
        f"(router={self._name!r})."
    )

RouterError

Raised when no route matches and no fallback is configured.

AgentStep

AgentStep(
    agent: BaseAgent,
    *,
    prompt_builder: Callable[[Any], str] | None = None,
)

Wraps a BaseAgent as a pipeline Step.

Creates a fresh GlobalState per invocation so each step execution is stateless. Uses a prompt_builder to convert arbitrary input into a string prompt for the agent.

Initialize with a BaseAgent and optional prompt builder.

Parameters:

Name	Type	Description	Default
agent	BaseAgent	The BaseAgent to wrap.	required
prompt_builder	Callable[[Any], str] | None	Converts input_data to a prompt string. Defaults to JSON for BaseModel, str() otherwise.	None

Source code in orqest/orchestration/step.py

def __init__(
    self,
    agent: BaseAgent,
    *,
    prompt_builder: Callable[[Any], str] | None = None,
) -> None:
    """Initialize with a BaseAgent and optional prompt builder.

    Args:
        agent: The BaseAgent to wrap.
        prompt_builder: Converts input_data to a prompt string.
            Defaults to JSON for BaseModel, str() otherwise.

    """
    self._agent = agent
    self._prompt_builder = prompt_builder or _default_prompt

step_name property

step_name: str

Return the wrapped agent's name.

execute async

execute(input_data: Any) -> Any

Run the agent with a fresh state built from input_data.

Source code in orqest/orchestration/step.py

async def execute(self, input_data: Any) -> Any:
    """Run the agent with a fresh state built from *input_data*."""
    state = GlobalState()
    prompt = self._prompt_builder(input_data)
    state.add_message("user", prompt)
    return await self._agent.run(state)

FunctionStep

FunctionStep(
    func: Callable[..., Any], *, name: str | None = None
)

Wraps an async callable as a pipeline Step.

Initialize with an async callable and optional name.

Parameters:

Name	Type	Description	Default
func	Callable[..., Any]	The async callable to wrap.	required
name	str | None	Step name; defaults to the function's name.	None

Source code in orqest/orchestration/step.py

def __init__(
    self,
    func: Callable[..., Any],
    *,
    name: str | None = None,
) -> None:
    """Initialize with an async callable and optional name.

    Args:
        func: The async callable to wrap.
        name: Step name; defaults to the function's __name__.

    """
    self._func = func
    self._name = name or getattr(func, "__name__", "function_step")

step_name property

step_name: str

Return the configured or inferred step name.

execute async

execute(input_data: Any) -> Any

Call the wrapped function with input_data.

Source code in orqest/orchestration/step.py

async def execute(self, input_data: Any) -> Any:
    """Call the wrapped function with *input_data*."""
    return await self._func(input_data)

Step

Minimal interface for an executable pipeline step.

step_name property

step_name: str

Human-readable name for this step.

execute async

execute(input_data: Any) -> Any

Run the step on input_data and return a result.

Source code in orqest/orchestration/step.py

async def execute(self, input_data: Any) -> Any:
    """Run the step on *input_data* and return a result."""
    ...

ErrorStrategy

How to handle a step failure during pipeline execution.

PipelineEvent dataclass

PipelineEvent(
    event_type: EventType,
    pipeline_name: str,
    step_name: str = "",
    step_index: int = -1,
    timestamp: datetime = (lambda: datetime.now(tz=UTC))(),
    data: dict[str, Any] = dict(),
    error: Exception | None = None,
)

Event emitted during pipeline execution for observability.

StepConfig dataclass

StepConfig(
    name: str = "",
    on_error: ErrorStrategy = ErrorStrategy.STOP,
    max_retries: int = 1,
)

Per-step configuration for pipeline execution.

Controls the step's name, error-handling strategy, and retry limits.