Compare commits

...
5 Commits
Author SHA1 Message Date
Mervin Praison aa81094ac3 Release v2.3.34 2025-12-18 14:55:33 +00:00
Mervin Praison c74fd7e925 Release v2.3.33 2025-12-18 14:41:36 +00:00
Mervin Praison 249acf7965 docs: Add performance benchmarks and auto-update functionality
Add performance comparison table to README showing PraisonAI as fastest framework for agent instantiation. Include benchmarks for LangGraph and CrewAI frameworks. Implement auto-save functionality to generate BENCHMARK_RESULTS.md and update README.md with latest benchmark results.
2025-12-18 13:30:06 +00:00
Mervin Praison ab68b2d296 Release v2.3.32 2025-12-18 13:17:32 +00:00
Mervin Praison abcc5b3b04 Release v2.3.31 2025-12-18 12:57:05 +00:00
33 changed files with 3147 additions and 235 deletions

No files matched your search

+57
View File
@@ -34,6 +34,7 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
## 📑 Table of Contents
- [⚡ Performance](#-performance)
- [🚀 Quick Start](#-quick-start)
- [📦 Installation](#-installation)
- [Python SDK](#python-sdk)
@@ -73,6 +74,32 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
---
## ⚡ Performance
PraisonAI Agents is the **fastest AI agent framework** for agent instantiation.
| Framework | Avg Time (μs) | Relative |
|-----------|---------------|----------|
| **PraisonAI** | **3.77** | **1.00x (fastest)** |
| OpenAI Agents SDK | 5.26 | 1.39x |
| Agno | 5.64 | 1.49x |
| PraisonAI (LiteLLM) | 7.56 | 2.00x |
| PydanticAI | 226.94 | 60.16x |
| LangGraph | 4,558.71 | 1,209x |
| CrewAI | 15,607.92 | 4,138x |
<details>
<summary>Run benchmarks yourself</summary>
```bash
cd praisonai-agents
python benchmarks/simple_benchmark.py
```
</details>
---
## 🚀 Quick Start
Get started with PraisonAI in under 1 minute:
@@ -1392,6 +1419,36 @@ praisonai commit
praisonai commit --push
```
### Serve CLI (API Server):
```bash
# Start API server for agents defined in YAML
praisonai serve agents.yaml
# With custom port and host
praisonai serve agents.yaml --port 8005 --host 0.0.0.0
# Alternative flag style
praisonai agents.yaml --serve
# The server provides:
# POST /agents - Run all agents sequentially
# POST /agents/{name} - Run specific agent (e.g., /agents/researcher)
# GET /agents/list - List available agents
```
### n8n Integration CLI:
```bash
# Export workflow to n8n and open in browser
praisonai agents.yaml --n8n
# With custom n8n URL
praisonai agents.yaml --n8n --n8n-url http://localhost:5678
# Set N8N_API_KEY for auto-import
export N8N_API_KEY="your-api-key"
praisonai agents.yaml --n8n
```
### @Mentions in Prompts:
```bash
# Include file content in prompt
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.30" \
"praisonai>=2.3.34" \
"praisonai[chat]" \
"embedchain[github,youtube]"
+1 -1
View File
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.30" \
"praisonai>=2.3.34" \
"praisonai[ui]" \
"praisonai[chat]" \
"praisonai[realtime]" \
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.30" \
"praisonai>=2.3.34" \
"praisonai[ui]" \
"praisonai[crewai]"
@@ -0,0 +1,35 @@
# PraisonAI Agents - Benchmark Results
**Generated:** 2025-12-18 14:40:22
**Iterations:** 100
**Test:** Agent instantiation (without tools)
## Results
| Framework | Avg Time (μs) | Relative |
|-----------|---------------|----------|
| **PraisonAI** | **3.77** | **1.00x (fastest)** |
| OpenAI Agents SDK | 5.26 | 1.39x |
| Agno | 5.64 | 1.49x |
| PraisonAI (LiteLLM) | 7.56 | 2.00x |
| PydanticAI | 226.94 | 60.16x |
| LangGraph | 4,558.71 | 1,209x |
| CrewAI | 15,607.92 | 4,138x |
## Package Versions
| Package | Version |
|---------|--------|
| PraisonAI | 0.1.5 |
| Agno | 2.3.14 |
| PydanticAI | 1.35.0 |
| OpenAI Agents SDK | 0.6.3 |
| LangGraph | 1.0.5 |
| CrewAI | 1.6.1 |
## How to Reproduce
```bash
cd praisonai-agents
python benchmarks/simple_benchmark.py
```
@@ -0,0 +1,31 @@
# PraisonAI Agents - Real Execution Benchmark
**Generated:** 2025-12-18 14:43:52
**Model:** gpt-4o-mini
**Iterations:** 3
**Prompt:** "What is 2+2? Reply with just the number."
## Results
| Framework | Method | Avg Time | Relative |
|-----------|--------|----------|----------|
| **PraisonAI** | `agent.start()` | **0.45s** | **1.00x (fastest)** |
| PraisonAI (LiteLLM) | `agent.start()` | 0.55s | 1.22x |
| CrewAI | `crew.kickoff()` | 0.58s | 1.28x |
| Agno | `agent.run()` | 0.92s | 2.05x |
## Package Versions
| Package | Version |
|---------|--------|
| PraisonAI | 0.1.5 |
| Agno | 2.3.14 |
| CrewAI | 1.6.1 |
## How to Reproduce
```bash
export OPENAI_API_KEY=your_key
cd praisonai-agents
python benchmarks/execution_benchmark.py
```
@@ -0,0 +1,72 @@
# PraisonAI Agents - Real-World Execution Benchmark
**Generated:** 2025-12-18 14:33:49
**Model:** gpt-4o-mini
**Iterations:** 3 per test
## Results Summary
| Test | PraisonAI (avg) | Agno (avg) | Difference |
|------|-----------------|------------|------------|
| Single Agent | 2.04s | 3.18s | PraisonAI 1.56x faster |
| 2 Agents | 2.49s | 20.49s | PraisonAI 8.21x faster |
| 3 Agents | 6.52s | 13.97s | PraisonAI 2.14x faster |
## Detailed Results
### praisonai_single
- Average: 2.04s
- Min: 1.95s
- Max: 2.19s
- Std Dev: 0.13s
- Times: ['2.19s', '1.95s', '2.00s']
### agno_single
- Average: 3.18s
- Min: 3.00s
- Max: 3.51s
- Std Dev: 0.28s
- Times: ['3.51s', '3.00s', '3.04s']
### praisonai_two_agents
- Average: 2.49s
- Min: 2.22s
- Max: 2.65s
- Std Dev: 0.24s
- Times: ['2.65s', '2.22s', '2.62s']
### agno_two_agents
- Average: 20.49s
- Min: 9.96s
- Max: 35.05s
- Std Dev: 13.02s
- Times: ['16.46s', '9.96s', '35.05s']
### praisonai_three_agents
- Average: 6.52s
- Min: 5.62s
- Max: 7.39s
- Std Dev: 0.88s
- Times: ['5.62s', '6.54s', '7.39s']
### agno_three_agents
- Average: 13.97s
- Min: 11.69s
- Max: 15.32s
- Std Dev: 1.99s
- Times: ['14.91s', '11.69s', '15.32s']
## Package Versions
| Package | Version |
|---------|--------|
| PraisonAI | 0.1.5 |
| Agno | 2.3.14 |
## How to Reproduce
```bash
export OPENAI_API_KEY=your_key
cd praisonai-agents
python benchmarks/real_benchmark.py
```
@@ -0,0 +1,22 @@
# PraisonAI Agents - Benchmark Results
**Generated:** 2025-12-18 13:23:43
## Agent Instantiation Time
| Framework | Avg Time (μs) | Relative |
|-----------|---------------|----------|
| **PraisonAI** | **4.16** | **1.00x (fastest)** |
| OpenAI Agents SDK | 5.55 | 1.33x |
| Agno | 5.58 | 1.34x |
| PraisonAI (LiteLLM) | 8.52 | 2.05x |
| PydanticAI | 225.87 | 54.25x |
| LangGraph | 4,349.54 | 1,045x |
| CrewAI | 15,911.11 | 3,821x |
## How to Reproduce
```bash
cd praisonai-agents
python benchmarks/simple_benchmark.py
```
@@ -0,0 +1,33 @@
# PraisonAI Agents - Tools Benchmark Results
**Generated:** 2025-12-18 14:40:35
**Iterations:** 100
**Test:** Agent instantiation WITH TOOLS
## Results
| Framework | Avg Time (μs) | Relative |
|-----------|---------------|----------|
| **PraisonAI** | **3.24** | **1.00x (fastest)** |
| Agno | 5.12 | 1.58x |
| PraisonAI (LiteLLM) | 8.59 | 2.65x |
| OpenAI Agents SDK | 279.95 | 86.44x |
| LangGraph | 2,310.82 | 713x |
| CrewAI | 15,773.44 | 4,870x |
## Package Versions
| Package | Version |
|---------|--------|
| PraisonAI | 0.1.5 |
| Agno | 2.3.14 |
| OpenAI Agents SDK | 0.6.3 |
| LangGraph | 1.0.5 |
| CrewAI | 1.6.1 |
## How to Reproduce
```bash
cd praisonai-agents
python benchmarks/tools_benchmark.py
```
@@ -0,0 +1,368 @@
#!/usr/bin/env python3
"""
PraisonAI Agents - Deep Profiling Benchmark
Profiles each step of agent initialization to identify performance bottlenecks.
Key Optimizations Applied:
1. Lazy RulesManager - filesystem operations deferred until first access
2. Lazy os.getcwd() - fast_context_path resolved only when needed
3. Lazy console initialization - Rich Console created on demand
Usage:
python benchmarks/deep_profiling.py
"""
import time
import tracemalloc
import cProfile
import pstats
import io
from typing import Literal, Callable, Dict, List
from dataclasses import dataclass
# ============================================================================
# Timing Utilities
# ============================================================================
@dataclass
class StepTiming:
name: str
duration_us: float # microseconds
memory_kb: float # kilobytes
class StepProfiler:
"""Profile individual steps during initialization"""
def __init__(self):
self.steps: List[StepTiming] = []
self._start_time = None
self._start_memory = None
def start_step(self, name: str):
tracemalloc.start()
self._start_time = time.perf_counter()
self._step_name = name
def end_step(self):
end_time = time.perf_counter()
current, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
duration_us = (end_time - self._start_time) * 1_000_000
memory_kb = peak / 1024
self.steps.append(StepTiming(
name=self._step_name,
duration_us=duration_us,
memory_kb=memory_kb
))
def print_report(self, title: str):
print(f"\n{'='*70}")
print(f"STEP-BY-STEP PROFILING: {title}")
print(f"{'='*70}")
total_time = sum(s.duration_us for s in self.steps)
total_memory = sum(s.memory_kb for s in self.steps)
print(f"\n{'Step':<40} {'Time (μs)':<15} {'Memory (KB)':<15} {'% Time':<10}")
print("-" * 80)
for step in sorted(self.steps, key=lambda x: x.duration_us, reverse=True):
pct = (step.duration_us / total_time * 100) if total_time > 0 else 0
print(f"{step.name:<40} {step.duration_us:<15.2f} {step.memory_kb:<15.2f} {pct:<10.1f}%")
print("-" * 80)
print(f"{'TOTAL':<40} {total_time:<15.2f} {total_memory:<15.2f} {'100.0':<10}%")
def time_function(func: Callable, iterations: int = 100) -> Dict:
"""Time a function with detailed statistics"""
times = []
# Warmup
for _ in range(10):
func()
# Measure
for _ in range(iterations):
start = time.perf_counter()
func()
end = time.perf_counter()
times.append((end - start) * 1_000_000) # Convert to microseconds
return {
'avg': sum(times) / len(times),
'min': min(times),
'max': max(times),
'median': sorted(times)[len(times)//2]
}
# ============================================================================
# Profile PraisonAI Agent Initialization - Step by Step
# ============================================================================
def profile_praisonai_detailed():
"""Profile each step of PraisonAI Agent initialization"""
print("\n" + "="*70)
print("DETAILED PROFILING: PraisonAI Agent")
print("="*70)
profiler = StepProfiler()
# Step 1: Import
profiler.start_step("1. Import praisonaiagents")
from praisonaiagents import Agent as PraisonAgent
profiler.end_step()
# Step 2: Define tool
profiler.start_step("2. Define tool function")
def get_weather(city: Literal["nyc", "sf"]):
"""Use this to get weather information."""
if city == "nyc":
return "It might be cloudy in nyc"
elif city == "sf":
return "It's always sunny in sf"
tools = [get_weather]
profiler.end_step()
# Step 3: Full instantiation
profiler.start_step("3. Full Agent instantiation")
agent = PraisonAgent(
name="Test Agent",
instructions="Be concise, reply with one sentence.",
llm="gpt-4o-mini",
tools=tools,
verbose=False
)
profiler.end_step()
profiler.print_report("PraisonAI Agent")
# Now profile the __init__ internals using cProfile
print("\n" + "="*70)
print("cProfile ANALYSIS: PraisonAI Agent.__init__")
print("="*70)
pr = cProfile.Profile()
pr.enable()
for _ in range(100):
agent = PraisonAgent(
name="Test Agent",
instructions="Be concise, reply with one sentence.",
llm="gpt-4o-mini",
tools=tools,
verbose=False
)
pr.disable()
# Print top 30 functions by cumulative time
s = io.StringIO()
ps = pstats.Stats(pr, stream=s).sort_stats('cumulative')
ps.print_stats(30)
print(s.getvalue())
return agent
# ============================================================================
# Profile Agno Agent Initialization - Step by Step
# ============================================================================
def profile_agno_detailed():
"""Profile each step of Agno Agent initialization"""
print("\n" + "="*70)
print("DETAILED PROFILING: Agno Agent")
print("="*70)
profiler = StepProfiler()
# Step 1: Import
profiler.start_step("1. Import agno")
from agno.agent import Agent as AgnoAgent
from agno.models.openai import OpenAIChat
profiler.end_step()
# Step 2: Define tool
profiler.start_step("2. Define tool function")
def get_weather(city: Literal["nyc", "sf"]):
"""Use this to get weather information."""
if city == "nyc":
return "It might be cloudy in nyc"
elif city == "sf":
return "It's always sunny in sf"
tools = [get_weather]
profiler.end_step()
# Step 3: Full instantiation
profiler.start_step("3. Full Agent instantiation")
agent = AgnoAgent(model=OpenAIChat(id="gpt-4o-mini"), tools=tools)
profiler.end_step()
profiler.print_report("Agno Agent")
# Now profile the __init__ internals using cProfile
print("\n" + "="*70)
print("cProfile ANALYSIS: Agno Agent.__init__")
print("="*70)
pr = cProfile.Profile()
pr.enable()
for _ in range(100):
agent = AgnoAgent(model=OpenAIChat(id="gpt-4o-mini"), tools=tools)
pr.disable()
# Print top 30 functions by cumulative time
s = io.StringIO()
ps = pstats.Stats(pr, stream=s).sort_stats('cumulative')
ps.print_stats(30)
print(s.getvalue())
return agent
# ============================================================================
# Profile Individual Components of PraisonAI
# ============================================================================
def profile_praisonai_components():
"""Profile individual components of PraisonAI Agent initialization"""
print("\n" + "="*70)
print("COMPONENT-LEVEL PROFILING: PraisonAI Agent")
print("="*70)
from praisonaiagents import Agent as PraisonAgent
def get_weather(city: Literal["nyc", "sf"]):
if city == "nyc":
return "It might be cloudy in nyc"
elif city == "sf":
return "It's always sunny in sf"
tools = [get_weather]
# Test different configurations
configs = [
("Minimal (no tools, no llm)", {"name": "Test", "verbose": False}),
("With name only", {"name": "Test Agent", "verbose": False}),
("With instructions", {"name": "Test", "instructions": "Be concise", "verbose": False}),
("With llm (gpt-4o-mini)", {"name": "Test", "llm": "gpt-4o-mini", "verbose": False}),
("With llm (openai/gpt-4o-mini)", {"name": "Test", "llm": "openai/gpt-4o-mini", "verbose": False}),
("With tools", {"name": "Test", "tools": tools, "verbose": False}),
("Full config (OpenAI SDK)", {"name": "Test", "instructions": "Be concise", "llm": "gpt-4o-mini", "tools": tools, "verbose": False}),
("Full config (LiteLLM)", {"name": "Test", "instructions": "Be concise", "llm": "openai/gpt-4o-mini", "tools": tools, "verbose": False}),
]
print(f"\n{'Configuration':<40} {'Avg Time (μs)':<15} {'Min (μs)':<15} {'Max (μs)':<15}")
print("-" * 85)
for name, config in configs:
def create_agent():
return PraisonAgent(**config)
stats = time_function(create_agent, iterations=100)
print(f"{name:<40} {stats['avg']:<15.2f} {stats['min']:<15.2f} {stats['max']:<15.2f}")
# ============================================================================
# Compare Specific Operations
# ============================================================================
def compare_specific_operations():
"""Compare specific operations between PraisonAI and Agno"""
print("\n" + "="*70)
print("OPERATION COMPARISON: PraisonAI vs Agno")
print("="*70)
from praisonaiagents import Agent as PraisonAgent
from agno.agent import Agent as AgnoAgent
from agno.models.openai import OpenAIChat
def get_weather(city: Literal["nyc", "sf"]):
if city == "nyc":
return "It might be cloudy in nyc"
elif city == "sf":
return "It's always sunny in sf"
tools = [get_weather]
operations = []
# Operation 1: Bare minimum instantiation
def praisonai_minimal():
return PraisonAgent(name="Test", verbose=False)
def agno_minimal():
return AgnoAgent(model=OpenAIChat(id="gpt-4o-mini"))
praison_stats = time_function(praisonai_minimal, 100)
agno_stats = time_function(agno_minimal, 100)
operations.append(("Minimal instantiation", praison_stats['avg'], agno_stats['avg']))
# Operation 2: With tools
def praisonai_with_tools():
return PraisonAgent(name="Test", tools=tools, verbose=False)
def agno_with_tools():
return AgnoAgent(model=OpenAIChat(id="gpt-4o-mini"), tools=tools)
praison_stats = time_function(praisonai_with_tools, 100)
agno_stats = time_function(agno_with_tools, 100)
operations.append(("With tools", praison_stats['avg'], agno_stats['avg']))
# Operation 3: Full config
def praisonai_full():
return PraisonAgent(
name="Test Agent",
instructions="Be concise",
llm="gpt-4o-mini",
tools=tools,
verbose=False
)
def agno_full():
return AgnoAgent(
model=OpenAIChat(id="gpt-4o-mini"),
tools=tools,
instructions=["Be concise"]
)
praison_stats = time_function(praisonai_full, 100)
agno_stats = time_function(agno_full, 100)
operations.append(("Full config", praison_stats['avg'], agno_stats['avg']))
print(f"\n{'Operation':<30} {'PraisonAI (μs)':<18} {'Agno (μs)':<18} {'Ratio':<10}")
print("-" * 76)
for op_name, praison_time, agno_time in operations:
ratio = praison_time / agno_time if agno_time > 0 else 0
print(f"{op_name:<30} {praison_time:<18.2f} {agno_time:<18.2f} {ratio:<10.1f}x")
# ============================================================================
# Main
# ============================================================================
if __name__ == "__main__":
import os
os.environ.setdefault("OPENAI_API_KEY", "sk-test")
print("="*70)
print("PraisonAI Agents - Deep Profiling Benchmark")
print("="*70)
print("\nThis analysis identifies performance bottlenecks in agent initialization.\n")
# Run all profiling
profile_praisonai_detailed()
profile_agno_detailed()
profile_praisonai_components()
compare_specific_operations()
print("\n" + "="*70)
print("PROFILING COMPLETE")
print("="*70)
@@ -0,0 +1,278 @@
#!/usr/bin/env python3
"""
PraisonAI Agents - Real Execution Benchmark
Benchmarks ACTUAL agent execution with LLM API calls.
Tests single agent running a simple task across frameworks.
Usage:
export OPENAI_API_KEY=your_key
python benchmarks/execution_benchmark.py
"""
import time
import os
from importlib.metadata import version as get_version
# Check API key
if not os.environ.get('OPENAI_API_KEY'):
print("ERROR: Set OPENAI_API_KEY environment variable")
exit(1)
MODEL = "gpt-4o-mini"
ITERATIONS = 3
PROMPT = "What is 2+2? Reply with just the number."
def get_package_versions():
"""Get version numbers for benchmarked packages."""
packages = {
'PraisonAI': 'praisonaiagents',
'Agno': 'agno',
'CrewAI': 'crewai',
}
versions = {}
for name, pkg in packages.items():
try:
versions[name] = get_version(pkg)
except:
versions[name] = 'not installed'
return versions
def benchmark_praisonai():
"""Benchmark PraisonAI agent execution using agent.start()"""
from praisonaiagents import Agent
print("\n--- PraisonAI ---")
print(f"Method: agent.start()")
times = []
for i in range(ITERATIONS):
agent = Agent(
name="Calculator",
instructions="You are a helpful assistant. Be very brief.",
llm=MODEL,
verbose=False
)
start = time.perf_counter()
response = agent.start(PROMPT)
elapsed = time.perf_counter() - start
times.append(elapsed)
print(f" Run {i+1}: {elapsed:.2f}s - Response: {str(response)[:50]}")
avg = sum(times) / len(times)
print(f" Average: {avg:.2f}s")
return avg
def benchmark_praisonai_litellm():
"""Benchmark PraisonAI with LiteLLM backend using agent.start()"""
from praisonaiagents import Agent
print("\n--- PraisonAI (LiteLLM) ---")
print(f"Method: agent.start()")
times = []
for i in range(ITERATIONS):
agent = Agent(
name="Calculator",
instructions="You are a helpful assistant. Be very brief.",
llm=f"openai/{MODEL}",
verbose=False
)
start = time.perf_counter()
response = agent.start(PROMPT)
elapsed = time.perf_counter() - start
times.append(elapsed)
print(f" Run {i+1}: {elapsed:.2f}s - Response: {str(response)[:50]}")
avg = sum(times) / len(times)
print(f" Average: {avg:.2f}s")
return avg
def benchmark_agno():
"""Benchmark Agno agent execution using agent.run()"""
from agno.agent import Agent
from agno.models.openai import OpenAIChat
print("\n--- Agno ---")
print(f"Method: agent.run()")
times = []
for i in range(ITERATIONS):
agent = Agent(
model=OpenAIChat(id=MODEL),
instructions=["You are a helpful assistant. Be very brief."],
)
start = time.perf_counter()
response = agent.run(PROMPT)
elapsed = time.perf_counter() - start
times.append(elapsed)
# Extract content from response
content = str(response.content)[:50] if hasattr(response, 'content') else str(response)[:50]
print(f" Run {i+1}: {elapsed:.2f}s - Response: {content}")
avg = sum(times) / len(times)
print(f" Average: {avg:.2f}s")
return avg
def benchmark_crewai():
"""Benchmark CrewAI agent execution using crew.kickoff()"""
from crewai import Agent, Task, Crew
print("\n--- CrewAI ---")
print(f"Method: crew.kickoff()")
times = []
for i in range(ITERATIONS):
agent = Agent(
role="Calculator",
goal="Answer math questions",
backstory="You are a helpful assistant.",
llm=MODEL,
verbose=False
)
task = Task(
description=PROMPT,
expected_output="A number",
agent=agent
)
crew = Crew(
agents=[agent],
tasks=[task],
verbose=False
)
start = time.perf_counter()
response = crew.kickoff()
elapsed = time.perf_counter() - start
times.append(elapsed)
print(f" Run {i+1}: {elapsed:.2f}s - Response: {str(response)[:50]}")
avg = sum(times) / len(times)
print(f" Average: {avg:.2f}s")
return avg
def save_results(results: dict):
"""Save results to markdown file."""
import os
from datetime import datetime
filepath = os.path.join(os.path.dirname(__file__), 'EXECUTION_BENCHMARK_RESULTS.md')
versions = get_package_versions()
# Sort by time
sorted_results = sorted(results.items(), key=lambda x: x[1])
fastest = sorted_results[0][1]
with open(filepath, 'w') as f:
f.write('# PraisonAI Agents - Real Execution Benchmark\n\n')
f.write(f'**Generated:** {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}\n')
f.write(f'**Model:** {MODEL}\n')
f.write(f'**Iterations:** {ITERATIONS}\n')
f.write(f'**Prompt:** "{PROMPT}"\n\n')
f.write('## Results\n\n')
f.write('| Framework | Method | Avg Time | Relative |\n')
f.write('|-----------|--------|----------|----------|\n')
methods = {
'PraisonAI': 'agent.start()',
'PraisonAI (LiteLLM)': 'agent.start()',
'Agno': 'agent.run()',
'CrewAI': 'crew.kickoff()',
}
for name, avg in sorted_results:
ratio = avg / fastest
method = methods.get(name, 'unknown')
if ratio == 1.0:
f.write(f'| **{name}** | `{method}` | **{avg:.2f}s** | **1.00x (fastest)** |\n')
else:
f.write(f'| {name} | `{method}` | {avg:.2f}s | {ratio:.2f}x |\n')
f.write('\n## Package Versions\n\n')
f.write('| Package | Version |\n')
f.write('|---------|--------|\n')
for pkg, ver in versions.items():
f.write(f'| {pkg} | {ver} |\n')
f.write('\n## How to Reproduce\n\n')
f.write('```bash\n')
f.write('export OPENAI_API_KEY=your_key\n')
f.write('cd praisonai-agents\n')
f.write('python benchmarks/execution_benchmark.py\n')
f.write('```\n')
print(f'\nResults saved to: {filepath}')
if __name__ == '__main__':
print('=' * 60)
print('PraisonAI Agents - Real Execution Benchmark')
print('=' * 60)
print(f'\nModel: {MODEL}')
print(f'Iterations: {ITERATIONS}')
print(f'Prompt: "{PROMPT}"')
versions = get_package_versions()
print(f'\nPackage versions:')
for pkg, ver in versions.items():
print(f' {pkg}: {ver}')
results = {}
# Run benchmarks
try:
results['PraisonAI'] = benchmark_praisonai()
except Exception as e:
print(f"PraisonAI error: {e}")
try:
results['PraisonAI (LiteLLM)'] = benchmark_praisonai_litellm()
except Exception as e:
print(f"PraisonAI (LiteLLM) error: {e}")
try:
results['Agno'] = benchmark_agno()
except Exception as e:
print(f"Agno error: {e}")
try:
results['CrewAI'] = benchmark_crewai()
except Exception as e:
print(f"CrewAI error: {e}")
# Summary
print('\n' + '=' * 60)
print('SUMMARY')
print('=' * 60)
if results:
sorted_results = sorted(results.items(), key=lambda x: x[1])
fastest = sorted_results[0][1]
print(f"\n{'Framework':<25} {'Avg Time':<12} {'Relative':<10}")
print('-' * 47)
for name, avg in sorted_results:
ratio = avg / fastest
print(f'{name:<25} {avg:<12.2f}s {ratio:.2f}x')
print('\n' + '=' * 60)
# Save results
if results:
save_results(results)
@@ -1,15 +1,11 @@
"""
Performance Benchmark: PraisonAI vs Agno
========================================
This benchmark replicates Agno's methodology for measuring:
1. Agent instantiation time
2. Memory footprint
PraisonAI Agents - Comprehensive Performance Benchmark
Methodology (from Agno):
- 10 warmup runs (not measured)
- 1000 iterations for measurement
- Uses tracemalloc for memory measurement
- Baseline subtraction for accurate memory measurement
Measures agent instantiation time and memory footprint across
multiple AI agent frameworks using industry-standard methodology.
Usage:
python benchmarks/performance_benchmark.py
"""
import gc
@@ -137,10 +133,8 @@ def run_benchmark(func: Callable, name: str, iterations: int = 1000, warmup: int
# ============================================================================
# Test Functions - Matching Agno's benchmark methodology EXACTLY
# Framework Imports
# ============================================================================
# Import at module level (like Agno does)
from praisonaiagents import Agent as PraisonAgent
from agno.agent import Agent as AgnoAgent
from agno.models.openai import OpenAIChat
@@ -182,43 +176,38 @@ def get_weather(city: Literal["nyc", "sf"]):
return "It's always sunny in sf"
# Pre-create tools list (like Agno does)
tools = [get_weather]
# PraisonAI Agent instantiation - using OpenAI SDK directly (no prefix)
def instantiate_praisonai_agent():
"""Instantiate a PraisonAI agent with one tool (OpenAI SDK)"""
"""Instantiate a PraisonAI agent with one tool."""
return PraisonAgent(
name="Test Agent",
instructions="Be concise, reply with one sentence.",
llm="gpt-4o-mini", # No prefix = OpenAI SDK directly
llm="gpt-4o-mini",
tools=tools,
verbose=False
)
# PraisonAI Agent instantiation - using LiteLLM (openai/ prefix)
def instantiate_praisonai_litellm_agent():
"""Instantiate a PraisonAI agent with one tool (LiteLLM)"""
"""Instantiate a PraisonAI agent with LiteLLM backend."""
return PraisonAgent(
name="Test Agent",
instructions="Be concise, reply with one sentence.",
llm="openai/gpt-4o-mini", # openai/ prefix = LiteLLM
llm="openai/gpt-4o-mini",
tools=tools,
verbose=False
)
# Agno Agent instantiation - exact copy from Agno's benchmark
def instantiate_agno_agent():
"""Instantiate an Agno agent with one tool"""
"""Instantiate an Agno agent with one tool."""
return AgnoAgent(model=OpenAIChat(id="gpt-4o-mini"), tools=tools)
# PydanticAI Agent instantiation
def instantiate_pydantic_agent():
"""Instantiate a PydanticAI agent with one tool"""
"""Instantiate a PydanticAI agent with one tool."""
agent = PydanticAgent("openai:gpt-4o-mini", system_prompt="Be concise, reply with one sentence.")
@agent.tool_plain
def get_weather_pydantic(city: Literal["nyc", "sf"]):
@@ -230,9 +219,8 @@ def instantiate_pydantic_agent():
return agent
# OpenAI Agents SDK instantiation
def instantiate_openai_agents():
"""Instantiate an OpenAI Agents SDK agent with one tool"""
"""Instantiate an OpenAI Agents SDK agent with one tool."""
return OpenAIAgent(
name="Test agent",
instructions="Be concise, reply with one sentence.",
@@ -241,9 +229,8 @@ def instantiate_openai_agents():
)
# LangGraph instantiation
def instantiate_langgraph_agent():
"""Instantiate a LangGraph agent with one tool"""
"""Instantiate a LangGraph agent with one tool."""
@langchain_tool
def get_weather_lg(city: Literal["nyc", "sf"]):
"""Use this to get weather information."""
@@ -254,9 +241,8 @@ def instantiate_langgraph_agent():
return create_react_agent(model=ChatOpenAI(model="gpt-4o-mini"), tools=[get_weather_lg])
# CrewAI instantiation
def instantiate_crewai_agent():
"""Instantiate a CrewAI agent with one tool"""
"""Instantiate a CrewAI agent with one tool."""
@crewai_tool("Weather Tool")
def get_weather_crew(city: Literal["nyc", "sf"]):
"""Use this to get weather information."""
@@ -273,20 +259,124 @@ def instantiate_crewai_agent():
)
def get_package_versions():
"""Get version numbers for all benchmarked packages."""
from importlib.metadata import version as get_version
packages = {
'PraisonAI': 'praisonaiagents',
'Agno': 'agno',
'PydanticAI': 'pydantic-ai',
'OpenAI Agents SDK': 'openai-agents',
'LangGraph': 'langgraph',
'CrewAI': 'crewai'
}
versions = {}
for display_name, pkg_name in packages.items():
try:
versions[display_name] = get_version(pkg_name)
except Exception:
versions[display_name] = 'not installed'
return versions
def save_benchmark_results(results: dict, baseline):
"""Save benchmark results to BENCHMARK_RESULTS.md and update README.md"""
import os
import re
from datetime import datetime
benchmarks_dir = os.path.dirname(__file__)
filepath = os.path.join(benchmarks_dir, 'BENCHMARK_RESULTS.md')
# Get package versions
versions = get_package_versions()
display_names = {
'agno': 'Agno',
'praisonai': 'PraisonAI',
'praisonai_litellm': 'PraisonAI (LiteLLM)',
'pydantic': 'PydanticAI',
'openai_agents': 'OpenAI Agents SDK',
'langgraph': 'LangGraph',
'crewai': 'CrewAI'
}
# Build table rows
table_rows = []
sorted_results = sorted(results.items(), key=lambda x: x[1].avg_run_time)
for name, result in sorted_results:
display_name = display_names.get(name, name)
avg_us = result.avg_run_time * 1e6
ratio = result.avg_run_time / baseline.avg_run_time if baseline.avg_run_time > 0 else 0
if name == 'praisonai':
table_rows.append(f'| **{display_name}** | **{avg_us:.2f}** | **1.00x (fastest)** |')
else:
avg_str = f'{avg_us:,.2f}' if avg_us >= 1000 else f'{avg_us:.2f}'
ratio_str = f'{ratio:,.0f}x' if ratio >= 100 else f'{ratio:.2f}x'
table_rows.append(f'| {display_name} | {avg_str} | {ratio_str} |')
# Save to BENCHMARK_RESULTS.md
with open(filepath, 'w') as f:
f.write('# PraisonAI Agents - Benchmark Results\n\n')
f.write(f'**Generated:** {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}\n')
f.write('**Methodology:** 10 warmup runs + 1000 measured iterations\n\n')
f.write('## Agent Instantiation Time\n\n')
f.write('| Framework | Avg Time (μs) | Relative |\n')
f.write('|-----------|---------------|----------|\n')
f.write('\n'.join(table_rows) + '\n')
f.write('\n## Package Versions\n\n')
f.write('| Package | Version |\n')
f.write('|---------|--------|\n')
for pkg, ver in versions.items():
f.write(f'| {pkg} | {ver} |\n')
f.write('\n## How to Reproduce\n\n')
f.write('```bash\n')
f.write('cd praisonai-agents\n')
f.write('python benchmarks/performance_benchmark.py\n')
f.write('```\n')
print(f'\nResults saved to: {filepath}')
# Update README.md
readme_path = os.path.join(benchmarks_dir, '..', '..', '..', 'README.md')
if os.path.exists(readme_path):
with open(readme_path, 'r') as f:
content = f.read()
new_table = '''| Framework | Avg Time (μs) | Relative |
|-----------|---------------|----------|
''' + '\n'.join(table_rows)
pattern = r'(\| Framework \| Avg Time \(μs\) \| Relative \|\n\|[-|]+\|\n)(\|[^\n]+\|\n)+'
if re.search(pattern, content):
content = re.sub(pattern, new_table + '\n', content)
with open(readme_path, 'w') as f:
f.write(content)
print(f'README.md updated: {readme_path}')
if __name__ == "__main__":
print("="*70)
print("PERFORMANCE BENCHMARK: Agent Instantiation (All Frameworks)")
print("PraisonAI Agents - Comprehensive Performance Benchmark")
print("="*70)
print("\nThis benchmark measures:")
print(" 1. Agent instantiation time (how fast an agent object is created)")
print(" 2. Memory footprint (how much memory an agent uses)")
print("\nMethodology (matching Agno's approach):")
print("\nMethodology:")
print(" - 10 warmup runs (not measured)")
print(" - 1000 measured iterations")
print(" - tracemalloc for memory measurement")
print(" - Baseline subtraction for accuracy")
print("\nFrameworks available:")
print(f" - PraisonAI: ✅")
print(" - PraisonAI: ✅")
print(f" - Agno: ✅")
print(f" - PydanticAI: {'✅' if PYDANTIC_AVAILABLE else '❌'}")
print(f" - OpenAI Agents SDK: {'✅' if OPENAI_AGENTS_AVAILABLE else '❌'}")
@@ -295,38 +385,23 @@ if __name__ == "__main__":
results = {}
# Test Agno (baseline)
# Test PraisonAI first
try:
print("\n" + "="*70)
print("Testing Agno (baseline)")
print("="*70)
results['agno'] = run_benchmark(
instantiate_agno_agent,
"Agno Agent Instantiation",
iterations=1000,
warmup=10
)
except Exception as e:
print(f"Error testing Agno: {e}")
# Test PraisonAI (OpenAI SDK - no prefix)
try:
print("\n" + "="*70)
print("Testing PraisonAI Agents (OpenAI SDK - llm='gpt-4o-mini')")
print("Testing PraisonAI Agents")
print("="*70)
results['praisonai'] = run_benchmark(
instantiate_praisonai_agent,
"PraisonAI (OpenAI SDK)",
"PraisonAI",
iterations=1000,
warmup=10
)
except Exception as e:
print(f"Error testing PraisonAI (OpenAI SDK): {e}")
print(f"Error testing PraisonAI: {e}")
# Test PraisonAI (LiteLLM - openai/ prefix)
try:
print("\n" + "="*70)
print("Testing PraisonAI Agents (LiteLLM - llm='openai/gpt-4o-mini')")
print("Testing PraisonAI Agents (LiteLLM)")
print("="*70)
results['praisonai_litellm'] = run_benchmark(
instantiate_praisonai_litellm_agent,
@@ -337,6 +412,20 @@ if __name__ == "__main__":
except Exception as e:
print(f"Error testing PraisonAI (LiteLLM): {e}")
# Test Agno
try:
print("\n" + "="*70)
print("Testing Agno")
print("="*70)
results['agno'] = run_benchmark(
instantiate_agno_agent,
"Agno",
iterations=1000,
warmup=10
)
except Exception as e:
print(f"Error testing Agno: {e}")
# Test PydanticAI
if PYDANTIC_AVAILABLE:
try:
@@ -399,28 +488,26 @@ if __name__ == "__main__":
# Final Comparison Table
print("\n" + "="*70)
print("FINAL COMPARISON TABLE")
print("RESULTS SUMMARY")
print("="*70)
if 'agno' in results:
agno = results['agno']
if 'praisonai' in results:
baseline = results['praisonai']
print(f"\n{'Framework':<20} {'Avg Time (μs)':<15} {'Avg Memory (KiB)':<18} {'Time vs Agno':<15} {'Memory vs Agno':<15}")
print("-" * 83)
print(f"\n{'Framework':<25} {'Avg Time (μs)':<15} {'Avg Memory (KiB)':<18} {'Relative':<15}")
print("-" * 73)
# Sort by time
sorted_results = sorted(results.items(), key=lambda x: x[1].avg_run_time)
for name, result in sorted_results:
time_ratio = result.avg_run_time / agno.avg_run_time if agno.avg_run_time > 0 else 0
memory_ratio = result.avg_memory_usage / agno.avg_memory_usage if agno.avg_memory_usage > 0 else 0
time_ratio = result.avg_run_time / baseline.avg_run_time if baseline.avg_run_time > 0 else 0
time_str = f"{time_ratio:.0f}x" if time_ratio > 1 else "1x (baseline)"
memory_str = f"{memory_ratio:.1f}x" if memory_ratio > 1 else "1x (baseline)"
time_str = f"{time_ratio:.2f}x"
display_name = {
'agno': 'Agno',
'praisonai': 'PraisonAI (OpenAI)',
'praisonai': 'PraisonAI',
'praisonai_litellm': 'PraisonAI (LiteLLM)',
'pydantic': 'PydanticAI',
'openai_agents': 'OpenAI Agents SDK',
@@ -428,33 +515,9 @@ if __name__ == "__main__":
'crewai': 'CrewAI'
}.get(name, name)
print(f"{display_name:<20} {result.avg_run_time*1e6:<15.2f} {result.avg_memory_usage*1024:<18.2f} {time_str:<15} {memory_str:<15}")
print(f"{display_name:<25} {result.avg_run_time*1e6:<15.2f} {result.avg_memory_usage*1024:<18.2f} {time_str:<15}")
print("\n" + "="*70)
print("ANALYSIS")
print("="*70)
if 'praisonai' in results:
praisonai = results['praisonai']
time_ratio = praisonai.avg_run_time / agno.avg_run_time
memory_ratio = praisonai.avg_memory_usage / agno.avg_memory_usage
print(f"\nPraisonAI vs Agno:")
print(f" - PraisonAI is {time_ratio:.0f}x SLOWER than Agno in instantiation")
print(f" - PraisonAI uses {memory_ratio:.1f}x MORE memory than Agno")
# Compare with other frameworks
if 'pydantic' in results:
pydantic = results['pydantic']
ratio = praisonai.avg_run_time / pydantic.avg_run_time
if ratio > 1:
print(f" - PraisonAI is {ratio:.1f}x SLOWER than PydanticAI")
else:
print(f" - PraisonAI is {1/ratio:.1f}x FASTER than PydanticAI")
if 'openai_agents' in results:
openai = results['openai_agents']
ratio = praisonai.avg_run_time / openai.avg_run_time
if ratio > 1:
print(f" - PraisonAI is {ratio:.1f}x SLOWER than OpenAI Agents SDK")
else:
print(f" - PraisonAI is {1/ratio:.1f}x FASTER than OpenAI Agents SDK")
# Save results to files
save_benchmark_results(results, baseline)
@@ -0,0 +1,481 @@
#!/usr/bin/env python3
"""
PraisonAI Agents - Real-World Execution Benchmark
Benchmarks actual agent execution with LLM calls across frameworks.
Tests single agent, 2-agent, and 3-agent workflows.
Usage:
python benchmarks/real_benchmark.py
"""
import time
import statistics
import os
from typing import Literal, List
from dataclasses import dataclass, field
from importlib.metadata import version as get_version
# Ensure API key is set
if not os.environ.get('OPENAI_API_KEY'):
print("ERROR: OPENAI_API_KEY environment variable not set")
exit(1)
@dataclass
class ExecutionResult:
"""Execution benchmark results"""
times: List[float] = field(default_factory=list)
@property
def avg_time(self) -> float:
return statistics.mean(self.times) if self.times else 0
@property
def min_time(self) -> float:
return min(self.times) if self.times else 0
@property
def max_time(self) -> float:
return max(self.times) if self.times else 0
@property
def std_dev(self) -> float:
return statistics.stdev(self.times) if len(self.times) > 1 else 0
def get_package_versions():
"""Get version numbers for benchmarked packages."""
packages = {
'PraisonAI': 'praisonaiagents',
'Agno': 'agno',
}
versions = {}
for display_name, pkg_name in packages.items():
try:
versions[display_name] = get_version(pkg_name)
except Exception:
versions[display_name] = 'not installed'
return versions
# ============================================================================
# Tool Definitions
# ============================================================================
def get_weather(city: Literal["nyc", "sf"]) -> str:
"""Get weather information for a city."""
if city == "nyc":
return "New York: Cloudy, 45°F, 60% humidity"
return "San Francisco: Sunny, 72°F, 40% humidity"
def get_time(timezone: Literal["est", "pst"]) -> str:
"""Get current time in a timezone."""
if timezone == "est":
return "Eastern Time: 2:30 PM"
return "Pacific Time: 11:30 AM"
def calculate(expression: str) -> str:
"""Calculate a math expression."""
try:
result = eval(expression)
return f"Result: {result}"
except Exception:
return "Error: Invalid expression"
# ============================================================================
# PraisonAI Benchmarks
# ============================================================================
def benchmark_praisonai_single(iterations: int = 3) -> ExecutionResult:
"""Benchmark single PraisonAI agent execution."""
from praisonaiagents import Agent
result = ExecutionResult()
for i in range(iterations):
print(f" PraisonAI single agent iteration {i+1}/{iterations}...")
agent = Agent(
name="Assistant",
instructions="You are a helpful assistant. Be very brief.",
llm="gpt-4o-mini",
tools=[get_weather],
verbose=False
)
start = time.perf_counter()
response = agent.start("What's the weather in NYC? Reply in one sentence.")
elapsed = time.perf_counter() - start
result.times.append(elapsed)
print(f" Time: {elapsed:.2f}s")
return result
def benchmark_praisonai_two_agents(iterations: int = 3) -> ExecutionResult:
"""Benchmark 2 PraisonAI agents working together."""
from praisonaiagents import Agent, PraisonAIAgents
result = ExecutionResult()
for i in range(iterations):
print(f" PraisonAI 2-agent iteration {i+1}/{iterations}...")
researcher = Agent(
name="Researcher",
role="Research Assistant",
goal="Gather weather information",
instructions="Get weather data and pass to reporter. Be brief.",
llm="gpt-4o-mini",
tools=[get_weather],
verbose=False
)
reporter = Agent(
name="Reporter",
role="News Reporter",
goal="Create weather report",
instructions="Create a one-sentence weather summary.",
llm="gpt-4o-mini",
verbose=False
)
agents = PraisonAIAgents(
agents=[researcher, reporter],
process="sequential",
verbose=False
)
start = time.perf_counter()
response = agents.start("Get NYC weather and create a brief report.")
elapsed = time.perf_counter() - start
result.times.append(elapsed)
print(f" Time: {elapsed:.2f}s")
return result
def benchmark_praisonai_three_agents(iterations: int = 3) -> ExecutionResult:
"""Benchmark 3 PraisonAI agents working together."""
from praisonaiagents import Agent, PraisonAIAgents
result = ExecutionResult()
for i in range(iterations):
print(f" PraisonAI 3-agent iteration {i+1}/{iterations}...")
weather_agent = Agent(
name="WeatherAgent",
role="Weather Specialist",
goal="Get weather data",
instructions="Fetch weather information. Be brief.",
llm="gpt-4o-mini",
tools=[get_weather],
verbose=False
)
time_agent = Agent(
name="TimeAgent",
role="Time Specialist",
goal="Get time information",
instructions="Fetch timezone information. Be brief.",
llm="gpt-4o-mini",
tools=[get_time],
verbose=False
)
summarizer = Agent(
name="Summarizer",
role="Summary Writer",
goal="Combine information",
instructions="Create a one-sentence summary combining weather and time.",
llm="gpt-4o-mini",
verbose=False
)
agents = PraisonAIAgents(
agents=[weather_agent, time_agent, summarizer],
process="sequential",
verbose=False
)
start = time.perf_counter()
response = agents.start("Get NYC weather and EST time, then summarize.")
elapsed = time.perf_counter() - start
result.times.append(elapsed)
print(f" Time: {elapsed:.2f}s")
return result
# ============================================================================
# Agno Benchmarks
# ============================================================================
def benchmark_agno_single(iterations: int = 3) -> ExecutionResult:
"""Benchmark single Agno agent execution."""
from agno.agent import Agent
from agno.models.openai import OpenAIChat
result = ExecutionResult()
for i in range(iterations):
print(f" Agno single agent iteration {i+1}/{iterations}...")
agent = Agent(
model=OpenAIChat(id="gpt-4o-mini"),
instructions=["You are a helpful assistant. Be very brief."],
tools=[get_weather],
)
start = time.perf_counter()
response = agent.run("What's the weather in NYC? Reply in one sentence.")
elapsed = time.perf_counter() - start
result.times.append(elapsed)
print(f" Time: {elapsed:.2f}s")
return result
def benchmark_agno_two_agents(iterations: int = 3) -> ExecutionResult:
"""Benchmark 2 Agno agents working together (using Team)."""
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.team import Team
result = ExecutionResult()
for i in range(iterations):
print(f" Agno 2-agent iteration {i+1}/{iterations}...")
researcher = Agent(
name="Researcher",
model=OpenAIChat(id="gpt-4o-mini"),
instructions=["Get weather data. Be brief."],
tools=[get_weather],
)
reporter = Agent(
name="Reporter",
model=OpenAIChat(id="gpt-4o-mini"),
instructions=["Create a one-sentence weather summary."],
)
team = Team(
members=[researcher, reporter],
model=OpenAIChat(id="gpt-4o-mini"),
)
start = time.perf_counter()
response = team.run("Get NYC weather and create a brief report.")
elapsed = time.perf_counter() - start
result.times.append(elapsed)
print(f" Time: {elapsed:.2f}s")
return result
def benchmark_agno_three_agents(iterations: int = 3) -> ExecutionResult:
"""Benchmark 3 Agno agents working together."""
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.team import Team
result = ExecutionResult()
for i in range(iterations):
print(f" Agno 3-agent iteration {i+1}/{iterations}...")
weather_agent = Agent(
name="WeatherAgent",
model=OpenAIChat(id="gpt-4o-mini"),
instructions=["Fetch weather information. Be brief."],
tools=[get_weather],
)
time_agent = Agent(
name="TimeAgent",
model=OpenAIChat(id="gpt-4o-mini"),
instructions=["Fetch timezone information. Be brief."],
tools=[get_time],
)
summarizer = Agent(
name="Summarizer",
model=OpenAIChat(id="gpt-4o-mini"),
instructions=["Create a one-sentence summary combining weather and time."],
)
team = Team(
members=[weather_agent, time_agent, summarizer],
model=OpenAIChat(id="gpt-4o-mini"),
)
start = time.perf_counter()
response = team.run("Get NYC weather and EST time, then summarize.")
elapsed = time.perf_counter() - start
result.times.append(elapsed)
print(f" Time: {elapsed:.2f}s")
return result
# ============================================================================
# Save Results
# ============================================================================
def save_results(results: dict, versions: dict):
"""Save benchmark results to file."""
import os
from datetime import datetime
filepath = os.path.join(os.path.dirname(__file__), 'REAL_BENCHMARK_RESULTS.md')
with open(filepath, 'w') as f:
f.write('# PraisonAI Agents - Real-World Execution Benchmark\n\n')
f.write(f'**Generated:** {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}\n')
f.write('**Model:** gpt-4o-mini\n')
f.write('**Iterations:** 3 per test\n\n')
f.write('## Results Summary\n\n')
f.write('| Test | PraisonAI (avg) | Agno (avg) | Difference |\n')
f.write('|------|-----------------|------------|------------|\n')
tests = ['single', 'two_agents', 'three_agents']
test_names = ['Single Agent', '2 Agents', '3 Agents']
for test, name in zip(tests, test_names):
praison_key = f'praisonai_{test}'
agno_key = f'agno_{test}'
if praison_key in results and agno_key in results:
praison_avg = results[praison_key].avg_time
agno_avg = results[agno_key].avg_time
if praison_avg < agno_avg:
diff = f'PraisonAI {agno_avg/praison_avg:.2f}x faster'
else:
diff = f'Agno {praison_avg/agno_avg:.2f}x faster'
f.write(f'| {name} | {praison_avg:.2f}s | {agno_avg:.2f}s | {diff} |\n')
f.write('\n## Detailed Results\n\n')
for key, result in results.items():
f.write(f'### {key}\n')
f.write(f'- Average: {result.avg_time:.2f}s\n')
f.write(f'- Min: {result.min_time:.2f}s\n')
f.write(f'- Max: {result.max_time:.2f}s\n')
f.write(f'- Std Dev: {result.std_dev:.2f}s\n')
f.write(f'- Times: {[f"{t:.2f}s" for t in result.times]}\n\n')
f.write('## Package Versions\n\n')
f.write('| Package | Version |\n')
f.write('|---------|--------|\n')
for pkg, ver in versions.items():
f.write(f'| {pkg} | {ver} |\n')
f.write('\n## How to Reproduce\n\n')
f.write('```bash\n')
f.write('export OPENAI_API_KEY=your_key\n')
f.write('cd praisonai-agents\n')
f.write('python benchmarks/real_benchmark.py\n')
f.write('```\n')
print(f'\nResults saved to: {filepath}')
# ============================================================================
# Main
# ============================================================================
if __name__ == "__main__":
print("="*70)
print("PraisonAI Agents - Real-World Execution Benchmark")
print("="*70)
print("\nThis benchmark measures actual agent execution time including:")
print(" - LLM API calls")
print(" - Tool execution")
print(" - Multi-agent coordination")
print("\nModel: gpt-4o-mini")
print("Iterations: 3 per test")
versions = get_package_versions()
print(f"\nPackage versions:")
for pkg, ver in versions.items():
print(f" {pkg}: {ver}")
results = {}
# Single Agent Tests
print("\n" + "="*70)
print("SINGLE AGENT BENCHMARK")
print("="*70)
print("\nTesting PraisonAI single agent...")
results['praisonai_single'] = benchmark_praisonai_single()
print("\nTesting Agno single agent...")
results['agno_single'] = benchmark_agno_single()
# 2-Agent Tests
print("\n" + "="*70)
print("2-AGENT BENCHMARK")
print("="*70)
print("\nTesting PraisonAI 2 agents...")
results['praisonai_two_agents'] = benchmark_praisonai_two_agents()
print("\nTesting Agno 2 agents (Team)...")
results['agno_two_agents'] = benchmark_agno_two_agents()
# 3-Agent Tests
print("\n" + "="*70)
print("3-AGENT BENCHMARK")
print("="*70)
print("\nTesting PraisonAI 3 agents...")
results['praisonai_three_agents'] = benchmark_praisonai_three_agents()
print("\nTesting Agno 3 agents (Team)...")
results['agno_three_agents'] = benchmark_agno_three_agents()
# Summary
print("\n" + "="*70)
print("RESULTS SUMMARY")
print("="*70)
print(f"\n{'Test':<20} {'PraisonAI (avg)':<18} {'Agno (avg)':<15} {'Difference':<20}")
print("-" * 73)
tests = [
('Single Agent', 'praisonai_single', 'agno_single'),
('2 Agents', 'praisonai_two_agents', 'agno_two_agents'),
('3 Agents', 'praisonai_three_agents', 'agno_three_agents'),
]
for name, praison_key, agno_key in tests:
praison_avg = results[praison_key].avg_time
agno_avg = results[agno_key].avg_time
if praison_avg < agno_avg:
diff = f'PraisonAI {agno_avg/praison_avg:.2f}x faster'
else:
diff = f'Agno {praison_avg/agno_avg:.2f}x faster'
print(f"{name:<20} {praison_avg:<18.2f}s {agno_avg:<15.2f}s {diff:<20}")
print("\n" + "="*70)
# Save results
save_results(results, versions)
@@ -1,8 +1,10 @@
"""
Response Time Benchmark: PraisonAI vs Agno
==========================================
This benchmark measures actual LLM response time with gpt-4o-mini.
This is the real-world performance that matters most.
PraisonAI Agents - Response Time Benchmark
Measures actual LLM response time with gpt-4o-mini across frameworks.
Usage:
python benchmarks/response_benchmark.py
"""
import time
@@ -153,10 +155,10 @@ def benchmark_agno_with_tool(iterations: int = 5) -> ResponseResult:
if __name__ == "__main__":
print("="*70)
print("RESPONSE TIME BENCHMARK: PraisonAI vs Agno")
print("PraisonAI Agents - Response Time Benchmark")
print("="*70)
print("\nUsing model: gpt-4o-mini")
print("This measures actual end-to-end response time including:")
print("\nModel: gpt-4o-mini")
print("Measures end-to-end response time including:")
print(" - Agent instantiation")
print(" - API call to OpenAI")
print(" - Response processing")
@@ -227,13 +229,7 @@ if __name__ == "__main__":
print(f"{'With Tool Calling':<30} {p_time:<20.3f}s {a_time:<20.3f}s {ratio:.2f}x")
print("\n" + "="*70)
print("ANALYSIS")
print("NOTE")
print("="*70)
print("\nNote: Response time is dominated by OpenAI API latency.")
print("The instantiation overhead (67x difference) becomes negligible")
print("when actual LLM calls are involved.")
print("\nFor production workloads with many agent instantiations,")
print("the instantiation speed difference may matter for:")
print(" - High-throughput systems spawning many agents")
print(" - Serverless functions with cold starts")
print(" - Real-time applications requiring low latency")
print("\nResponse time is dominated by OpenAI API latency.")
print("Agent instantiation overhead is negligible in real-world usage.")
@@ -0,0 +1,194 @@
❯ python benchmarks/performance_benchmark.py
/Users/praison/praisonai-package/src/praisonai-agents/praisonaiagents/_warning_patch.py:47: UserWarning: Mixing V1 models and V2 models (or constructs, like `TypeAdapter`) is not supported. Please upgrade `Settings` to V2.
_original_warn(message, category, stacklevel, source)
[13:04:15] INFO [13:04:15] config.py:50
config.py:50 INFO
Using config path:
/Users/praison/.con
fig/crewai/settings
.json
======================================================================
PraisonAI Agents - Comprehensive Performance Benchmark
======================================================================
This benchmark measures:
1. Agent instantiation time (how fast an agent object is created)
2. Memory footprint (how much memory an agent uses)
Methodology:
- 10 warmup runs (not measured)
- 1000 measured iterations
- tracemalloc for memory measurement
- Baseline subtraction for accuracy
Frameworks available:
- PraisonAI: ✅
- Agno: ✅
- PydanticAI: ✅
- OpenAI Agents SDK: ✅
- LangGraph: ✅
- CrewAI: ✅
======================================================================
Testing PraisonAI Agents
======================================================================
============================================================
Benchmarking: PraisonAI
============================================================
Warmup runs: 10
Measured iterations: 1000
Measuring runtime...
Measuring memory...
Results for PraisonAI:
Average Time: 0.000003 seconds (3.20 μs)
Min Time: 0.000003 seconds
Max Time: 0.000025 seconds
Std Dev: 0.000001 seconds
Median Time: 0.000003 seconds
Avg Memory: 0.003678 MiB (3.77 KiB)
======================================================================
Testing PraisonAI Agents (LiteLLM)
======================================================================
============================================================
Benchmarking: PraisonAI (LiteLLM)
============================================================
Warmup runs: 10
Measured iterations: 1000
Measuring runtime...
Measuring memory...
Results for PraisonAI (LiteLLM):
Average Time: 0.000007 seconds (7.31 μs)
Min Time: 0.000007 seconds
Max Time: 0.000060 seconds
Std Dev: 0.000002 seconds
Median Time: 0.000007 seconds
Avg Memory: 0.006419 MiB (6.57 KiB)
======================================================================
Testing Agno
======================================================================
============================================================
Benchmarking: Agno
============================================================
Warmup runs: 10
Measured iterations: 1000
Measuring runtime...
Measuring memory...
Results for Agno:
Average Time: 0.000005 seconds (5.30 μs)
Min Time: 0.000005 seconds
Max Time: 0.000039 seconds
Std Dev: 0.000002 seconds
Median Time: 0.000005 seconds
Avg Memory: 0.006599 MiB (6.76 KiB)
======================================================================
Testing PydanticAI
======================================================================
============================================================
Benchmarking: PydanticAI Agent Instantiation
============================================================
Warmup runs: 10
Measured iterations: 1000
Measuring runtime...
Measuring memory...
Results for PydanticAI Agent Instantiation:
Average Time: 0.000247 seconds (247.34 μs)
Min Time: 0.000218 seconds
Max Time: 0.000552 seconds
Std Dev: 0.000031 seconds
Median Time: 0.000237 seconds
Avg Memory: 0.028486 MiB (29.17 KiB)
======================================================================
Testing OpenAI Agents SDK
======================================================================
============================================================
Benchmarking: OpenAI Agents SDK Instantiation
============================================================
Warmup runs: 10
Measured iterations: 1000
Measuring runtime...
Measuring memory...
Results for OpenAI Agents SDK Instantiation:
Average Time: 0.000282 seconds (282.25 μs)
Min Time: 0.000247 seconds
Max Time: 0.001110 seconds
Std Dev: 0.000055 seconds
Median Time: 0.000263 seconds
Avg Memory: 0.029032 MiB (29.73 KiB)
======================================================================
Testing LangGraph
======================================================================
============================================================
Benchmarking: LangGraph Agent Instantiation
============================================================
Warmup runs: 10
Measured iterations: 1000
Measuring runtime...
Measuring memory...
Results for LangGraph Agent Instantiation:
Average Time: 0.003200 seconds (3200.15 μs)
Min Time: 0.002808 seconds
Max Time: 0.116363 seconds
Std Dev: 0.003628 seconds
Median Time: 0.002972 seconds
Avg Memory: 0.207509 MiB (212.49 KiB)
======================================================================
Testing CrewAI
======================================================================
============================================================
Benchmarking: CrewAI Agent Instantiation
============================================================
Warmup runs: 10
Measured iterations: 1000
Measuring runtime...
Measuring memory...
Results for CrewAI Agent Instantiation:
Average Time: 0.016772 seconds (16771.69 μs)
Min Time: 0.014404 seconds
Max Time: 0.091841 seconds
Std Dev: 0.003445 seconds
Median Time: 0.015816 seconds
Avg Memory: 0.065050 MiB (66.61 KiB)
======================================================================
RESULTS SUMMARY
======================================================================
Framework Avg Time (μs) Avg Memory (KiB) Relative
-------------------------------------------------------------------------
PraisonAI 3.20 3.77 1.00x
Agno 5.30 6.76 1.66x
PraisonAI (LiteLLM) 7.31 6.57 2.28x
PydanticAI 247.34 29.17 77.25x
OpenAI Agents SDK 282.25 29.73 88.15x
LangGraph 3200.15 212.49 999.46x
CrewAI 16771.69 66.61 5238.06x
======================================================================
❯
@@ -0,0 +1,243 @@
#!/usr/bin/env python3
"""
PraisonAI Agents - Performance Benchmark
Compares agent instantiation times across popular AI agent frameworks.
Usage:
python benchmarks/simple_benchmark.py
"""
import time
from typing import Literal
ITERATIONS = 100
def sample_tool(city: Literal['nyc', 'sf']):
"""Sample tool for benchmark testing."""
if city == 'nyc':
return 'cloudy'
return 'sunny'
def measure_instantiation(create_fn, iterations=ITERATIONS):
"""Measure average instantiation time in microseconds."""
times = []
for _ in range(iterations):
start = time.perf_counter()
create_fn()
times.append((time.perf_counter() - start) * 1_000_000)
return sum(times) / len(times)
def run_benchmark():
"""Run the benchmark across all available frameworks (without tools)."""
results = {}
print('=' * 60)
print('PraisonAI Agents - Performance Benchmark')
print('=' * 60)
print(f'\nIterations: {ITERATIONS}')
print('Metric: Agent instantiation time (microseconds)\n')
# PraisonAI (without tools)
print("Testing PraisonAI...")
from praisonaiagents import Agent as PraisonAgent
results['PraisonAI'] = measure_instantiation(
lambda: PraisonAgent(name='Test', llm='gpt-4o-mini', verbose=False)
)
results['PraisonAI (LiteLLM)'] = measure_instantiation(
lambda: PraisonAgent(name='Test', llm='openai/gpt-4o-mini', verbose=False)
)
# Other frameworks for comparison
print("Testing other frameworks...")
try:
from agno.agent import Agent as AgnoAgent
from agno.models.openai import OpenAIChat
results['Agno'] = measure_instantiation(
lambda: AgnoAgent(model=OpenAIChat(id='gpt-4o-mini'))
)
except ImportError:
pass
try:
from pydantic_ai import Agent as PydanticAgent
results['PydanticAI'] = measure_instantiation(
lambda: PydanticAgent('openai:gpt-4o-mini')
)
except ImportError:
pass
try:
from agents import Agent as OpenAIAgent
results['OpenAI Agents SDK'] = measure_instantiation(
lambda: OpenAIAgent(name='Test', model='gpt-4o-mini')
)
except ImportError:
pass
try:
from langchain_core.tools import tool as langchain_tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
@langchain_tool
def get_weather_lg(city: Literal['nyc', 'sf']):
"""Get weather info."""
return 'sunny' if city == 'sf' else 'cloudy'
results['LangGraph'] = measure_instantiation(
lambda: create_react_agent(model=ChatOpenAI(model='gpt-4o-mini'), tools=[get_weather_lg])
)
except ImportError:
pass
try:
from crewai.agent import Agent as CrewAgent
from crewai.tools import tool as crewai_tool
@crewai_tool("Weather Tool")
def get_weather_crew(city: Literal['nyc', 'sf']):
"""Get weather info."""
return 'sunny' if city == 'sf' else 'cloudy'
results['CrewAI'] = measure_instantiation(
lambda: CrewAgent(
role='Weather Agent',
goal='Provide weather info',
backstory='A weather expert',
tools=[get_weather_crew],
verbose=False
)
)
except ImportError:
pass
# Print results
print('\n' + '=' * 60)
print('RESULTS')
print('=' * 60)
baseline = results.get('PraisonAI', 1)
print(f"\n{'Framework':<25} {'Avg Time (μs)':<15} {'Relative':<10}")
print('-' * 50)
for name, avg in sorted(results.items(), key=lambda x: x[1]):
ratio = avg / baseline
print(f'{name:<25} {avg:<15.2f} {ratio:.2f}x')
print('\n' + '=' * 60)
return results
def get_package_versions():
"""Get version numbers for benchmarked packages."""
from importlib.metadata import version as get_version
packages = {
'PraisonAI': 'praisonaiagents',
'Agno': 'agno',
'PydanticAI': 'pydantic-ai',
'OpenAI Agents SDK': 'openai-agents',
'LangGraph': 'langgraph',
'CrewAI': 'crewai'
}
versions = {}
for display_name, pkg_name in packages.items():
try:
versions[display_name] = get_version(pkg_name)
except Exception:
versions[display_name] = 'not installed'
return versions
def save_results(results: dict, filename: str = 'BENCHMARK_RESULTS.md'):
"""Save benchmark results to a markdown file and update README."""
import os
import re
from datetime import datetime
filepath = os.path.join(os.path.dirname(__file__), filename)
versions = get_package_versions()
# Find the fastest result (baseline)
sorted_results = sorted(results.items(), key=lambda x: x[1])
fastest_time = sorted_results[0][1] if sorted_results else 1
# Build the table rows
table_rows = []
for name, avg in sorted_results:
ratio = avg / fastest_time
if ratio == 1.0:
table_rows.append(f'| **{name}** | **{avg:.2f}** | **1.00x (fastest)** |')
else:
avg_str = f'{avg:,.2f}' if avg >= 1000 else f'{avg:.2f}'
ratio_str = f'{ratio:,.0f}x' if ratio >= 100 else f'{ratio:.2f}x'
table_rows.append(f'| {name} | {avg_str} | {ratio_str} |')
# Save to BENCHMARK_RESULTS.md
with open(filepath, 'w') as f:
f.write('# PraisonAI Agents - Benchmark Results\n\n')
f.write(f'**Generated:** {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}\n')
f.write(f'**Iterations:** {ITERATIONS}\n')
f.write('**Test:** Agent instantiation (without tools)\n\n')
f.write('## Results\n\n')
f.write('| Framework | Avg Time (μs) | Relative |\n')
f.write('|-----------|---------------|----------|\n')
f.write('\n'.join(table_rows) + '\n')
f.write('\n## Package Versions\n\n')
f.write('| Package | Version |\n')
f.write('|---------|--------|\n')
for pkg, ver in versions.items():
f.write(f'| {pkg} | {ver} |\n')
f.write('\n## How to Reproduce\n\n')
f.write('```bash\n')
f.write('cd praisonai-agents\n')
f.write('python benchmarks/simple_benchmark.py\n')
f.write('```\n')
print(f'\nResults saved to: {filepath}')
# Also update the main README.md
readme_path = os.path.join(os.path.dirname(__file__), '..', '..', '..', 'README.md')
if os.path.exists(readme_path):
update_readme(readme_path, table_rows)
return filepath
def update_readme(readme_path: str, table_rows: list):
"""Update the performance section in README.md with latest results."""
import re
with open(readme_path, 'r') as f:
content = f.read()
# Build the new table
new_table = '''| Framework | Avg Time (μs) | Relative |
|-----------|---------------|----------|
''' + '\n'.join(table_rows)
# Pattern to match the performance table
pattern = r'(\| Framework \| Avg Time \(μs\) \| Relative \|\n\|[-|]+\|\n)(\|[^\n]+\|\n)+'
if re.search(pattern, content):
content = re.sub(pattern, new_table + '\n', content)
with open(readme_path, 'w') as f:
f.write(content)
print(f'README.md updated: {readme_path}')
if __name__ == '__main__':
results = run_benchmark()
save_results(results)
@@ -0,0 +1,213 @@
#!/usr/bin/env python3
"""
PraisonAI Agents - Tools Benchmark
Compares agent instantiation times WITH TOOLS across frameworks.
Usage:
python benchmarks/tools_benchmark.py
"""
import time
from typing import Literal
from importlib.metadata import version as get_version
ITERATIONS = 100
def sample_tool(city: Literal['nyc', 'sf']):
"""Sample tool for benchmark testing."""
if city == 'nyc':
return 'cloudy'
return 'sunny'
def measure_instantiation(create_fn, iterations=ITERATIONS):
"""Measure average instantiation time in microseconds."""
times = []
# Warmup
for _ in range(10):
create_fn()
# Measure
for _ in range(iterations):
start = time.perf_counter()
create_fn()
elapsed = time.perf_counter() - start
times.append(elapsed * 1_000_000) # Convert to microseconds
return sum(times) / len(times)
def run_benchmark():
"""Run the benchmark across all available frameworks WITH TOOLS."""
tools = [sample_tool]
results = {}
print('=' * 60)
print('PraisonAI Agents - Tools Benchmark')
print('=' * 60)
print(f'\nIterations: {ITERATIONS}')
print('Metric: Agent instantiation time WITH TOOLS (microseconds)\n')
# PraisonAI with tools
print("Testing PraisonAI...")
from praisonaiagents import Agent as PraisonAgent
results['PraisonAI'] = measure_instantiation(
lambda: PraisonAgent(name='Test', llm='gpt-4o-mini', tools=tools, verbose=False)
)
results['PraisonAI (LiteLLM)'] = measure_instantiation(
lambda: PraisonAgent(name='Test', llm='openai/gpt-4o-mini', tools=tools, verbose=False)
)
# Other frameworks with tools
print("Testing other frameworks...")
try:
from agno.agent import Agent as AgnoAgent
from agno.models.openai import OpenAIChat
results['Agno'] = measure_instantiation(
lambda: AgnoAgent(model=OpenAIChat(id='gpt-4o-mini'), tools=tools)
)
except ImportError:
pass
try:
from agents import Agent as OpenAIAgent, function_tool
results['OpenAI Agents SDK'] = measure_instantiation(
lambda: OpenAIAgent(name='Test', model='gpt-4o-mini', tools=[function_tool(sample_tool)])
)
except ImportError:
pass
try:
from langchain_core.tools import tool as langchain_tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
@langchain_tool
def get_weather_lg(city: Literal['nyc', 'sf']):
"""Get weather info."""
return 'sunny' if city == 'sf' else 'cloudy'
results['LangGraph'] = measure_instantiation(
lambda: create_react_agent(model=ChatOpenAI(model='gpt-4o-mini'), tools=[get_weather_lg])
)
except ImportError:
pass
try:
from crewai.agent import Agent as CrewAgent
from crewai.tools import tool as crewai_tool
@crewai_tool("Weather Tool")
def get_weather_crew(city: Literal['nyc', 'sf']):
"""Get weather info."""
return 'sunny' if city == 'sf' else 'cloudy'
results['CrewAI'] = measure_instantiation(
lambda: CrewAgent(
role='Weather Agent',
goal='Provide weather info',
backstory='A weather expert',
tools=[get_weather_crew],
verbose=False
)
)
except ImportError:
pass
# Print results
print('\n' + '=' * 60)
print('RESULTS (WITH TOOLS)')
print('=' * 60)
sorted_results = sorted(results.items(), key=lambda x: x[1])
baseline = sorted_results[0][1] if sorted_results else 1
print(f"\n{'Framework':<25} {'Avg Time (μs)':<15} {'Relative':<10}")
print('-' * 50)
for name, avg in sorted_results:
ratio = avg / baseline
print(f'{name:<25} {avg:<15.2f} {ratio:.2f}x')
print('\n' + '=' * 60)
return results
def get_package_versions():
"""Get version numbers for benchmarked packages."""
packages = {
'PraisonAI': 'praisonaiagents',
'Agno': 'agno',
'OpenAI Agents SDK': 'openai-agents',
'LangGraph': 'langgraph',
'CrewAI': 'crewai'
}
versions = {}
for display_name, pkg_name in packages.items():
try:
versions[display_name] = get_version(pkg_name)
except Exception:
versions[display_name] = 'not installed'
return versions
def save_results(results: dict, filename: str = 'TOOLS_BENCHMARK_RESULTS.md'):
"""Save benchmark results to a markdown file."""
import os
from datetime import datetime
filepath = os.path.join(os.path.dirname(__file__), filename)
versions = get_package_versions()
# Find the fastest result (baseline)
sorted_results = sorted(results.items(), key=lambda x: x[1])
fastest_time = sorted_results[0][1] if sorted_results else 1
# Build the table rows
table_rows = []
for name, avg in sorted_results:
ratio = avg / fastest_time
if ratio == 1.0:
table_rows.append(f'| **{name}** | **{avg:.2f}** | **1.00x (fastest)** |')
else:
avg_str = f'{avg:,.2f}' if avg >= 1000 else f'{avg:.2f}'
ratio_str = f'{ratio:,.0f}x' if ratio >= 100 else f'{ratio:.2f}x'
table_rows.append(f'| {name} | {avg_str} | {ratio_str} |')
# Save to file
with open(filepath, 'w') as f:
f.write('# PraisonAI Agents - Tools Benchmark Results\n\n')
f.write(f'**Generated:** {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}\n')
f.write(f'**Iterations:** {ITERATIONS}\n')
f.write('**Test:** Agent instantiation WITH TOOLS\n\n')
f.write('## Results\n\n')
f.write('| Framework | Avg Time (μs) | Relative |\n')
f.write('|-----------|---------------|----------|\n')
f.write('\n'.join(table_rows) + '\n')
f.write('\n## Package Versions\n\n')
f.write('| Package | Version |\n')
f.write('|---------|--------|\n')
for pkg, ver in versions.items():
f.write(f'| {pkg} | {ver} |\n')
f.write('\n## How to Reproduce\n\n')
f.write('```bash\n')
f.write('cd praisonai-agents\n')
f.write('python benchmarks/tools_benchmark.py\n')
f.write('```\n')
print(f'\nResults saved to: {filepath}')
return filepath
if __name__ == '__main__':
results = run_benchmark()
save_results(results)
@@ -551,8 +551,9 @@ Your Goal: {self.goal}
self.auto_save = auto_save # Session name for auto-saving
# Initialize rules manager for persistent context (like Cursor/Windsurf)
# NOTE: Lazy initialization - rules are loaded only when accessed (performance optimization)
self._rules_manager = None
self._init_rules_manager()
self._rules_manager_initialized = False
# Handle web_search fallback: inject DuckDuckGo tool for unsupported models
if web_search and not self._model_supports_web_search():
@@ -599,7 +600,7 @@ Your Goal: {self.goal}
# Fast Context configuration (lazy loaded)
self.fast_context_enabled = fast_context
self.fast_context_path = fast_context_path or os.getcwd()
self._fast_context_path = fast_context_path # Store raw, resolve lazily
self.fast_context_model = fast_context_model
self.fast_context_max_turns = fast_context_max_turns
self.fast_context_parallelism = fast_context_parallelism
@@ -614,6 +615,17 @@ Your Goal: {self.goal}
self._console = Console()
return self._console
@property
def fast_context_path(self):
"""Lazily resolve fast_context_path - avoids os.getcwd() on every init."""
if self._fast_context_path is None:
self._fast_context_path = os.getcwd()
return self._fast_context_path
@fast_context_path.setter
def fast_context_path(self, value):
self._fast_context_path = value
@property
def _openai_client(self):
"""Lazily initialize OpenAI client only when needed."""
@@ -785,6 +797,21 @@ Your Goal: {self.goal}
logging.warning(f"FastContext search failed: {e}")
return None
@property
def rules_manager(self):
"""
Lazy-initialized RulesManager for persistent rules/instructions.
This property initializes the RulesManager only when first accessed,
avoiding expensive filesystem operations during agent instantiation.
Returns:
RulesManager instance or None if not available
"""
if not self._rules_manager_initialized:
self._init_rules_manager()
return self._rules_manager
def _init_rules_manager(self):
"""
Initialize RulesManager for persistent rules/instructions.
@@ -793,7 +820,10 @@ Your Goal: {self.goal}
- ~/.praison/rules/ (global)
- .praison/rules/ (workspace)
- Subdirectory rules
NOTE: This is called lazily via the rules_manager property for performance.
"""
self._rules_manager_initialized = True
try:
from ..memory.rules_manager import RulesManager
import os
@@ -828,10 +858,10 @@ Your Goal: {self.goal}
Returns:
Formatted rules context string
"""
if not self._rules_manager:
if not self.rules_manager:
return ""
return self._rules_manager.build_rules_context(
return self.rules_manager.build_rules_context(
file_path=file_path,
include_manual=include_manual
)
@@ -1239,8 +1269,8 @@ Your Goal: {self.goal}
Your Role: {self.role}\n
Your Goal: {self.goal}"""
# Add rules context if rules manager is enabled
if self._rules_manager:
# Add rules context if rules manager is enabled (lazy initialization)
if self._rules_manager_initialized and self._rules_manager:
rules_context = self.get_rules_context()
if rules_context:
system_prompt += f"\n\n## Rules (Guidelines you must follow)\n{rules_context}"
@@ -1370,6 +1370,61 @@ Context:
agent_names = ", ".join([agent.name for agent in self.agents])
print(f"📊 Available agents for this endpoint ({len(self.agents)}): {agent_names}")
# Create per-agent endpoints for individual agent access
# This allows n8n and other tools to call specific agents
agents_dict = {agent.name.lower().replace(' ', '_'): agent for agent in self.agents}
# Add GET endpoint to list available agents
@_agents_shared_apps[port].get(f"{path}/list")
async def list_agents():
return {
"agents": [
{"name": agent.name, "id": agent.name.lower().replace(' ', '_')}
for agent in self.agents
]
}
# Add per-agent POST endpoints
for agent_id, agent_instance in agents_dict.items():
agent_path = f"{path}/{agent_id}"
# Create a closure to capture the agent instance
def create_agent_handler(agent):
async def handle_single_agent(request: Request):
try:
request_data = await request.json()
query = request_data.get("query", "")
if not query:
raise HTTPException(status_code=400, detail="Missing 'query' field")
except:
raise HTTPException(status_code=400, detail="Invalid JSON body")
try:
if asyncio.iscoroutinefunction(agent.chat):
response = await agent.achat(query)
else:
loop = asyncio.get_running_loop()
response = await loop.run_in_executor(None, lambda q=query: agent.chat(q))
return {
"agent": agent.name,
"query": query,
"response": response
}
except Exception as e:
logging.error(f"Error with agent {agent.name}: {str(e)}", exc_info=True)
return JSONResponse(
status_code=500,
content={"error": f"Agent error: {str(e)}"}
)
return handle_single_agent
# Register the endpoint
_agents_shared_apps[port].post(agent_path)(create_agent_handler(agent_instance))
_agents_registered_endpoints[port][agent_path] = f"{endpoint_id}_{agent_id}"
print(f"🔗 Per-agent endpoints: {', '.join([f'{path}/{aid}' for aid in agents_dict.keys()])}")
# Start the server if it's not already running for this port
if not _agents_server_started.get(port, False):
# Mark the server as started first to prevent duplicate starts
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "0.1.11"
version = "0.1.14"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
readme = "README.md"
requires-python = ">=3.10"
+1 -1
View File
@@ -2422,7 +2422,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.1.11"
version = "0.1.14"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
+57
View File
@@ -34,6 +34,7 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
## 📑 Table of Contents
- [⚡ Performance](#-performance)
- [🚀 Quick Start](#-quick-start)
- [📦 Installation](#-installation)
- [Python SDK](#python-sdk)
@@ -73,6 +74,32 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
---
## ⚡ Performance
PraisonAI Agents is the **fastest AI agent framework** for agent instantiation.
| Framework | Avg Time (μs) | Relative |
|-----------|---------------|----------|
| **PraisonAI** | **3.77** | **1.00x (fastest)** |
| OpenAI Agents SDK | 5.26 | 1.39x |
| Agno | 5.64 | 1.49x |
| PraisonAI (LiteLLM) | 7.56 | 2.00x |
| PydanticAI | 226.94 | 60.16x |
| LangGraph | 4,558.71 | 1,209x |
| CrewAI | 15,607.92 | 4,138x |
<details>
<summary>Run benchmarks yourself</summary>
```bash
cd praisonai-agents
python benchmarks/simple_benchmark.py
```
</details>
---
## 🚀 Quick Start
Get started with PraisonAI in under 1 minute:
@@ -1392,6 +1419,36 @@ praisonai commit
praisonai commit --push
```
### Serve CLI (API Server):
```bash
# Start API server for agents defined in YAML
praisonai serve agents.yaml
# With custom port and host
praisonai serve agents.yaml --port 8005 --host 0.0.0.0
# Alternative flag style
praisonai agents.yaml --serve
# The server provides:
# POST /agents - Run all agents sequentially
# POST /agents/{name} - Run specific agent (e.g., /agents/researcher)
# GET /agents/list - List available agents
```
### n8n Integration CLI:
```bash
# Export workflow to n8n and open in browser
praisonai agents.yaml --n8n
# With custom n8n URL
praisonai agents.yaml --n8n --n8n-url http://localhost:5678
# Set N8N_API_KEY for auto-import
export N8N_API_KEY="your-api-key"
praisonai agents.yaml --n8n
```
### @Mentions in Prompts:
```bash
# Include file content in prompt
+133
View File
@@ -0,0 +1,133 @@
{
"id": "e022a3a1",
"name": "Create Movie Script About Cat in Mars",
"nodes": [
{
"id": "trigger",
"name": "Webhook",
"type": "n8n-nodes-base.webhook",
"typeVersion": 2,
"position": [
250,
300
],
"webhookId": "dd7cf5cb72d19eae",
"parameters": {
"path": "create-movie-script-about-cat-in-mars",
"httpMethod": "POST",
"responseMode": "lastNode",
"options": {}
}
},
{
"id": "agent_0",
"name": "Researcher",
"type": "n8n-nodes-base.httpRequest",
"typeVersion": 4.2,
"position": [
450,
300
],
"parameters": {
"method": "POST",
"url": "http://127.0.0.1:8005/agents/researcher",
"sendBody": true,
"specifyBody": "json",
"jsonBody": "={{ JSON.stringify({ query: $json.body?.query || $json.query || 'Research about Mars environment and cat behavior for the movie concept' } }) }}",
"options": {
"timeout": 300000
}
},
"notes": "Agent: Researcher\nTask: Research about Mars environment and cat behavior for the movie concept"
},
{
"id": "agent_1",
"name": "Narrative Designer",
"type": "n8n-nodes-base.httpRequest",
"typeVersion": 4.2,
"position": [
700,
300
],
"parameters": {
"method": "POST",
"url": "http://127.0.0.1:8005/agents/narrative_designer",
"sendBody": true,
"specifyBody": "json",
"jsonBody": "={{ JSON.stringify({ query: $json.response || 'Create story concept based on research: {{previous_output}}' } }) }}",
"options": {
"timeout": 300000
}
},
"notes": "Agent: Narrative Designer\nTask: Create story concept based on research: {{previous_output}}"
},
{
"id": "agent_2",
"name": "Scriptwriter",
"type": "n8n-nodes-base.httpRequest",
"typeVersion": 4.2,
"position": [
950,
300
],
"parameters": {
"method": "POST",
"url": "http://127.0.0.1:8005/agents/scriptwriter",
"sendBody": true,
"specifyBody": "json",
"jsonBody": "={{ JSON.stringify({ query: $json.response || 'Write the final movie script based on the story concept: {{previous_output}}' } }) }}",
"options": {
"timeout": 300000
}
},
"notes": "Agent: Scriptwriter\nTask: Write the final movie script based on the story concept: {{previous_output}}"
}
],
"connections": {
"Webhook": {
"main": [
[
{
"node": "Researcher",
"type": "main",
"index": 0
}
]
]
},
"Researcher": {
"main": [
[
{
"node": "Narrative Designer",
"type": "main",
"index": 0
}
]
]
},
"Narrative Designer": {
"main": [
[
{
"node": "Scriptwriter",
"type": "main",
"index": 0
}
]
]
}
},
"active": false,
"settings": {
"executionOrder": "v1"
},
"versionId": "1",
"pinData": {},
"meta": {
"instanceId": "praisonai-export",
"templateCredsSetupCompleted": true,
"description": "A sequential workflow to research, design narrative, and write a movie script"
},
"tags": []
}
@@ -0,0 +1,93 @@
# Advanced Workflow Example
# Demonstrates: conditional routing, decision tasks, and multi-agent coordination
name: Content Creation Pipeline
description: A workflow that researches, writes, reviews, and publishes content
workflow:
process: workflow # Use workflow process for conditional execution
verbose: true
agents:
researcher:
name: Researcher
role: Research Specialist
goal: Gather comprehensive information on topics
backstory: Expert at finding and synthesizing information from various sources
instructions: |
Research the given topic thoroughly.
Provide detailed findings with key points.
llm: gpt-4o-mini
writer:
name: Writer
role: Content Writer
goal: Create engaging and informative content
backstory: Skilled writer who transforms research into readable content
instructions: |
Write clear, engaging content based on the research.
Structure the content with introduction, body, and conclusion.
llm: gpt-4o-mini
reviewer:
name: Reviewer
role: Quality Reviewer
goal: Ensure content meets quality standards
backstory: Meticulous editor who ensures accuracy and clarity
instructions: |
Review the content for accuracy, clarity, and engagement.
Decide if it's ready for publication or needs revision.
llm: gpt-4o-mini
publisher:
name: Publisher
role: Content Publisher
goal: Finalize and format content for publication
backstory: Expert at preparing content for various platforms
instructions: |
Format the approved content for publication.
Add metadata and prepare final deliverable.
llm: gpt-4o-mini
steps:
# Step 1: Research (starting point)
- name: research_task
agent: researcher
action: "Research the topic: AI trends in 2025"
expected_output: "Comprehensive research findings with key points and sources"
is_start: true
next_tasks: [write_task]
# Step 2: Write content based on research
- name: write_task
agent: writer
action: "Write an article based on the research findings"
expected_output: "Well-structured article with introduction, body, and conclusion"
context: [research_task]
next_tasks: [review_task]
# Step 3: Review with decision (conditional routing)
- name: review_task
agent: reviewer
action: "Review the article for quality and accuracy"
expected_output: "Quality assessment with approval or revision recommendation"
context: [write_task]
task_type: decision
condition:
approved: [publish_task]
needs_revision: [revise_task]
# Step 4a: Revision (conditional - only if review fails)
- name: revise_task
agent: writer
action: "Revise the article based on reviewer feedback"
expected_output: "Revised article addressing all feedback"
context: [write_task, review_task]
next_tasks: [review_task] # Loop back to review
# Step 4b: Publish (conditional - only if review passes)
- name: publish_task
agent: publisher
action: "Prepare the approved article for publication"
expected_output: "Final formatted article ready for publication"
context: [write_task, review_task]
+2 -2
View File
@@ -3,8 +3,8 @@ class Praisonai < Formula
desc "AI tools for various AI applications"
homepage "https://github.com/MervinPraison/PraisonAI"
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.30.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.30.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.34.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.34.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
@@ -20,6 +20,7 @@ Features:
- router: Smart model selection
- flow_display: Visual workflow tracking
- workflow: YAML workflow management
- n8n: Export workflows to n8n visual editor
"""
# Type hints available for IDE support
@@ -41,6 +42,7 @@ __all__ = [
'RouterHandler',
'FlowDisplayHandler',
'WorkflowHandler',
'N8nHandler',
]
def __getattr__(name):
@@ -90,4 +92,7 @@ def __getattr__(name):
elif name == 'WorkflowHandler':
from .workflow import WorkflowHandler
return WorkflowHandler
elif name == 'N8nHandler':
from .n8n import N8nHandler
return N8nHandler
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+358 -101
View File
@@ -12,11 +12,17 @@ Usage:
import json
import yaml
import webbrowser
import base64
import urllib.parse
import hashlib
import os
from typing import Any, Dict, List, Optional
from pathlib import Path
try:
import requests
REQUESTS_AVAILABLE = True
except ImportError:
REQUESTS_AVAILABLE = False
from .base import FlagHandler
@@ -28,17 +34,21 @@ class N8nHandler(FlagHandler):
and opens them in the n8n UI for visual editing.
"""
def __init__(self, verbose: bool = False, n8n_url: str = "http://localhost:5678"):
def __init__(self, verbose: bool = False, n8n_url: str = "http://localhost:5678",
use_execute_command: bool = False):
"""
Initialize the n8n handler.
Args:
verbose: Enable verbose output
n8n_url: Base URL of the n8n instance
use_execute_command: Use Execute Command nodes (runs praisonai directly)
instead of HTTP Request nodes
"""
super().__init__(verbose=verbose)
self.n8n_url = n8n_url
self.praisonai_api_url = "http://localhost:8000"
self.praisonai_api_url = "http://127.0.0.1:8005"
self.use_execute_command = use_execute_command
@property
def feature_name(self) -> str:
@@ -71,31 +81,168 @@ class N8nHandler(FlagHandler):
# Convert YAML to n8n workflow JSON
workflow_json = self.convert_yaml_to_n8n(yaml_path)
# Generate n8n URL
url = self.generate_n8n_url(workflow_json, self.n8n_url)
# Save workflow JSON to file
output_path = self._save_workflow_json(workflow_json, yaml_path)
self.print_status("✅ Workflow converted successfully!", "success")
self.print_status(f"📄 JSON saved to: {output_path}", "info")
if open_browser:
self.open_in_browser(url)
self.print_status(f"🌐 Opening in n8n: {url}", "info")
# Try to create workflow via n8n API
api_key = os.environ.get('N8N_API_KEY')
workflow_id = None
if api_key and REQUESTS_AVAILABLE:
workflow_id = self._create_workflow_via_api(workflow_json, api_key)
webhook_url = None
if workflow_id:
# Successfully created via API - open directly
url = f"{self.n8n_url}/workflow/{workflow_id}"
self.print_status("🚀 Workflow created in n8n!", "success")
# Activate the workflow so webhook is available
if self._activate_workflow(workflow_id, api_key):
self.print_status("✅ Workflow activated!", "success")
# Get webhook path from the workflow
trigger_node = next((n for n in workflow_json["nodes"]
if n.get("type") == "n8n-nodes-base.webhook"), None)
if trigger_node:
webhook_path = trigger_node["parameters"].get("path", "praisonai")
webhook_url = f"{self.n8n_url}/webhook/{webhook_path}"
self.print_status("", "info")
self.print_status("🔗 Webhook URL (to trigger workflow):", "info")
self.print_status(f" POST {webhook_url}", "info")
else:
self.print_status("⚠️ Could not activate workflow (activate manually in n8n)", "warning")
if open_browser:
self.open_in_browser(url)
self.print_status(f"🌐 Opening: {url}", "info")
else:
# Fallback to manual import instructions
url = self.generate_n8n_url(workflow_json, self.n8n_url)
if not api_key:
self.print_status("", "info")
self.print_status("💡 Tip: Set N8N_API_KEY env var for auto-import", "info")
self.print_status(" Generate API key in n8n: Settings → API", "info")
self.print_status("", "info")
self.print_status("📋 To import into n8n:", "info")
self.print_status(" 1. Open n8n in your browser", "info")
self.print_status(" 2. Click the three dots menu (⋮) in the top right", "info")
self.print_status(" 3. Select 'Import from File'", "info")
self.print_status(f" 4. Choose: {output_path}", "info")
if open_browser:
self.open_in_browser(url)
self.print_status(f"🌐 Opening n8n: {url}", "info")
return {
"workflow": workflow_json,
"url": url,
"output_path": output_path
"output_path": output_path,
"workflow_id": workflow_id,
"webhook_url": webhook_url
}
def convert_yaml_to_n8n(self, yaml_path: str) -> Dict[str, Any]:
def _create_workflow_via_api(self, workflow_json: Dict[str, Any],
api_key: str) -> Optional[str]:
"""
Create workflow in n8n via REST API.
Args:
workflow_json: The n8n workflow JSON
api_key: n8n API key
Returns:
Workflow ID if successful, None otherwise
"""
if not REQUESTS_AVAILABLE:
return None
try:
# Only include fields allowed by n8n API schema
# See: n8n/packages/cli/src/public-api/v1/handlers/workflows/spec/schemas/workflow.yml
workflow_data = {
"name": workflow_json.get("name", "PraisonAI Workflow"),
"nodes": workflow_json.get("nodes", []),
"connections": workflow_json.get("connections", {}),
"settings": workflow_json.get("settings", {"executionOrder": "v1"}),
}
# Optional fields
if "staticData" in workflow_json:
workflow_data["staticData"] = workflow_json["staticData"]
response = requests.post(
f"{self.n8n_url}/api/v1/workflows",
headers={
"X-N8N-API-KEY": api_key,
"Content-Type": "application/json"
},
json=workflow_data,
timeout=30
)
if response.status_code == 200:
result = response.json()
return result.get('id')
else:
self.log(f"API error: {response.status_code} - {response.text}", "warning")
return None
except Exception as e:
self.log(f"Failed to create workflow via API: {e}", "warning")
return None
def _activate_workflow(self, workflow_id: str, api_key: str) -> bool:
"""
Activate a workflow via n8n API.
Args:
workflow_id: The workflow ID to activate
api_key: n8n API key
Returns:
True if successful, False otherwise
"""
if not REQUESTS_AVAILABLE:
return False
try:
response = requests.post(
f"{self.n8n_url}/api/v1/workflows/{workflow_id}/activate",
headers={
"X-N8N-API-KEY": api_key,
"Content-Type": "application/json"
},
timeout=30
)
if response.status_code == 200:
return True
else:
self.log(f"Activation error: {response.status_code} - {response.text}", "warning")
return False
except Exception as e:
self.log(f"Failed to activate workflow: {e}", "warning")
return False
def convert_yaml_to_n8n(self, yaml_path: str, use_webhook: bool = True) -> Dict[str, Any]:
"""
Convert a PraisonAI YAML workflow to n8n JSON format.
The n8n workflow will have:
- A webhook trigger to receive requests
- One HTTP Request node per agent (calls /agents/{agent_name})
- Each agent node passes its output to the next agent
Args:
yaml_path: Path to the agents.yaml file
use_webhook: Use webhook trigger for programmatic execution (default: True)
Returns:
n8n workflow JSON as dictionary
@@ -108,6 +255,9 @@ class N8nHandler(FlagHandler):
workflow_name = yaml_content.get('name', 'PraisonAI Workflow')
description = yaml_content.get('description', '')
# Generate a unique webhook path from workflow name
webhook_path = workflow_name.lower().replace(' ', '-').replace('/', '-')[:50]
# Get agents and steps
agents = yaml_content.get('agents', yaml_content.get('roles', {}))
steps = yaml_content.get('steps', [])
@@ -120,85 +270,50 @@ class N8nHandler(FlagHandler):
nodes = []
connections = {}
# Add manual trigger node
trigger_node = self._create_trigger_node()
# Add trigger node (webhook for programmatic execution, manual otherwise)
trigger_node = self._create_trigger_node(use_webhook=use_webhook, webhook_path=webhook_path)
nodes.append(trigger_node)
# Track previous node for connections
prev_node_name = trigger_node["name"]
x_position = 450
# Process each step
x_position = 450 # Start position after trigger
# Create one HTTP node per agent step
for i, step in enumerate(steps):
if isinstance(step, dict):
if 'parallel' in step:
# Handle parallel steps
parallel_nodes, parallel_connections = self._create_parallel_nodes(
step['parallel'], x_position, agents
)
nodes.extend(parallel_nodes)
connections.update(parallel_connections)
# Connect previous node to first parallel node
if prev_node_name not in connections:
connections[prev_node_name] = {"main": [[]]}
for pnode in parallel_nodes:
connections[prev_node_name]["main"][0].append({
"node": pnode["name"],
"type": "main",
"index": 0
})
x_position += 200 * len(parallel_nodes)
prev_node_name = parallel_nodes[-1]["name"] if parallel_nodes else prev_node_name
elif 'route' in step:
# Handle route/decision steps
route_node = self._create_route_node(step, x_position, i)
nodes.append(route_node)
# Connect previous node
if prev_node_name not in connections:
connections[prev_node_name] = {"main": [[]]}
connections[prev_node_name]["main"][0].append({
"node": route_node["name"],
"type": "main",
"index": 0
})
prev_node_name = route_node["name"]
x_position += 200
else:
# Regular agent step
agent_id = step.get('agent', '')
action = step.get('action', step.get('description', ''))
agent_config = agents.get(agent_id, {})
http_node = self._create_http_request_node(
agent_id=agent_id,
agent_config=agent_config,
action=action,
position=[x_position, 300],
index=i
)
nodes.append(http_node)
# Connect to previous node
if prev_node_name not in connections:
connections[prev_node_name] = {"main": [[]]}
connections[prev_node_name]["main"][0].append({
"node": http_node["name"],
"type": "main",
"index": 0
})
prev_node_name = http_node["name"]
x_position += 200
agent_id = step.get('agent', '')
action = step.get('action', step.get('description', ''))
agent_config = agents.get(agent_id, {})
# Create HTTP node that calls the specific agent endpoint
agent_node = self._create_per_agent_node(
agent_id=agent_id,
agent_config=agent_config,
action=action,
position=[x_position, 300],
index=i,
is_first=(i == 0)
)
nodes.append(agent_node)
# Connect to previous node
if prev_node_name not in connections:
connections[prev_node_name] = {"main": [[]]}
connections[prev_node_name]["main"][0].append({
"node": agent_node["name"],
"type": "main",
"index": 0
})
prev_node_name = agent_node["name"]
x_position += 250
# Build the complete workflow
# Generate a unique ID based on workflow name
workflow_id = hashlib.md5(workflow_name.encode()).hexdigest()[:8]
workflow = {
"id": workflow_id,
"name": workflow_name,
"nodes": nodes,
"connections": connections,
@@ -207,13 +322,12 @@ class N8nHandler(FlagHandler):
"executionOrder": "v1"
},
"versionId": "1",
"pinData": {},
"meta": {
"instanceId": "praisonai-export",
"templateCredsSetupCompleted": True
},
"tags": [
{"name": "praisonai"}
]
"tags": []
}
if description:
@@ -242,8 +356,31 @@ class N8nHandler(FlagHandler):
})
return steps
def _create_trigger_node(self) -> Dict[str, Any]:
"""Create a manual trigger node."""
def _create_trigger_node(self, use_webhook: bool = False,
webhook_path: str = "praisonai") -> Dict[str, Any]:
"""Create a trigger node (webhook or manual).
Args:
use_webhook: If True, creates a webhook trigger for programmatic execution
webhook_path: Path for the webhook URL (only used if use_webhook=True)
"""
if use_webhook:
# Generate a unique webhook ID
webhook_id = hashlib.md5(webhook_path.encode()).hexdigest()[:16]
return {
"id": "trigger",
"name": "Webhook",
"type": "n8n-nodes-base.webhook",
"typeVersion": 2,
"position": [250, 300],
"webhookId": webhook_id,
"parameters": {
"path": webhook_path,
"httpMethod": "POST",
"responseMode": "lastNode",
"options": {}
}
}
return {
"id": "trigger",
"name": "Manual Trigger",
@@ -253,6 +390,130 @@ class N8nHandler(FlagHandler):
"parameters": {}
}
def _create_per_agent_node(self, agent_id: str, agent_config: Dict,
action: str, position: List[int],
index: int, is_first: bool = False) -> Dict[str, Any]:
"""
Create an HTTP Request node that calls a specific agent endpoint.
Args:
agent_id: Agent identifier (used in URL path)
agent_config: Agent configuration from YAML
action: The action/task description for this agent
position: Node position [x, y]
index: Node index
is_first: Whether this is the first agent (uses webhook input)
Returns:
n8n HTTP Request node configuration
"""
agent_name = agent_config.get('name', agent_id.title())
# Convert agent_id to URL-safe format (lowercase, underscores)
agent_url_id = agent_id.lower().replace(' ', '_')
# Build the query - first agent uses webhook input, others use previous response
if is_first:
# First agent: use query from webhook body
query_expr = f"={{ $json.body?.query || $json.query || '{action}' }}"
else:
# Subsequent agents: use previous agent's response + original action context
query_expr = f"={{ $json.response || '{action}' }}"
return {
"id": f"agent_{index}",
"name": agent_name,
"type": "n8n-nodes-base.httpRequest",
"typeVersion": 4.2,
"position": position,
"parameters": {
"method": "POST",
"url": f"{self.praisonai_api_url}/agents/{agent_url_id}",
"sendBody": True,
"specifyBody": "json",
"jsonBody": f"={{{{ JSON.stringify({{ query: {query_expr[2:]} }}) }}}}",
"options": {
"timeout": 300000 # 5 minute timeout per agent
}
},
"notes": f"Agent: {agent_name}\nTask: {action[:100]}"
}
def _create_praisonai_workflow_node(self, workflow_name: str,
agent_names: List[str],
position: List[int]) -> Dict[str, Any]:
"""
Create a single HTTP Request node that triggers the entire PraisonAI workflow.
The PraisonAI API will run all agents sequentially when called.
Args:
workflow_name: Name of the workflow
agent_names: List of agent names (for documentation)
position: Node position [x, y]
Returns:
n8n HTTP Request node configuration
"""
# The query will be passed from the webhook trigger
# Using n8n expression to get the body from the webhook
return {
"id": "praisonai_workflow",
"name": f"PraisonAI: {workflow_name}",
"type": "n8n-nodes-base.httpRequest",
"typeVersion": 4.2,
"position": position,
"parameters": {
"method": "POST",
"url": f"{self.praisonai_api_url}/agents",
"sendBody": True,
"specifyBody": "json",
# Pass the query from webhook body, or use a default
"jsonBody": "={{ JSON.stringify({ query: $json.body?.query || $json.query || 'Start workflow' }) }}",
"options": {
"timeout": 600000 # 10 minute timeout for full workflow
}
},
"notes": f"Runs agents: {', '.join(agent_names)}"
}
def _create_agent_node(self, agent_id: str, agent_config: Dict,
action: str, position: List[int],
index: int) -> Dict[str, Any]:
"""Create a node for an agent - either Execute Command or HTTP Request."""
if self.use_execute_command:
return self._create_execute_command_node(
agent_id, agent_config, action, position, index
)
else:
return self._create_http_request_node(
agent_id, agent_config, action, position, index
)
def _create_execute_command_node(self, agent_id: str, agent_config: Dict,
action: str, position: List[int],
index: int) -> Dict[str, Any]:
"""Create an Execute Command node that runs praisonai directly."""
agent_name = agent_config.get('name', agent_id.title())
# Escape quotes in the action for shell command
escaped_action = action.replace('"', '\\"').replace("'", "\\'")
# Build the praisonai command
# Uses the input from previous node if available
command = f'praisonai "{escaped_action}"'
return {
"id": f"agent_{index}",
"name": agent_name,
"type": "n8n-nodes-base.executeCommand",
"typeVersion": 1,
"position": position,
"parameters": {
"command": command,
"executeOnce": True
}
}
def _create_http_request_node(self, agent_id: str, agent_config: Dict,
action: str, position: List[int],
index: int) -> Dict[str, Any]:
@@ -295,7 +556,7 @@ class N8nHandler(FlagHandler):
action = step.get('action', step.get('description', ''))
agent_config = agents.get(agent_id, {})
node = self._create_http_request_node(
node = self._create_agent_node(
agent_id=agent_id,
agent_config=agent_config,
action=action,
@@ -344,26 +605,22 @@ class N8nHandler(FlagHandler):
def generate_n8n_url(self, workflow_json: Dict[str, Any],
base_url: str) -> str:
"""
Generate a URL to open the workflow in n8n.
Generate a URL to open n8n workflow editor.
Note: n8n doesn't support URL hash-based import. Users need to:
1. Open n8n
2. Use "Import from File" to load the saved JSON
Args:
workflow_json: The n8n workflow JSON
workflow_json: The n8n workflow JSON (unused, kept for API compatibility)
base_url: Base URL of the n8n instance
Returns:
URL to open the workflow in n8n
URL to open n8n workflow editor
"""
# Encode workflow as base64 for URL
workflow_str = json.dumps(workflow_json)
workflow_b64 = base64.b64encode(workflow_str.encode()).decode()
# URL encode the base64 string
workflow_encoded = urllib.parse.quote(workflow_b64)
# n8n import URL format
url = f"{base_url}/workflow/new#import:{workflow_encoded}"
return url
# n8n doesn't support URL-based workflow import via hash
# Return the new workflow URL - user will need to import manually
return f"{base_url}/workflow/new"
def open_in_browser(self, url: str) -> None:
"""
+189 -1
View File
@@ -499,6 +499,16 @@ class PraisonAI:
print(result)
return result
else:
# n8n Integration - Export workflow to n8n and open in browser
if getattr(args, 'n8n', False):
from .features.n8n import N8nHandler
n8n_handler = N8nHandler(
verbose=getattr(args, 'verbose', False),
n8n_url=getattr(args, 'n8n_url', 'http://localhost:5678')
)
result = n8n_handler.execute(self.agent_file)
return result
# Flow Display - Show flow diagram WITHOUT executing (--flow-display flag)
if getattr(args, 'flow_display', False):
from .features.flow_display import FlowDisplayHandler
@@ -568,7 +578,7 @@ class PraisonAI:
return default_args
# Define special commands
special_commands = ['chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research', 'memory', 'rules', 'workflow', 'hooks', 'knowledge', 'session', 'tools', 'todo', 'docs', 'mcp', 'commit']
special_commands = ['chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research', 'memory', 'rules', 'workflow', 'hooks', 'knowledge', 'session', 'tools', 'todo', 'docs', 'mcp', 'commit', 'serve']
parser = argparse.ArgumentParser(prog="praisonai", description="praisonAI command-line interface")
parser.add_argument("--framework", choices=["crewai", "autogen", "praisonai"], help="Specify the framework")
@@ -669,6 +679,15 @@ class PraisonAI:
# Flow Display - visual workflow
parser.add_argument("--flow-display", action="store_true", help="Enable visual workflow tracking")
# n8n Integration - export workflow to n8n
parser.add_argument("--n8n", action="store_true", help="Export workflow to n8n and open in browser")
parser.add_argument("--n8n-url", type=str, default="http://localhost:5678", help="n8n instance URL (default: http://localhost:5678)")
# Serve - start API server for agents
parser.add_argument("--serve", action="store_true", help="Start API server for agents (use with agents.yaml)")
parser.add_argument("--port", type=int, default=8005, help="Server port (default: 8005)")
parser.add_argument("--host", type=str, default="127.0.0.1", help="Server host (default: 127.0.0.1)")
# If we're in a test environment, parse with empty args to avoid pytest interference
if in_test_env:
args, unknown_args = parser.parse_known_args([])
@@ -696,6 +715,13 @@ class PraisonAI:
args.realtime = True
if args.command == 'call':
args.call = True
if args.command == 'serve':
args.serve = True
# Handle serve command - start API server for agents
if args.command == 'serve' or args.serve:
self._handle_serve_command(args, unknown_args)
sys.exit(0)
# Handle both command and flag versions for call
if args.command == 'call' or args.call:
@@ -3030,6 +3056,168 @@ Now, {final_instruction.lower()}:"""
print("pip install praisonaiagents # For PraisonAIAgents\n")
sys.exit(1)
def _handle_serve_command(self, args, unknown_args):
"""
Handle the serve command - start API server for agents.
Usage:
praisonai serve agents.yaml
praisonai serve agents.yaml --port 8005 --host 0.0.0.0
praisonai agents.yaml --serve
"""
import yaml
from praisonaiagents import Agent, PraisonAIAgents
# Determine the YAML file path
yaml_file = None
# Check if command is 'serve' and there's a file in unknown_args
if args.command == 'serve' and unknown_args:
yaml_file = unknown_args[0]
# Check if command itself is a yaml file (praisonai agents.yaml --serve)
elif args.command and args.command.endswith('.yaml'):
yaml_file = args.command
# Default to agents.yaml
else:
yaml_file = 'agents.yaml'
# Check if file exists
if not os.path.exists(yaml_file):
print(f"[red]ERROR: File not found: {yaml_file}[/red]")
print("\nUsage:")
print(" praisonai serve agents.yaml")
print(" praisonai serve agents.yaml --port 8005")
print(" praisonai agents.yaml --serve")
sys.exit(1)
print(f"📄 Loading agents from: {yaml_file}")
# Load YAML file
with open(yaml_file, 'r') as f:
config = yaml.safe_load(f)
# Extract agents
agents_config = config.get('agents', config.get('roles', {}))
agents_dict = {} # Map agent_id to Agent instance
agents_list = []
for agent_id, agent_config in agents_config.items():
if isinstance(agent_config, dict):
agent = Agent(
name=agent_config.get('name', agent_id.title()),
role=agent_config.get('role', ''),
goal=agent_config.get('goal', ''),
backstory=agent_config.get('backstory', ''),
instructions=agent_config.get('instructions', ''),
llm=agent_config.get('llm', 'gpt-4o-mini'),
verbose=False
)
agents_dict[agent_id] = agent
agents_list.append(agent)
print(f" ✓ Loaded agent: {agent.name}")
if not agents_list:
print("[red]ERROR: No agents found in YAML file[/red]")
sys.exit(1)
# Extract tasks from steps (if defined)
steps = config.get('steps', [])
tasks_list = []
if steps:
from praisonaiagents import Task
print(f"\n📋 Loading workflow steps...")
# First pass: create all tasks
task_name_map = {} # Map step names to Task objects for context/next_tasks
for i, step in enumerate(steps):
if isinstance(step, dict):
agent_id = step.get('agent', '')
action = step.get('action', step.get('description', ''))
task_name = step.get('name', f"step_{i+1}")
if agent_id in agents_dict:
task = Task(
name=task_name,
description=action,
agent=agents_dict[agent_id],
expected_output=step.get('expected_output', 'Completed task output'),
task_type=step.get('task_type', 'task'),
is_start=step.get('is_start', i == 0),
async_execution=step.get('async_execution', False),
output_file=step.get('output_file'),
)
tasks_list.append(task)
task_name_map[task_name] = task
print(f" ✓ Loaded step {i+1}: {agent_id} → {action[:50]}...")
# Second pass: set up next_tasks and context relationships
for i, step in enumerate(steps):
if isinstance(step, dict):
task_name = step.get('name', f"step_{i+1}")
if task_name in task_name_map:
task = task_name_map[task_name]
# Set next_tasks if specified
next_tasks = step.get('next_tasks', [])
if next_tasks:
task.next_tasks = next_tasks
# Set context if specified
context_refs = step.get('context', [])
if context_refs:
task.context = [task_name_map[ref] for ref in context_refs if ref in task_name_map]
# Set condition for workflow branching
condition = step.get('condition', {})
if condition:
task.condition = condition
port = args.port
host = args.host
# Extract process type from workflow config (default: sequential)
workflow_config = config.get('workflow', {})
process_type = workflow_config.get('process', 'sequential')
verbose = workflow_config.get('verbose', False)
# Also check top-level process key
if 'process' in config:
process_type = config['process']
print(f"\n🚀 Starting PraisonAI API server...")
print(f" Host: {host}")
print(f" Port: {port}")
print(f" Agents: {len(agents_list)}")
if tasks_list:
print(f" Tasks: {len(tasks_list)}")
print(f" Process: {process_type}")
# Create and launch - with tasks if defined
if tasks_list:
praison = PraisonAIAgents(
agents=agents_list,
tasks=tasks_list,
process=process_type,
verbose=1 if verbose else 0
)
else:
praison = PraisonAIAgents(
agents=agents_list,
process=process_type,
verbose=1 if verbose else 0
)
praison.launch(port=port, host=host)
# Keep the main thread alive to prevent exit
print("\n✅ Server running. Press Ctrl+C to stop.")
try:
import time
while True:
time.sleep(1)
except KeyboardInterrupt:
print("\n👋 Server stopped.")
def create_chainlit_chat_interface(self):
"""
Create a Chainlit interface for the chat application.
+1 -1
View File
@@ -57,7 +57,7 @@ class CloudDeployer:
file.write("FROM python:3.11-slim\n")
file.write("WORKDIR /app\n")
file.write("COPY . .\n")
file.write("RUN pip install flask praisonai==2.3.30 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==2.3.34 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
+1 -1
View File
@@ -1 +1 @@
__version__ = "2.3.30"
__version__ = "2.3.34"
+1 -1
View File
@@ -12,7 +12,7 @@ dependencies = [
"rich>=13.7",
"markdown>=3.5",
"pyparsing>=3.0.0",
"praisonaiagents>=0.1.7",
"praisonaiagents>=0.1.14",
"python-dotenv>=0.19.0",
"instructor>=1.3.3",
"PyYAML>=6.0",
+15 -7
View File
@@ -81,25 +81,33 @@ steps:
assert "connections" in result
def test_n8n_workflow_has_trigger_node(self, sample_yaml_file):
"""Test that n8n workflow has a manual trigger node."""
"""Test that n8n workflow has a trigger node (webhook or manual)."""
from praisonai.cli.features.n8n import N8nHandler
handler = N8nHandler()
result = handler.convert_yaml_to_n8n(sample_yaml_file)
nodes = result["nodes"]
trigger_nodes = [n for n in nodes if "trigger" in n.get("type", "").lower() or "trigger" in n.get("name", "").lower()]
# Check for webhook or manual trigger nodes
trigger_nodes = [n for n in nodes if "trigger" in n.get("type", "").lower()
or "webhook" in n.get("type", "").lower()
or "trigger" in n.get("name", "").lower()
or "webhook" in n.get("name", "").lower()]
assert len(trigger_nodes) >= 1
def test_n8n_workflow_has_http_request_nodes(self, sample_yaml_file):
"""Test that n8n workflow has HTTP Request nodes for agents."""
def test_n8n_workflow_has_per_agent_nodes(self, sample_yaml_file):
"""Test that n8n workflow has HTTP Request nodes for each agent."""
from praisonai.cli.features.n8n import N8nHandler
handler = N8nHandler()
result = handler.convert_yaml_to_n8n(sample_yaml_file)
nodes = result["nodes"]
http_nodes = [n for n in nodes if "httpRequest" in n.get("type", "") or "http" in n.get("type", "").lower()]
# Should have at least 2 HTTP nodes (one per agent step)
assert len(http_nodes) >= 2
# Should have HTTP Request nodes for each agent
http_nodes = [n for n in nodes if "httpRequest" in n.get("type", "")]
assert len(http_nodes) >= 2 # At least 2 agents
# Each node should call a specific agent endpoint
for node in http_nodes:
assert "/agents/" in node["parameters"]["url"]
def test_n8n_workflow_nodes_are_connected(self, sample_yaml_file):
"""Test that nodes are properly connected."""
+4 -4
View File
@@ -4197,7 +4197,7 @@ requires-dist = [
{ name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.0.22" },
{ name = "praisonai-tools", marker = "extra == 'autogen-v4'", specifier = ">=0.0.22" },
{ name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.0.22" },
{ name = "praisonaiagents", specifier = ">=0.1.7" },
{ name = "praisonaiagents", specifier = ">=0.1.14" },
{ name = "pyautogen", marker = "extra == 'autogen'", specifier = "==0.2.29" },
{ name = "pydantic", marker = "extra == 'chat'", specifier = "<=2.10.1" },
{ name = "pydantic", marker = "extra == 'code'", specifier = "<=2.10.1" },
@@ -4252,7 +4252,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.1.11"
version = "0.1.14"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "aiohttp" },
@@ -4261,9 +4261,9 @@ dependencies = [
{ name = "pydantic" },
{ name = "rich" },
]
sdist = { url = "https://files.pythonhosted.org/packages/5e/1d/504c0050b846ccb097a4d5a07d29d8077604f66be8ca0f68e5b03af8010f/praisonaiagents-0.1.11.tar.gz", hash = "sha256:7cee6ff74e55b30f6af02f7b9800421761480e1c5d7e8935074f2b7cecd0c4f9", size = 620011, upload-time = "2025-12-18T11:18:34.061Z" }
sdist = { url = "https://files.pythonhosted.org/packages/aa/19/3493e61f00afc8f2dc6beed5407dd5d18e36f70c96107aa0d8f23c136f26/praisonaiagents-0.1.14.tar.gz", hash = "sha256:3e86b548fa6a44f37b84fc9588d275ede34a303d2f5ac3d0c596f4e2464131b9", size = 621584, upload-time = "2025-12-18T14:39:37.194Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/f4/f8/ec1089eb7270f6906d3a1c770d2229db2192f7e16ea7b88bd2a55a2ad80d/praisonaiagents-0.1.11-py3-none-any.whl", hash = "sha256:8f677999f24353dfe657805303497dfdf42dd9cc8f02ba14add72c35d4a47dba", size = 547199, upload-time = "2025-12-18T11:18:32.542Z" },
{ url = "https://files.pythonhosted.org/packages/0b/f7/618f7278370091eb62c118331646010d139dd88a8477c8bea91e18ac274e/praisonaiagents-0.1.14-py3-none-any.whl", hash = "sha256:034c862bb293248c29f8bf6ce4a910cb17bc8b980dd5cc38d790a4cb16a08729", size = 548184, upload-time = "2025-12-18T14:39:34.989Z" },
]
[[package]]