latent.agents.pipeline.types¶

Pipeline agent types — phases, transitions, and state.

Classes¶

FunctionPhase¶

FunctionPhase(name: str, transitions: list[Transition | Respond] = list(), fn: Callable[[PipelineState, list[Message]], Any] = lambda s, m: None)

Run arbitrary Python code. The function handles its own logic.

fn receives (state, messages) and returns output stored in state.outputs[phase.name]. Must be async def.

Phase¶

Phase(name: str, transitions: list[Transition | Respond] = list())

Base phase with transitions. Not used directly for execution.

Pipeline¶

Pipeline(name: str, phases: list[Phase], initial: str, model: str = '', max_total_phases: int = 20)

A directed graph of processing phases with deterministic routing.

Args: name: Pipeline identifier. phases: Ordered list of phases (declaration order preserved). initial: Name of the first phase to execute. model: Default model for phases that don't override. max_total_phases: Safety limit against infinite routing loops.

PipelineState¶

PipelineState(current_phase: str, outputs: dict[str, Any] = dict(), retry_counts: dict[str, int] = dict(), phase_history: list[str] = list(), metadata: dict[str, Any] = dict())

Mutable runtime state for one pipeline execution.

PromptPhase¶

PromptPhase(name: str, transitions: list[Transition | Respond] = list(), instructions: str | Callable[[PipelineState, list[Message]], str] = '', output_schema: type | None = None, model: str | None = None, temperature: float = 0.0, max_tokens: int = 4096)

Single LLM call. Structured output if output_schema is set, free text otherwise.

ReActPhase¶

ReActPhase(name: str, transitions: list[Transition | Respond] = list(), instructions: str | Callable[[PipelineState, list[Message]], str] = '', tools: list[Callable] = list(), max_iterations: int = 3, output_schema: type | None = None, model: str | None = None, temperature: float = 0.0, max_tokens: int = 4096)

Bounded tool-calling loop via a scoped ReActAgent.

Respond¶

Respond(output: Callable[[Any, PipelineState], str], when: Callable[[Any, PipelineState], bool] | None = None, side_effect: Callable[[Any, PipelineState], None] | None = None, instructions: str | Callable[[Any, PipelineState], str] | None = None, model: str | None = None)

Terminal transition — emit output as TextDelta and stop the pipeline.

Lives in a phase's transitions list alongside Transition. When matched, the pipeline calls output(phase_output, state) to get the final text, emits it as TextDelta, and returns.

If instructions is set, the output string is passed through a final LLM formatting call (e.g. to ensure language/tone consistency) before emission. If None, the output is emitted as-is.

Args: output: (phase_output, state) -> str producing the response text. when: Predicate. None = always match. side_effect: Optional callback before output. instructions: Optional system prompt for a final formatting LLM call. Can be a string or (phase_output, state) -> str. model: Model override for the formatting call.

Transition¶

Transition(to: str, when: Callable[[Any, PipelineState], bool] | None = None, side_effect: Callable[[Any, PipelineState], None] | None = None)

Route to another phase. First match in the transitions list wins.

Args: to: Target phase name. when: Predicate (phase_output, state) -> bool. None = always match (default). side_effect: Optional (phase_output, state) -> None called before routing.

Methods¶

PipelineState.retries¶

retries(phase_name: str) -> int

Number of times a phase has been re-entered.