Sequential processing runs agents in a developer-defined order, passing each agent’s context to the next. It fits tasks with a natural progression such as research, writing, and review. The fixed flow is predictable and easy to debug. section 9.3 section 9.3.2
Implement the pipeline around a shared workflow interface:
- Give the first agent both user_input and the existing execution context. section 9.3.2
- Pass only the updated context to later agents, avoiding duplicate user input. section 9.3.2
- Reset final_result and current_step before each execution so one agent’s termination state does not affect the next. section 9.3.2
- Return the final agent’s result as the workflow result. section 9.3.2
- Implement run() on the workflow so it can be composed wherever an agent is expected. section 9.3.1
Research, write, review
Create researcher, writer, and editor agents, then run them as SequentialWorkflow(agents=[researcher, writer, editor]). The researcher’s findings become the writer’s context, and the draft becomes the editor’s input. section 9.3.2 section 9.3.2
Listing 9.5 SequentialWorkflow implementation
class SequentialWorkflow(Agent):
def __init__(self, agents: List[Agent], name: str = "sequential_workflow"):
self.agents = agents
self.name = name
async def run(
self,
user_input: Optional[str] = None,
context: Optional[ExecutionContext] = None,
verbose: bool = False,
) -> AgentResult:
result = None
for i, agent in enumerate(self.agents):
if context is not None:
# Reset state for next agent execution
context.final_result = None
context.current_step = 0
if i == 0:
# First agent: pass both user_input and context
result = await agent.run(user_input, context=context, verbose=verbose)
else:
# Subsequent agents: pass only context (no user_input)
result = await agent.run(context=context, verbose=verbose)
context = result.context
return result
Inspect the run() implementation to see input branching, context propagation, and per-agent state resets. section 9.3.2
