latent.agents.pipeline.decorators¶

Decorator-based API for defining PipelineAgent phases.

Decorators mark methods as pipeline phases, discovered automatically when PipelineAgent is instantiated without an explicit Pipeline.

Usage::

class MyAgent(PipelineAgent):
    @phase(output_schema=ClassifyResult)
    def classify(self, state, messages):
        return "Classify the user query."

    @tool(transitions={"respond": None})
    def lookup(self, state, messages):
        return do_search(state.outputs["classify"].query)

    @respond
    def respond(self, state, messages):
        return state.outputs["lookup"]

Functions¶

discover_phases¶

discover_phases(instance: Any) -> list[tuple[str, dict, Callable]] | None

Scan instance for @phase/@tool/@respond/@subagent methods.

Returns a list of (method_name, meta_dict, bound_method) in declaration order, or None if no phase methods are found.

phase¶

phase(model: str | None = None, output_schema: type | None = None, temperature: float = 0.0, max_tokens: int = 4096, transitions: dict[str, Callable | None] | None = None) -> Callable

LLM call phase. Method returns instructions string, framework calls LLM.

Signature: (self, state: PipelineState, messages: list[Message]) -> str

respond¶

respond(fn: Callable | None = None, model: str | None = None) -> Callable

Terminal phase. Emits method return value as TextDelta.

If model is given, return value is used as a prompt and the LLM response is streamed instead.

Can be used bare (@respond) or with args (@respond(model=...)). Signature: (self, state, messages) -> str

subagent¶

subagent(tools: list[Callable] | None = None, max_iterations: int = 3, model: str | None = None, temperature: float = 0.0, max_tokens: int = 4096, transitions: dict[str, Callable | None] | None = None) -> Callable

Scoped agent loop. Method returns instructions, framework runs bounded ReAct loop.

Signature: (self, state, messages) -> str

tool¶

tool(transitions: dict[str, Callable | None] | None = None) -> Callable

Python function phase. Runs code, result stored in state.outputs.

Method can be sync or async. Signature: (self, state, messages) -> Any