Demonstrates step-by-step agent execution where each agent processes the output of the previous agent in sequence. Perfect for content creation pipelines, data processing workflows, and multi-stage analysis.
- Sequential Workflows: Step-by-step agent execution
- Data Flow: How information passes between agents
- Flow Orchestration: Using TFrameX Flow and patterns
- Pipeline Design: Creating effective processing pipelines
- Content Creation: Real-world content development workflow
sequential-pattern/
โโโ README.md # This guide
โโโ requirements.txt # Dependencies
โโโ .env.example # Environment template
โโโ main.py # Main application
โโโ docs/
โโโ sequential_flows.md # Flow design patterns
# Install dependencies
pip install -r requirements.txt
# Configure environment
cp .env.example .env
# Edit .env with your LLM settings
# Run the example
python main.pyInput โ Agent 1 โ Agent 2 โ Agent 3 โ Final Output
โ โ โ
Step 1 Step 2 Step 3
(Plan) (Write) (Edit)
- ContentPlanner: Analyzes topic and creates structured plan
- ContentWriter: Writes content based on the plan
- ContentEditor: Reviews and improves the content
# Input: "Create a blog post about renewable energy for small businesses"
#
# Step 1 - ContentPlanner:
# โ Analyzes topic
# โ Identifies target audience
# โ Creates content structure
# โ Defines key points
#
# Step 2 - ContentWriter:
# โ Takes the plan from Step 1
# โ Writes engaging content
# โ Follows the structure
# โ Expands on key points
#
# Step 3 - ContentEditor:
# โ Takes the content from Step 2
# โ Reviews for grammar and style
# โ Improves clarity and flow
# โ Provides final polished versionComplete automated content creation pipeline:
python main.py
# Select option 1See each individual step in the sequence:
python main.py
# Select option 2Chat with individual agents:
python main.py
# Select option 3- Each step builds on the previous
- Transparent progression
- Easy to debug and optimize
- Each agent has a focused role
- Expertise in specific tasks
- Reusable across different flows
- Iterative refinement
- Multiple passes for quality
- Structured improvement process
- Easy to modify individual steps
- Add or remove agents as needed
- Clear separation of concerns
@app.agent(
name="StepOneAgent",
description="First step in the process",
system_prompt="Your role in the sequential process..."
)
async def step_one_agent():
pass# Create sequential flow
workflow = Flow(
flow_name="MySequentialFlow",
description="Step-by-step processing workflow"
)
# Add steps in order
workflow.add_step("StepOneAgent")
workflow.add_step("StepTwoAgent")
workflow.add_step("StepThreeAgent")
# Register with app
app.register_flow(workflow)async with app.run_context() as rt:
initial_input = Message(role="user", content="Process this...")
result = await rt.run_flow("MySequentialFlow", initial_input)
print(result.current_message.content)- Blog post writing
- Report generation
- Documentation creation
- Marketing copy development
- Data cleaning pipelines
- Analysis workflows
- Report generation
- Quality assurance processes
- Research processes
- Due diligence workflows
- Assessment procedures
- Evaluation pipelines
- Story development
- Design workflows
- Product development
- Creative reviews
# Add conditional logic within agents
@app.agent(
name="ConditionalAgent",
system_prompt="If the content needs revision, suggest improvements. Otherwise, approve it."
)
async def conditional_agent():
pass# Agents can handle errors and provide feedback
@app.agent(
name="RobustAgent",
system_prompt="If the previous step failed, provide alternative approach or error correction."
)
async def robust_agent():
pass# Agents can act as quality checkpoints
@app.agent(
name="QualityGate",
system_prompt="Review the work and only pass it forward if it meets quality standards."
)
async def quality_gate():
pass- Keep system prompts focused
- Use clear, specific instructions
- Minimize unnecessary processing
- Optimize the number of steps
- Balance specialization vs overhead
- Consider parallel alternatives for independent tasks
- Use appropriate history limits
- Clear unnecessary context
- Manage token usage efficiently
# Run each step individually to identify issues
step1_result = await rt.call_agent("Agent1", input_message)
step2_result = await rt.call_agent("Agent2", step1_result)
step3_result = await rt.call_agent("Agent3", step2_result)# Examine flow context at each step
flow_context = await rt.run_flow("MyFlow", input_message)
print("History:", flow_context.history)
print("Shared Data:", flow_context.shared_data)- Enable detailed logging
- Monitor agent performance
- Track success/failure rates
- Measure processing times
After mastering sequential patterns:
- Try Parallel Patterns: Parallel Pattern Example
- Explore Router Patterns: Router Pattern Example
- Advanced Workflows: Code Review System
- Build Custom Flows: Create your own sequential workflows
- Single Responsibility: Each agent should have one clear purpose
- Clear Interfaces: Define what each agent expects and produces
- Error Resilience: Plan for failures and edge cases
- Testability: Make each step independently testable
- Start with simple 2-3 step flows
- Test each agent individually first
- Use descriptive agent names and descriptions
- Document the flow purpose and expected outcomes
This example is provided under the MIT License.