A powerful Python workflow orchestration framework with advanced resource management and observability
PuffinFlow is a high-performance Python framework for building production-ready LLM workflows and multi-agent systems.
Perfect for AI engineers, data scientists, and backend developers who need to build reliable, scalable, and observable workflow orchestration systems.
Install PuffinFlow:
pip install puffinflow
Create your first agent with state management:
from puffinflow import Agent, state
class DataProcessor(Agent):
@state(cpu=2.0, memory=1024.0)
async def fetch_data(self, context):
"""Fetch data from external source."""
data = await get_external_data()
context.set_variable("raw_data", data)
return "validate_data" if data else "error"
@state(cpu=1.0, memory=512.0)
async def validate_data(self, context):
"""Validate the fetched data."""
data = context.get_variable("raw_data")
if self.is_valid(data):
return "process_data"
return "error"
@state(cpu=4.0, memory=2048.0)
async def process_data(self, context):
"""Process the validated data."""
data = context.get_variable("raw_data")
result = await self.transform_data(data)
context.set_output("processed_data", result)
return "complete"
# Run the agent
agent = DataProcessor("data-processor")
result = await agent.run()
Production-Ready Performance: Sub-millisecond latency for basic operations with throughput exceeding 12,000 ops/s.
Intelligent Resource Management: Automatic allocation and management of CPU, memory, and other resources with built-in quotas and limits.
Zero-Configuration Observability: Comprehensive monitoring with OpenTelemetry integration, custom metrics, distributed tracing, and real-time alerting.
Built-in Reliability: Circuit breakers, bulkheads, timeout handling, and leak detection ensure robust operation under failure conditions.
Multi-Agent Coordination: Scale from single agents to complex multi-agent workflows with teams, pools, and orchestrators.
Seamless Development Experience: Prototype quickly and transition to production without code rewrites.
PuffinFlow delivers exceptional performance in production workloads. Our comprehensive benchmark suite compares PuffinFlow against leading orchestration frameworks.
Native API Framework Performance (vs LangGraph and LlamaIndex)
| Framework | Total Execution | Framework Overhead | Efficiency | Concurrent Workflows | Success Rate |
|---|---|---|---|---|---|
| 🥇 PuffinFlow | 1.5ms | 41.9% | 58.1% | 5 workflows | 100% |
| 🥈 LlamaIndex | 1.5ms | 52.6% | 47.4% | 4 workflows | 100% |
| 🥉 LangGraph | 2.2ms | 62.7% | 37.3% | 3 workflows | 100% |
Simple Workflow Performance
| Framework | Execution Time | vs PuffinFlow | Performance Rating |
|---|---|---|---|
| 🥇 PuffinFlow | 0.8ms | Baseline | 🚀 Best |
| 🥈 LlamaIndex | 1.5ms | +88% slower | ✅ Good |
| 🥉 LangGraph | 12.4ms | +1,450% slower | ⚠️ Poor |
Complex Workflow Performance
| Framework | Execution Time | vs PuffinFlow | Performance Rating |
|---|---|---|---|
| 🥇 PuffinFlow | 1.0ms | Baseline | 🚀 Best |
| 🥈 LlamaIndex | 1.5ms | +50% slower | ✅ Good |
| 🥉 LangGraph | 1.8ms | +80% slower | ⚠️ Fair |
Multi-Agent Workflow Performance
| Framework | Execution Time | vs PuffinFlow | Performance Rating |
|---|---|---|---|
| 🥇 PuffinFlow | 2.1ms | Baseline | 🚀 Best |
| 🥈 LlamaIndex | 3.7ms | +76% slower | ✅ Good |
| 🥉 LangGraph | 5.8ms | +176% slower | ⚠️ Poor |
Error Recovery Workflow Performance
| Framework | Execution Time | vs Best | Performance Rating |
|---|---|---|---|
| 🥇 LlamaIndex | 0.5ms | Baseline | 🚀 Best |
| 🥈 LangGraph | 0.6ms | +20% slower | 🚀 Excellent |
| 🥉 PuffinFlow | 0.8ms | +60% slower | ✅ Good |
Overall Multi-Workflow Average
| Framework | Average Time | vs PuffinFlow | Overall Rating |
|---|---|---|---|
| 🥇 PuffinFlow | 1.2ms | Baseline | 🚀 Champion |
| 🥈 LlamaIndex | 1.8ms | +50% slower | ✅ Strong |
| 🥉 LangGraph | 5.1ms | +325% slower | ⚠️ Variable |
🏆 Comprehensive Performance Analysis vs LangGraph and LlamaIndex
Core Execution Performance (Measured)
Resource Efficiency (Measured)
Standardized Concurrent Performance (Measured)
Core Workflow Performance (Measured)
Overall Multi-Workflow Performance (Measured)
Testing Coverage
Key Performance Insights
Latest benchmarks test both native API patterns and core workflow capabilities across all three frameworks. All concurrent workflow testing uses standardized 3-workflow loads for fair comparison. Testing covers the 3 essential workflow patterns for production use: simple single-task execution, complex multi-step dependencies, and parallel multi-agent coordination using each framework's recommended API design patterns.
class ImageProcessor(Agent):
@state(cpu=2.0, memory=1024.0)
async def resize_image(self, context):
image_url = context.get_variable("image_url")
resized = await resize_image(image_url, size=(800, 600))
context.set_variable("resized_image", resized)
return "add_watermark"
@state(cpu=1.0, memory=512.0)
async def add_watermark(self, context):
image = context.get_variable("resized_image")
watermarked = await add_watermark(image)
context.set_variable("final_image", watermarked)
return "upload_to_storage"
@state(cpu=1.0, memory=256.0)
async def upload_to_storage(self, context):
image = context.get_variable("final_image")
url = await upload_to_s3(image)
context.set_output("result_url", url)
return "complete"
class MLTrainer(Agent):
@state(cpu=8.0, memory=4096.0)
async def train_model(self, context):
dataset = context.get_variable("dataset")
model = await train_neural_network(dataset)
context.set_variable("model", model)
context.set_output("accuracy", model.accuracy)
if model.accuracy > 0.9:
return "deploy_model"
return "retrain_with_more_data"
@state(cpu=2.0, memory=1024.0)
async def deploy_model(self, context):
model = context.get_variable("model")
await deploy_to_production(model)
context.set_output("deployment_status", "success")
return "complete"
from puffinflow import create_team, AgentTeam
# Coordinate multiple agents
email_team = create_team([
EmailValidator("validator"),
EmailProcessor("processor"),
EmailTracker("tracker")
])
# Execute with built-in coordination
result = await email_team.execute_parallel()
Data Pipelines: Build resilient ETL workflows with automatic retries, resource management, and comprehensive monitoring.
ML Workflows: Orchestrate training pipelines, model deployment, and inference workflows with checkpointing and observability.
Microservices: Coordinate distributed services with circuit breakers, bulkheads, and intelligent load balancing.
Event Processing: Handle high-throughput event streams with backpressure control and automatic scaling.
API Orchestration: Coordinate complex API interactions with built-in retry policies and error handling.
PuffinFlow integrates seamlessly with popular Python frameworks:
FastAPI & Django: Native async support for web application integration with automatic resource management.
Celery & Redis: Enhance existing task queues with stateful workflows, advanced coordination, and monitoring.
OpenTelemetry: Complete observability stack with distributed tracing, metrics, and monitoring platform integration.
Kubernetes: Production-ready deployment with container orchestration and cloud-native observability.
PuffinFlow is built on a robust, production-tested architecture:
We welcome contributions from the community. Please see our Contributing Guide for details on how to get started.
PuffinFlow is released under the MIT License. Free for commercial and personal use.
Ready to build production-ready workflows?
FAQs
A powerful Python workflow orchestration framework with advanced resource management and observability
We found that puffinflow demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Research
/Security News
Socket researchers discovered nine malicious NuGet packages that use time-delayed payloads to crash applications and corrupt industrial control systems.
Security News
Socket CTO Ahmad Nassri discusses why supply chain attacks now target developer machines and what AI means for the future of enterprise security.
Security News
Learn the essential steps every developer should take to stay secure on npm and reduce exposure to supply chain attacks.