Pipelines

Multi-stage flows where one stage can reject the next.

A staged pipeline runs one agent through a sequence of stages. Each stage has its own prompt, its own tool set, and optionally a verdict tool it must call before the pipeline moves on. A shared fact store carries structured data between stages, so nothing depends on the agent summarising its own work into text and reading it back.

from agentino.pipeline.staged import FactStore, StageDef, StagedPipeline


class TriageFacts(FactStore):
    task_type: str = ""

    def to_context(self) -> str:
        return f"task_type={self.task_type}"


stages = [
    StageDef(name="classify", prompt="Decide what kind of task this is.",
             tools=["read_file"], verdict_tool="classify_verdict"),
    StageDef(name="execute", prompt="Carry it out.",
             tools=["read_file", "write_file"]),
    StageDef(name="report", prompt="Summarise what changed.",
             tools=["report"], verdict_tool="report"),
]

pipeline = StagedPipeline(stages=stages, facts=TriageFacts())
result = await pipeline.run(agent_template, task_text)

Every stage field

	Default
`name`	—	Stage identifier, used in results and logs
`prompt`	`""`	Instructions for this stage alone
`tools`	all	Tool names this stage may call. A stage that should not write does not get the write tool
`verdict_tool`	—	A tool the stage must call to finish. Without it a stage ends when the agent stops talking
`gate`	—	A gate that must be marked before this stage runs
`max_turns`	`10`	Cap on this stage's tool-calling loop
`skip_condition`	—	`f(facts) -> bool`; true skips the stage
`repeatable`	`false`	Whether the stage may run more than once
`max_cycles`	`3`	How many times, when it is repeatable
`on_fail`	`"retry"`	What to do when the stage fails
`number`	`0`	Ordering, when it is not the list order

Pipeline options

	Default
`stages`	—	The stages, in order
`facts`	—	A `FactStore` carried between them
`max_reprompts`	`3`	Retries when a stage does not produce its verdict
`global_max_cycles`	`20`	Ceiling across the whole run, so a repeatable stage cannot loop forever
`failure_check`	—	`f(result, facts) -> bool`; your own definition of a failed stage

Why the tool set is per stage

Restricting tools per stage is the part that carries weight. A classify stage with only read_file cannot write, whatever the prompt says and whatever the model decides — the tool is not there to call. That is a structural guarantee rather than an instruction, which is the difference between a pipeline and a long prompt.

verdict_tool is the other half: a stage that must call a named tool to finish cannot end by producing prose that merely looks like a decision.

The fact store

Stages share a FactStore — any object with a to_context() method. Its output is injected into the next stage's prompt.

The alternative is asking the agent to summarise its own work as text and parse it back, which loses structure and fails quietly when the wording drifts. A field on an object does not.