Skip to content

Composition — pipelines

The composition primitives are for flows you can write as a regular function: do A, then B, then C — with optional fan-out and merge.

Composition pipelines Three ways to compose agents, each returning one PipelineResult. Sequential pipeline: Research gets the task; Draft and then Review each get the previous agent's output plus the original task; the last output becomes final_output. Parallel pipeline: Web search, Data query and Docs retrieval all get the same task (unless a task_map gives one its own) and run at once; a merge step concatenates their non-empty outputs, or keeps only the output of the last agent in the list, into final_output. Loop agent: a Reviser agent runs, then a check runs its condition on the output; if the condition is not met the Reviser runs again on its output plus the task; the loop ends when the condition is met or when max_loops runs are reached, and the last output becomes final_output. If any agent raises, the result's success is False. Sequential pipeline task output + task output + task last output Research agents[0] Draft agents[1] Review agents[2] PipelineResult final_output Parallel pipeline same task Web search agents[0] Data query agents[1] Docs retrieval agents[2] Merge concatenate or last PipelineResult final_output Loop agent task not met: output + task met max_loops reached Reviser agent Check condition(output) PipelineResult final_output Each run returns one PipelineResult. If any agent raises, the result's success is False.

What it is

Three BaseModel-shaped pipeline classes, each wrapping one or more agents:

Class Shape
SequentialPipeline(agents=[...]) output of agent N feeds agent N+1
ParallelPipeline(agents=[...]) one input fans out to all N agents; results merge
LoopAgent(agent=..., max_loops=N) run one agent repeatedly until a condition holds or N is hit

Each has an async .run(task) that calls each of its agents' arun (or run_sync, for agents without arun) and returns one PipelineResult. A pipeline has no arun or run_sync of its own, and its result carries final_output rather than .message. So a pipeline is not itself agent-shaped, and pipelines do not nest: a pipeline placed inside another never runs, and the outer pipeline returns success=False.

Inside an active run_context, the pipeline's composition.* stage, fan-out and loop events go to the same event bus as its agents' events.

When to use it

  • ✅ The flow is describable as a function — "do A, then B, then C".
  • ✅ Fan-out is symmetric — all branches do similar work on the same input (e.g., answer one question across web search, a data query, and docs retrieval).
  • ✅ The flow is a report chain — research → draft → review, each step feeding the next.
  • ✅ You need revise-until-confidence — wrap the report writer in a LoopAgent until the confidence bar clears.
  • ✅ You don't need cycles, conditional branches, or per-node retry policies.

When NOT to use it

  • ❌ You need cycles that depend on state — use StateGraph.
  • ❌ A central agent should decide which expert runs — use Orchestrator.
  • ❌ The branches need to talk to each other — use Swarm.

Code

from tulip.agent.composition import (
    SequentialPipeline, ParallelPipeline, LoopAgent,
)

# Sequential: research → draft → review → publish
pipeline = SequentialPipeline(agents=[
    research,
    draft,
    review,
    publish,
])

result = await pipeline.run("Why did checkout conversion drop last week?")

# Parallel: answer one question across web search, a data query, and docs
parallel = ParallelPipeline(agents=[
    web_search_agent,
    data_query_agent,
    docs_retrieval_agent,
])
answers = await parallel.run(
    "Checkout conversion last week: what changed? which deploys? known issues?"
)

# Loop: revise the draft until the review clears, max 5 loops
revise = LoopAgent(agent=reviser_agent, max_loops=5)
final = await revise.run(initial_draft)

Notebooks

Source

agent/composition.py — SequentialPipeline, ParallelPipeline, LoopAgent.

See also