Compare commits

...
14 Commits
Author SHA1 Message Date
Mervin Praison d3f6eb4096 Release v3.5.2 2026-01-08 21:14:45 +00:00
Mervin Praison c691b0a4c0 Release v3.5.1 2026-01-08 20:12:14 +00:00
Mervin Praison b59675eeb4 Release v3.5.0 2026-01-08 19:55:06 +00:00
Mervin Praison 7e0739c6c4 Fix test_api.py to mock urllib.request.urlopen instead of subprocess.run 2026-01-08 16:14:35 +00:00
Mervin Praison 7519bc85d2 Fix test_api.py mock setup for stderr.read() 2026-01-08 16:03:38 +00:00
Mervin Praison c7471092d0 Update test_typer_cli.py: chat is now a Typer command, not legacy 2026-01-08 15:52:35 +00:00
Mervin Praison c344ee1bad Update test_chat_flag.py to not require deprecated chat_mode flag 2026-01-08 15:42:12 +00:00
Mervin Praison 77bbf1ae66 Fix test_approval_system.py to handle tuple return from parse_args() 2026-01-08 15:33:37 +00:00
Mervin Praison 695d4d8843 Add console.py to output module (was gitignored)
The output/ directory was in .gitignore, so console.py wasn't committed.
Force-adding it to fix the ModuleNotFoundError in CI.
2026-01-08 15:28:17 +00:00
Mervin Praison 08f0736db5 Fix async test infrastructure and CLI output module export
Core SDK fixes:
- Add local async test runner in conftest.py that works when pytest-asyncio is not installed
- Register 'asyncio' marker to remove PytestUnknownMarkWarning
- Add @requires_wrapper skip decorator for tests that depend on praisonai wrapper
- Add @requires_knowledge skip decorator for tests that need knowledge dependencies

Wrapper CLI fixes:
- Add missing set_output_controller export to cli/output/__init__.py
- Update test_doctor_cli.py to test actual CLI subcommands instead of non-existent options

Result: Core SDK unit tests pass (1181 passed, 21 skipped), wrapper integration tests pass (17 passed)
2026-01-08 15:07:35 +00:00
Mervin Praison 1e46093edf Release v3.4.1 2026-01-08 14:36:54 +00:00
Mervin Praison 1a71aac6b8 Release v3.4.0 2026-01-08 13:29:57 +00:00
Mervin Praison 287f42dfee Remove unused test and verification files 2026-01-08 12:50:16 +00:00
Mervin Praison 57b637948f Release v3.3.1 2026-01-08 12:28:06 +00:00
524 changed files with 16228 additions and 3485 deletions

No files matched your search

+1
View File
@@ -20,6 +20,7 @@ watch.sh
docs.sh
other
output
!src/praisonai/praisonai/cli/output/
.files
threads.db
+4 -4
View File
@@ -538,11 +538,11 @@ python app.py
Create app.py file and add the code below:
```python
from praisonaiagents import Agent, PraisonAIAgents
from praisonaiagents import Agent, Agents
research_agent = Agent(instructions="Research about AI")
summarise_agent = Agent(instructions="Summarise research agent's findings")
agents = PraisonAIAgents(agents=[research_agent, summarise_agent])
agents = Agents(agents=[research_agent, summarise_agent])
agents.start()
```
@@ -3380,11 +3380,11 @@ See [Agent Memory](#6-agent-memory-zero-dependencies) for more options.
<summary><strong>How do I run multiple agents together?</strong></summary>
```python
from praisonaiagents import Agent, PraisonAIAgents
from praisonaiagents import Agent, Agents
agent1 = Agent(instructions="Research topics")
agent2 = Agent(instructions="Summarize findings")
agents = PraisonAIAgents(agents=[agent1, agent2])
agents = Agents(agents=[agent1, agent2])
agents.start()
```
-71
View File
@@ -1,71 +0,0 @@
"""
Auto-generated API server for PraisonAI agents.
"""
from flask import Flask, request, jsonify
from flask_cors import CORS
from praisonai import PraisonAI
import os
app = Flask(__name__)
# CORS configuration
CORS(app)
# Authentication
AUTH_ENABLED = False
AUTH_TOKEN = None
def check_auth():
"""Check authentication if enabled."""
if not AUTH_ENABLED:
return True
token = request.headers.get('Authorization', '').replace('Bearer ', '')
return token == AUTH_TOKEN
@app.route('/health', methods=['GET'])
def health():
"""Health check endpoint."""
return jsonify({"status": "ok", "service": "praisonai-api"})
@app.route('/chat', methods=['POST'])
def chat():
"""Chat endpoint for agent interaction."""
if not check_auth():
return jsonify({"error": "Unauthorized"}), 401
data = request.get_json()
if not data or 'message' not in data:
return jsonify({"error": "Message required"}), 400
try:
praisonai = PraisonAI(agent_file="agents.yaml")
result = praisonai.run()
return jsonify({
"response": result,
"status": "success"
})
except Exception as e:
return jsonify({
"error": str(e),
"status": "error"
}), 500
@app.route('/agents', methods=['GET'])
def list_agents():
"""List available agents."""
if not check_auth():
return jsonify({"error": "Unauthorized"}), 401
return jsonify({
"agents": ["default"],
"agent_file": "agents.yaml"
})
if __name__ == '__main__':
app.run(
host='0.0.0.0',
port=8080,
debug=False
)
+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>=3.3.0" \
"praisonai>=3.5.2" \
"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>=3.3.0" \
"praisonai>=3.5.2" \
"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>=3.3.0" \
"praisonai>=3.5.2" \
"praisonai[ui]" \
"praisonai[crewai]"
-203
View File
@@ -1,203 +0,0 @@
# Profiling CLI
PraisonAI provides unified profiling capabilities for performance analysis and debugging, with **full visibility from ENTER to first response**.
## Quick Start
```bash
# Basic profiling
praisonai "What is 2+2?" --profile
# Deep profiling with call graph, decision trace, and module breakdown
praisonai "What is 2+2?" --profile --profile-deep
# JSON output format
praisonai "What is 2+2?" --profile --profile-format json
```
## Time to First Response (TTFR)
**TTFR** is the key user-perceived latency metric. It measures the time from when you press ENTER until the first visible output appears.
### What contributes to TTFR?
1. **CLI Entry** - Time to reach the CLI entry point
2. **CLI Parse** - Argparse/Typer routing time
3. **Routing** - Command routing to handler
4. **Imports** - Loading required modules
5. **Agent Init** - Initializing the agent
6. **Network Request** - Sending request to LLM API
7. **First Token** - Receiving first token (streaming)
8. **First Output** - Rendering first visible output
### How to reduce TTFR
1. **Warm runs** - Imports are cached after first run
2. **Streaming mode** - See output as tokens arrive
3. **Lighter models** - Faster response times
4. **Local models** - Eliminate network latency
## CLI Flags
| Flag | Description | Default |
|------|-------------|---------|
| `--profile` | Enable profiling with timing breakdown | Disabled |
| `--profile-deep` | Enable deep profiling with call graph (higher overhead) | Disabled |
| `--profile-format` | Output format: `text` or `json` | `text` |
## Profile Command
For more detailed profiling options, use the `profile` subcommand:
```bash
# Profile a query
praisonai profile query "What is 2+2?"
# With deep call tracing
praisonai profile query "Hello" --deep
# Limit functions shown
praisonai profile query "Test" --limit 20
# Save artifacts to directory
praisonai profile query "Test" --save ./profile_results
# JSON output
praisonai profile query "Test" --format json
```
### Profile Query Options
| Option | Description | Default |
|--------|-------------|---------|
| `--model, -m` | Model to use | Default model |
| `--stream/--no-stream` | Use streaming mode | No stream |
| `--deep` | Enable deep call tracing | Disabled |
| `--limit, -n` | Top N functions to show | 30 |
| `--sort, -s` | Sort by: `cumulative` or `tottime` | `cumulative` |
| `--show-callers` | Show caller functions | Disabled |
| `--show-callees` | Show callee functions | Disabled |
| `--save` | Save artifacts to path | None |
| `--format, -f` | Output format: `text` or `json` | `text` |
## Other Profile Commands
```bash
# Profile module import times
praisonai profile imports
# Profile CLI startup time
praisonai profile startup
# Run comprehensive profiling suite
praisonai profile suite
# Create performance snapshots for comparison
praisonai profile snapshot --baseline
praisonai profile snapshot current --compare
# Configure performance optimizations
praisonai profile optimize --show
```
## Output Format
### Text Report
```
======================================================================
PraisonAI Profile Report
======================================================================
Run ID: abc12345
Timestamp: 2024-01-02T12:00:00.000000Z
Method: cli_direct
Version: 3.0.1
## Execution Timeline
---------------------------------------------
Imports : 500.00 ms
Agent Init : 0.50 ms
Execution : 2000.00 ms
───────────────────────────────────────────
⏱ Time to First Response : 2500.50 ms
TOTAL : 2500.50 ms
## Timing Breakdown
----------------------------------------
Imports: 500.00 ms
Agent Init: 0.50 ms
Execution: 2000.00 ms
─────────────────────────────────────
TOTAL: 2500.50 ms
## Decision Trace (deep profile only)
----------------------------------------
Agent Config: default
Model: gpt-4
Streaming: disabled
Profile Layer: 2
## Top Functions by Cumulative Time
----------------------------------------------------------------------
Function Calls Cumul (ms)
----------------------------------------------------------------------
request 1 1800.00
send 1 1750.00
...
```
### JSON Report
```json
{
"schema_version": "1.0",
"run_id": "abc12345",
"timestamp": "2024-01-02T12:00:00.000000Z",
"invocation": {
"method": "cli_direct",
"flags": {"layer": 1, "stream": false, "model": null},
"praisonai_version": "3.0.1",
"python_version": "3.12.0"
},
"timing": {
"total_ms": 2500.5,
"imports_ms": 500.0,
"agent_init_ms": 0.5,
"execution_ms": 2000.0
},
"functions": [...],
"response_preview": "four"
}
```
## Profiling Layers
| Layer | Overhead | Features |
|-------|----------|----------|
| 0 | < 1ms | Wall-clock phases only |
| 1 | ~5% | + cProfile function stats |
| 2 | ~15% | + Call graph (callers/callees) |
- `--profile` uses Layer 1
- `--profile --profile-deep` uses Layer 2
## Performance Tips
1. **Use Layer 0 for production monitoring** - minimal overhead
2. **Use Layer 1 for development profiling** - good balance
3. **Use Layer 2 only for deep debugging** - higher overhead
## Observational Equivalence
The following commands produce equivalent profile reports:
```bash
# CLI direct
praisonai "prompt" --profile
# Profile command
praisonai profile query "prompt"
```
Both use the same unified profiling architecture and produce identical schema output.
-257
View File
@@ -1,257 +0,0 @@
# Profiling Module
PraisonAI provides a unified profiling module for programmatic performance analysis.
## Quick Start
```python
from praisonai.cli.execution import (
ExecutionRequest,
Profiler,
ProfilerConfig,
)
# Create execution request
request = ExecutionRequest(prompt="What is 2+2?")
# Configure profiler
config = ProfilerConfig(layer=1)
# Run with profiling
profiler = Profiler(config)
result, report = profiler.profile_sync(request)
# Access results
print(f"Output: {result.output}")
print(f"Total time: {report.timing.total_ms}ms")
```
## ExecutionRequest
Immutable request descriptor for execution.
```python
from praisonai.cli.execution import ExecutionRequest
# Basic request
request = ExecutionRequest(prompt="Hello")
# With model
request = ExecutionRequest(
prompt="Hello",
model="gpt-4",
agent_name="MyAgent",
)
# With streaming
request = ExecutionRequest(
prompt="Hello",
stream=True,
)
```
### Parameters
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `prompt` | str | Required | The prompt to execute |
| `agent_name` | str | "Assistant" | Name of the agent |
| `agent_instructions` | str | None | Custom instructions |
| `model` | str | None | Model to use |
| `stream` | bool | False | Enable streaming |
| `tools` | tuple | () | Tools to provide |
## ProfilerConfig
Configuration for profiling behavior.
```python
from praisonai.cli.execution import ProfilerConfig
# Layer 0: Wall-clock only (< 1ms overhead)
config = ProfilerConfig(layer=0)
# Layer 1: + cProfile stats (~5% overhead)
config = ProfilerConfig(layer=1)
# Layer 2: + Call graph (~15% overhead)
config = ProfilerConfig(
layer=2,
show_callers=True,
show_callees=True,
)
# From CLI flags
config = ProfilerConfig.from_flags(
profile=True,
deep=True,
network=False,
)
```
### Parameters
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `layer` | int | 1 | Profiling layer (0, 1, or 2) |
| `limit` | int | 30 | Max functions to show |
| `sort_by` | str | "cumulative" | Sort by: cumulative, time, calls |
| `show_callers` | bool | False | Show caller functions |
| `show_callees` | bool | False | Show callee functions |
| `track_network` | bool | False | Track network requests |
| `output_format` | str | "text" | Output format: text, json |
| `save_path` | str | None | Path to save artifacts |
## Profiler
The main profiler class that wraps execution.
```python
from praisonai.cli.execution import Profiler, ProfilerConfig, ExecutionRequest
config = ProfilerConfig(layer=1)
profiler = Profiler(config)
request = ExecutionRequest(prompt="Hello")
result, report = profiler.profile_sync(request, invocation_method="api")
```
### Methods
#### `profile_sync(request, invocation_method="profile_command")`
Profile synchronous execution.
**Parameters:**
- `request`: ExecutionRequest - The execution request
- `invocation_method`: str - How this was invoked (for report)
**Returns:** Tuple of (ExecutionResult, ProfileReport)
## ProfileReport
The canonical profile report with timing and statistics.
```python
# Access timing breakdown
print(f"Total: {report.timing.total_ms}ms")
print(f"Imports: {report.timing.imports_ms}ms")
print(f"Agent Init: {report.timing.agent_init_ms}ms")
print(f"Execution: {report.timing.execution_ms}ms")
# Access function stats (layer 1+)
if report.functions:
for func in report.functions[:5]:
print(f"{func.name}: {func.cumulative_time_ms}ms")
# Access call graph (layer 2)
if report.call_graph:
print(f"Edges: {len(report.call_graph.edges)}")
# Export formats
json_str = report.to_json()
text_str = report.to_text()
dict_data = report.to_dict()
```
### Schema Version
The report uses schema version `1.0`. Schema is frozen after release - new fields are always optional with defaults.
### Mandatory Fields
- `schema_version`: str
- `run_id`: str
- `timestamp`: str (ISO 8601)
- `invocation`: InvocationInfo
- `timing`: TimingBreakdown
- `response_preview`: str
### Optional Fields
- `functions`: List[FunctionStat] (layer 1+)
- `call_graph`: CallGraph (layer 2)
- `network`: List[RequestTiming] (if track_network)
## ExecutionResult
Result of execution with output and metadata.
```python
# Access output
print(result.output)
print(result.success)
print(result.run_id)
# Check for errors
if result.error:
print(f"Error: {result.error}")
# Duration
print(f"Duration: {result.duration_ms}ms")
```
## Data Bounds
Profile data is strictly bounded to prevent memory issues:
| Data Type | Maximum |
|-----------|---------|
| Function stats | 1000 |
| Call graph edges | 5000 |
| Network requests | 100 |
## Example: Full Profiling Workflow
```python
from praisonai.cli.execution import (
ExecutionRequest,
Profiler,
ProfilerConfig,
)
# Configure deep profiling
config = ProfilerConfig(
layer=2,
limit=50,
show_callers=True,
show_callees=True,
)
# Create request
request = ExecutionRequest(
prompt="Explain quantum computing in simple terms",
model="gpt-4",
)
# Profile execution
profiler = Profiler(config)
result, report = profiler.profile_sync(request)
# Print text report
print(report.to_text())
# Save JSON report
with open("profile.json", "w") as f:
f.write(report.to_json())
# Analyze timing
print(f"\nTiming Analysis:")
print(f" Import overhead: {report.timing.imports_ms / report.timing.total_ms * 100:.1f}%")
print(f" Execution time: {report.timing.execution_ms / report.timing.total_ms * 100:.1f}%")
# Top 5 hotspots
if report.functions:
print(f"\nTop 5 Hotspots:")
for func in report.functions[:5]:
print(f" {func.name}: {func.cumulative_time_ms:.2f}ms ({func.calls} calls)")
```
## Invariants
The profiling module guarantees:
1. **Output Equivalence**: Profiled execution produces identical output to non-profiled execution
2. **Schema Stability**: Schema version 1.0 is frozen; new fields are always optional
3. **Bounded Data**: All profile data respects maximum bounds
4. **Zero Overhead When Disabled**: No profiling code runs unless explicitly enabled
-53
View File
@@ -1,53 +0,0 @@
# Templates CLI
```bash
praisonai templates list
```
```bash
praisonai templates search video
```
```bash
praisonai templates info ai-video-editor
```
```bash
praisonai templates run ai-video-editor --input video.mp4 --output edited.mp4
```
```bash
praisonai templates run ai-video-editor --input video.mp4 --output edited.mp4 --offline
```
```bash
praisonai templates run ai-video-editor --input video.mp4 --output edited.mp4 --strict-tools
```
```bash
praisonai templates run ai-video-editor --input video.mp4 --tools ./my_tools.py
```
```bash
praisonai templates run ai-video-editor --input video.mp4 --tools-dir ./tools
```
```bash
praisonai templates run ai-video-editor --input video.mp4 --tools-source praisonai_tools.video
```
```bash
praisonai templates run ai-video-editor --input video.mp4 --no-template-tools-py
```
```bash
praisonai templates init my-project --template ai-video-editor
```
```bash
praisonai templates cache clear
```
```bash
praisonai templates cache list
```
-104
View File
@@ -1,104 +0,0 @@
# Templates Module
## Loading Templates
```python
from praisonai.templates.loader import TemplateLoader
loader = TemplateLoader()
# Load from local path
template = loader.load("./my-template")
# Load from URI
template = loader.load("ai-video-editor")
# Load offline
template = loader.load("ai-video-editor", offline=True)
# Check requirements
missing = loader.check_requirements(template)
```
## Template Discovery
```python
from praisonai.templates.discovery import TemplateDiscovery
discovery = TemplateDiscovery(
include_package=True,
include_defaults=True,
)
# Find template
result = discovery.find_template("ai-video-editor")
# List all templates
templates = discovery.list_templates()
```
## Template Config
```python
from praisonai.templates.loader import TemplateConfig
# Access template properties
template.name
template.description
template.version
template.author
template.requires
template.workflow_file
template.agents_file
template.path
```
## Tool Override Integration
```python
from praisonai.templates.tool_override import (
create_tool_registry_with_overrides,
resolve_tools,
)
# Build registry with template sources
tools_sources = template.requires.get("tools_sources", [])
registry = create_tool_registry_with_overrides(
tools_sources=tools_sources,
template_dir=str(template.path),
include_defaults=True,
)
# Resolve tool names
tools = resolve_tools(["shell_tool"], registry=registry)
```
## TEMPLATE.yaml Schema
```yaml
name: my-template
version: "1.0.0"
description: Template description
requires:
tools: [shell_tool]
packages: [praisonai-tools]
env: [OPENAI_API_KEY]
tools_sources:
- praisonai_tools.video
- ./local_tools.py
workflow: workflow.yaml
agents: agents.yaml
config:
input:
type: string
required: true
output:
type: string
defaults:
preset: podcast
```
-65
View File
@@ -1,65 +0,0 @@
# Tools CLI
```bash
praisonai tools list
```
```bash
praisonai tools info shell_tool
```
```bash
praisonai tools info internet_search
```
```bash
praisonai tools search "web"
```
```bash
praisonai tools search "file"
```
```bash
praisonai tools doctor
```
```bash
praisonai tools doctor --json
```
```bash
praisonai tools resolve shell_tool
```
```bash
praisonai tools resolve internet_search --template ai-video-editor
```
```bash
praisonai tools resolve my_tool --tools ./my_tools.py
```
```bash
praisonai tools resolve my_tool --tools-dir ./tools
```
```bash
praisonai tools discover
```
```bash
praisonai tools discover --include praisonai_tools.video
```
```bash
praisonai tools discover --entrypoints
```
```bash
praisonai tools show-sources
```
```bash
praisonai tools show-sources --template ai-video-editor
```
-90
View File
@@ -1,90 +0,0 @@
# Tools Module
## Tool Registry
```python
from praisonai.templates.tool_override import (
create_tool_registry_with_overrides,
resolve_tools,
)
# Create registry with defaults
registry = create_tool_registry_with_overrides(include_defaults=True)
# Create registry with overrides
registry = create_tool_registry_with_overrides(
override_files=["my_tools.py"],
override_dirs=["./tools"],
include_defaults=True,
)
# Create registry with tools_sources
registry = create_tool_registry_with_overrides(
tools_sources=["praisonai_tools.video"],
template_dir="./my-template",
include_defaults=True,
)
# Resolve tool names to callables
tools = resolve_tools(
["shell_tool", "internet_search"],
registry=registry,
)
```
## Tool Override Loader
```python
from praisonai.templates.tool_override import ToolOverrideLoader
loader = ToolOverrideLoader()
# Load from file
tools = loader.load_from_file("my_tools.py")
# Load from directory
tools = loader.load_from_directory("./tools")
# Load from module
tools = loader.load_from_module("praisonai_tools.video")
# Get default tool directories
dirs = loader.get_default_tool_dirs()
```
## Resolution Order
1. CLI `--tools` files (highest priority)
2. CLI `--tools-dir` directories
3. Template `tools_sources` (from TEMPLATE.yaml)
4. Template-local `tools.py`
5. Default dirs (`~/.praison/tools`, `~/.config/praison/tools`)
6. Package discovery (`praisonai_tools`)
7. Built-in tools (lowest priority)
## Creating Custom Tools
```python
from praisonai_tools import tool
@tool
def my_custom_tool(query: str) -> str:
"""My custom tool description.
Args:
query: The search query
"""
return f"Result for: {query}"
```
## Template tools_sources
```yaml
# TEMPLATE.yaml
requires:
tools: [shell_tool]
tools_sources:
- praisonai_tools.video
- ./local_tools.py
- ./tools_dir/
```
+158
View File
@@ -0,0 +1,158 @@
"""
Agent-Centric API Example
Demonstrates the new consolidated feature parameters for PraisonAI Agents.
These params provide a cleaner, more intuitive API with progressive disclosure.
New Consolidated Params:
- memory: Union[bool, MemoryConfig, MemoryManager] - Memory and session management
- knowledge: Union[bool, List[str], KnowledgeConfig, KnowledgeBase] - RAG and retrieval
- planning: Union[bool, PlanningConfig] - Planning mode
- reflection: Union[bool, ReflectionConfig] - Self-reflection
- guardrails: Union[bool, Callable, GuardrailConfig] - Safety validation
- web: Union[bool, WebConfig] - Web search and fetch
- context: Union[bool, ManagerConfig, ContextManager] - Context management
"""
from praisonaiagents import (
Agent,
# Config classes for progressive disclosure
MemoryConfig,
KnowledgeConfig,
PlanningConfig,
ReflectionConfig,
GuardrailConfig,
WebConfig,
)
def example_simple_enable():
"""Example 1: Simple boolean enable (easiest)"""
print("\n=== Example 1: Simple Boolean Enable ===")
agent = Agent(
instructions="You are a helpful assistant",
memory=True, # Enable memory with defaults
reflection=True, # Enable self-reflection
web=True, # Enable web search + fetch
)
print(f"Agent created with memory={agent._memory_instance is not None}")
print(f" self_reflect={agent.self_reflect}")
print(f" web_search={agent.web_search}")
def example_with_config():
"""Example 2: Using config objects (more control)"""
print("\n=== Example 2: Using Config Objects ===")
agent = Agent(
instructions="You are a research assistant",
memory=MemoryConfig(
backend="file",
user_id="researcher_001",
auto_memory=True,
),
reflection=ReflectionConfig(
min_iterations=1,
max_iterations=5,
llm="gpt-4o-mini",
),
web=WebConfig(
search=True,
fetch=False, # Only search, no full page fetch
max_results=3,
),
)
print(f"Agent created with user_id={agent.user_id}")
print(f" {agent.max_reflect}")
print(f" web_search={agent.web_search}, web_fetch={agent.web_fetch}")
def example_guardrails():
"""Example 3: Guardrails with callable or config"""
print("\n=== Example 3: Guardrails ===")
# Option A: Direct callable
def my_validator(output):
"""Simple validator that checks response length."""
is_valid = len(output.raw) < 1000
return (is_valid, output if is_valid else "Response too long")
agent_a = Agent(
instructions="Be concise",
guardrails=my_validator, # Direct callable
)
print(f"Agent A: guardrail is callable = {callable(agent_a.guardrail)}")
# Option B: Using GuardrailConfig
agent_b = Agent(
instructions="Be helpful and safe",
guardrails=GuardrailConfig(
llm_validator="Ensure the response is helpful, accurate, and safe",
max_retries=3,
),
)
print(f"Agent B: guardrail = {agent_b.guardrail}")
print(f" max_retries = {agent_b.max_guardrail_retries}")
def example_planning():
"""Example 4: Planning mode"""
print("\n=== Example 4: Planning Mode ===")
agent = Agent(
instructions="You are a project planner",
planning=PlanningConfig(
reasoning=True,
read_only=True, # Plan mode - only read operations
),
)
print(f"Agent created with planning={agent.planning}")
print(f" plan_mode={agent.plan_mode}")
print(f" planning_reasoning={agent.planning_reasoning}")
def example_backward_compatible():
"""Example 5: Using new consolidated params"""
print("\n=== Example 5: Consolidated Params ===")
# New consolidated params
agent = Agent(
instructions="Test agent",
reflection=True,
web=WebConfig(search=True, fetch=False),
)
print(f"Consolidated params work: self_reflect={agent.self_reflect}")
print(f" web_search={agent.web_search}, web_fetch={agent.web_fetch}")
def example_real_chat():
"""Example 6: Real chat with new API"""
print("\n=== Example 6: Real Chat ===")
agent = Agent(
instructions="You are a helpful math tutor. Be concise.",
reflection=False,
web=False,
verbose=False,
)
response = agent.start("What is the square root of 144?")
print(f"Response: {response}")
if __name__ == "__main__":
print("=" * 60)
print("PraisonAI Agent-Centric API Examples")
print("=" * 60)
example_simple_enable()
example_with_config()
example_guardrails()
example_planning()
example_backward_compatible()
example_real_chat()
print("\n" + "=" * 60)
print("All examples completed!")
print("=" * 60)
@@ -0,0 +1,43 @@
"""
Basic Caching Configuration Example
Demonstrates using caching for response and prompt caching.
"""
import os
from praisonaiagents import Agent, CachingConfig
# Ensure API key is set from environment
assert os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY must be set"
# Default caching (enabled)
agent_cached = Agent(
instructions="You are a helpful assistant.",
caching=True,
)
# Disable caching
agent_no_cache = Agent(
instructions="You are a helpful assistant.",
caching=False,
)
# Custom caching configuration
agent_custom = Agent(
instructions="You are a helpful assistant.",
caching=CachingConfig(
enabled=True,
prompt_caching=True, # Enable prompt caching for supported providers
),
)
if __name__ == "__main__":
print("Testing CachingConfig...")
print(f"Cached agent cache: {agent_cached.cache}")
print(f"No-cache agent cache: {agent_no_cache.cache}")
print(f"Custom agent cache: {agent_custom.cache}")
result = agent_cached.chat("What is caching?")
print(f"Result: {result}")
print("\nCachingConfig tests passed!")
+72
View File
@@ -128,6 +128,78 @@ def main():
budget_dict = budgeter.to_dict()
print(f"Budget as dict: {budget_dict}")
# Example 6: Overflow handling strategies overview
print("\n6. Overflow Handling Strategies")
print("-" * 40)
strategies = [
("truncate", "Remove oldest messages first", "Fast, simple", "Loses early context"),
("sliding_window", "Keep N most recent messages", "Preserves recent", "Loses early context"),
("prune_tools", "Truncate old tool outputs", "Keeps messages", "May lose tool details"),
("summarize", "Replace old messages with summary", "Preserves meaning", "Slower, uses API"),
("smart", "Combine strategies intelligently", "Best balance", "More complex"),
]
print(f"{'Strategy':<15} | {'Description':<30} | {'Pros':<18} | {'Cons'}")
print("-" * 90)
for name, desc, pros, cons in strategies:
print(f"{name:<15} | {desc:<30} | {pros:<18} | {cons}")
# Example 7: Simulated overflow playbook
print("\n7. Overflow Playbook (Thresholds → Actions)")
print("-" * 40)
playbook = [
(0.70, "INFO", "Monitor usage, no action needed"),
(0.80, "NOTICE", "Consider optimization soon"),
(0.90, "WARNING", "Trigger auto-compact if enabled"),
(0.95, "CRITICAL", "Aggressive optimization required"),
(1.00, "OVERFLOW", "Immediate truncation to prevent API error"),
]
print(f"{'Usage':<8} | {'Level':<10} | {'Action'}")
print("-" * 60)
for threshold, level, action in playbook:
print(f"{threshold:>6.0%} | {level:<10} | {action}")
# Example 8: Strategy comparison table
print("\n8. Strategy Comparison")
print("-" * 40)
print("Strategy | Preserves | Loses | Best For")
print("-" * 70)
print("truncate | Recent msgs | Old msgs | Simple chats")
print("sliding_window | Last N msgs | Earlier msgs | Long conversations")
print("prune_tools | All messages | Tool details | Tool-heavy agents")
print("summarize | Context meaning| Exact wording | Important history")
print("smart | Balanced | Minimal | Production use")
# Example 9: Default strategy for interactive mode
print("\n9. Default Strategy: Interactive Mode")
print("-" * 40)
print("Interactive mode default: context=False (zero overhead)")
print("To enable context management in interactive mode:")
print(" praisonai chat --context # Enable with defaults")
print(" Or in code:")
print(" agent = Agent(instructions='...', context=True)")
print("")
print("When enabled, defaults are:")
print(" - auto_compact: True")
print(" - compact_threshold: 0.8 (80% usage)")
print(" - strategy: smart")
print(" - output_reserve: model-specific (8K-16K)")
# Example 10: Default strategy for auto-agents mode
print("\n10. Default Strategy: Auto-Agents Mode")
print("-" * 40)
print("Auto-agents mode default: context=False (zero overhead)")
print("To enable for multi-agent workflows:")
print(" from praisonaiagents import Agents")
print(" agents = Agents(agents=[...], context=True)")
print("")
print("Recommended config for long tasks:")
print(" context=ManagerConfig(")
print(" auto_compact=True,")
print(" compact_threshold=0.7, # Earlier trigger for safety")
print(" strategy='smart',")
print(" output_reserve=16384, # Larger reserve for complex outputs")
print(" )")
print("\n" + "=" * 60)
print("✓ Context budgeting examples complete!")
print("=" * 60)
+5 -5
View File
@@ -2,14 +2,14 @@
"""
Multi-Agent Context Isolation Example
Demonstrates how to use multi-agent context with PraisonAIAgents via context= param,
Demonstrates how to use multi-agent context with Agents via context= param,
and also shows low-level MultiAgentLedger usage for advanced scenarios.
Agent-Centric Quick Start:
from praisonaiagents import Agent, PraisonAIAgents
from praisonaiagents import Agent, Agents
from praisonaiagents.context import ManagerConfig
agents = PraisonAIAgents(
agents = Agents(
agents=[agent1, agent2],
context=ManagerConfig(policy="isolated"), # or "shared"
)
@@ -17,7 +17,7 @@ Agent-Centric Quick Start:
import os
import tempfile
from praisonaiagents import Agent, PraisonAIAgents
from praisonaiagents import Agent, Agents
from praisonaiagents.context import (
ManagerConfig,
MultiAgentLedger,
@@ -49,7 +49,7 @@ def agent_centric_example():
# Multi-agent setup with isolated context policy
# (each agent maintains its own context)
print("Created agents with context=True")
print("Use PraisonAIAgents(agents=[...], context=ManagerConfig(policy='isolated'))")
print("Use Agents(agents=[...], context=ManagerConfig(policy='isolated'))")
print("for multi-agent context isolation")
@@ -13,10 +13,10 @@ from praisonaiagents import Agent
from praisonaiagents.context import (
ContextBudgeter,
ContextLedgerManager,
ContextMonitor,
get_optimizer,
OptimizerStrategy,
estimate_messages_tokens,
format_percent,
)
@@ -37,7 +37,7 @@ def main():
# Get budget info
budget = budgeter.allocate()
print(f"\nModel: gpt-4o-mini")
print("\nModel: gpt-4o-mini")
print(f"Context budget: {budget.usable:,} tokens")
print(f"Output reserve: {budget.output_reserve:,} tokens")
@@ -76,7 +76,7 @@ def main():
current_tokens = ledger.get_total()
utilization = budgeter.get_utilization(current_tokens)
print(f"[Context: {current_tokens} tokens, {utilization:.1%} used]")
print(f"[Context: {current_tokens} tokens, {format_percent(utilization)} used]")
# Check if optimization needed
if utilization > 0.8:
@@ -101,7 +101,7 @@ def main():
final_utilization = budgeter.get_utilization(final_tokens)
print(f"Total tokens used: {final_tokens:,}")
print(f"Context utilization: {final_utilization:.1%}")
print(f"Context utilization: {format_percent(final_utilization)}")
print(f"Messages in conversation: {len(conversation)}")
print(f"Remaining capacity: {budgeter.get_remaining(final_tokens):,} tokens")
+2 -2
View File
@@ -156,7 +156,7 @@ def example_5_multi_agent():
print("Example 5: Multi-Agent Context Management")
print("=" * 60)
from praisonaiagents import PraisonAIAgents, Task
from praisonaiagents import Agents, Task
config = ManagerConfig(
auto_compact=True,
@@ -181,7 +181,7 @@ def example_5_multi_agent():
task2 = Task(description="Write a summary", agent=writer)
# Create multi-agent system with context management
agents = PraisonAIAgents(
agents = Agents(
agents=[researcher, writer],
tasks=[task1, task2],
context=config, # Shared context config
@@ -0,0 +1,48 @@
"""
Basic Execution Configuration Example
Demonstrates using execution presets for quick configuration.
"""
import os
from praisonaiagents import Agent
# Ensure API key is set from environment
assert os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY must be set"
# Default execution (max_iter=20, max_retry_limit=2)
agent_default = Agent(
instructions="You are a helpful assistant.",
)
# Fast execution - fewer iterations, quick responses
agent_fast = Agent(
instructions="You are a helpful assistant.",
execution="fast",
)
# Thorough execution - more iterations, comprehensive responses
agent_thorough = Agent(
instructions="You are a research assistant.",
execution="thorough",
)
# Unlimited execution - for complex tasks
agent_unlimited = Agent(
instructions="You are a complex problem solver.",
execution="unlimited",
)
if __name__ == "__main__":
print("Testing execution presets...")
print(f"Default agent max_iter: {agent_default.max_iter}")
print(f"Fast agent max_iter: {agent_fast.max_iter}")
print(f"Thorough agent max_iter: {agent_thorough.max_iter}")
print(f"Unlimited agent max_iter: {agent_unlimited.max_iter}")
# Test fast agent
print("\n--- Fast Execution ---")
result = agent_fast.chat("What is the capital of France?")
print(f"Result: {result}")
print("\nAll execution preset tests passed!")
@@ -0,0 +1,45 @@
"""
ExecutionConfig Example
Demonstrates using ExecutionConfig for fine-grained control over execution limits.
"""
import os
from praisonaiagents import Agent, ExecutionConfig
# Ensure API key is set from environment
assert os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY must be set"
# Custom execution configuration
agent = Agent(
instructions="You are a helpful assistant.",
execution=ExecutionConfig(
max_iter=30,
max_rpm=100,
max_execution_time=60,
max_retry_limit=3,
),
)
# Rate-limited agent for API quota management
agent_rate_limited = Agent(
instructions="You are a helpful assistant.",
execution=ExecutionConfig(
max_iter=20,
max_rpm=10, # Only 10 requests per minute
),
)
if __name__ == "__main__":
print("Testing ExecutionConfig...")
print(f"Custom agent max_iter: {agent.max_iter}")
print(f"Custom agent max_rpm: {agent.max_rpm}")
print(f"Custom agent max_execution_time: {agent.max_execution_time}")
print(f"Custom agent max_retry_limit: {agent.max_retry_limit}")
# Test custom agent
print("\n--- Custom Execution ---")
result = agent.chat("What is machine learning?")
print(f"Result: {result}")
print("\nExecutionConfig tests passed!")
+8 -8
View File
@@ -4,7 +4,7 @@ Fixed example demonstrating proper guardrail usage with PraisonAI Agents.
This addresses the issues reported in issue #875.
"""
from praisonaiagents import Agent, Task, GuardrailResult, PraisonAIAgents, TaskOutput
from praisonaiagents import Agent, Task, GuardrailResult, Agents, TaskOutput
from typing import Tuple, Any
import trafilatura
@@ -67,7 +67,7 @@ agent = Agent(
goal="Summarize web content concisely",
instructions="You are a helpful assistant that summarizes web content",
llm="gemini/gemini-2.5-flash-lite-preview-06-17",
self_reflect=False,
reflection=False,
verbose=True,
tools=[get_url_context] # FIX: tools must be a list, not a single function
)
@@ -77,7 +77,7 @@ task_with_guardrailresult = Task(
name="summarise article with GuardrailResult",
description="get the context of this url: https://blog.google/technology/ai/dolphingemma/ and produce a summary below 500 characters",
agent=agent,
guardrail=validate_length_guardrailresult, # Using GuardrailResult
guardrails=validate_length_guardrailresult, # Using GuardrailResult
expected_output="summary of the article below 500 characters",
max_retries=3 # Will retry up to 3 times if guardrail fails
)
@@ -87,7 +87,7 @@ task_with_tuple = Task(
name="summarise article with tuple",
description="get the context of this url: https://blog.google/technology/ai/dolphingemma/ and produce a summary below 500 characters",
agent=agent,
guardrail=validate_length_tuple, # Using Tuple[bool, Any]
guardrails=validate_length_tuple, # Using Tuple[bool, Any]
expected_output="summary of the article below 500 characters",
max_retries=3
)
@@ -97,13 +97,13 @@ task_with_llm_guardrail = Task(
name="summarise with LLM guardrail",
description="get the context of this url: https://blog.google/technology/ai/dolphingemma/ and produce a summary",
agent=agent,
guardrail="Ensure the summary is professional, factual, and between 100-500 characters",
guardrails="Ensure the summary is professional, factual, and between 100-500 characters",
expected_output="professional summary of the article"
)
# Run with GuardrailResult example
print("=== Running with GuardrailResult guardrail ===")
agents_gr = PraisonAIAgents(
agents_gr = Agents(
agents=[agent],
tasks=[task_with_guardrailresult]
)
@@ -113,7 +113,7 @@ agents_gr = PraisonAIAgents(
# Run with Tuple example
print("\n=== Running with Tuple[bool, Any] guardrail ===")
agents_tuple = PraisonAIAgents(
agents_tuple = Agents(
agents=[agent],
tasks=[task_with_tuple]
)
@@ -123,7 +123,7 @@ agents_tuple = PraisonAIAgents(
# Run with LLM guardrail example
print("\n=== Running with LLM-based guardrail ===")
agents_llm = PraisonAIAgents(
agents_llm = Agents(
agents=[agent],
tasks=[task_with_llm_guardrail]
)
@@ -0,0 +1,30 @@
"""
Basic Guardrails Configuration Example
Demonstrates using guardrails for output validation.
"""
import os
from praisonaiagents import Agent
# Ensure API key is set from environment
assert os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY must be set"
def validate_length(output):
"""Validate that output is not too short."""
if len(str(output.raw)) < 10:
return False, "Response too short"
return True, output
# Enable guardrails with a validator function
agent = Agent(
instructions="You are a helpful assistant.",
guardrails=validate_length,
)
if __name__ == "__main__":
print("Testing guardrails...")
result = agent.chat("Explain what Python is.")
print(f"Result: {result}")
print("\nGuardrails tests passed!")
@@ -0,0 +1,43 @@
"""
GuardrailConfig Example
Demonstrates using GuardrailConfig for fine-grained control.
"""
import os
from praisonaiagents import Agent, GuardrailConfig
# Ensure API key is set from environment
assert os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY must be set"
def validate_no_code(output):
"""Validate that output doesn't contain code blocks."""
if "```" in str(output.raw):
return False, "Response should not contain code blocks"
return True, output
# Custom guardrail configuration
agent = Agent(
instructions="You are a helpful assistant. Explain concepts without code.",
guardrails=GuardrailConfig(
validator=validate_no_code,
max_retries=3,
on_fail="retry",
),
)
# LLM-based validation
agent_llm = Agent(
instructions="You are a helpful assistant.",
guardrails=GuardrailConfig(
llm_validator="Ensure the response is helpful, accurate, and professional.",
max_retries=2,
),
)
if __name__ == "__main__":
print("Testing GuardrailConfig...")
result = agent.chat("What is a variable in programming?")
print(f"Result: {result}")
print("\nGuardrailConfig tests passed!")
@@ -14,7 +14,7 @@ Usage:
from praisonai_tools.observability import obs
from praisonai_tools.observability.base import SpanKind
from praisonaiagents import Agent, PraisonAIAgents
from praisonaiagents import Agent, Agents
# Initialize observability
obs.init()
+2 -2
View File
@@ -13,7 +13,7 @@ Features demonstrated:
import sys
sys.path.insert(0, '/Users/praison/praisonai-package/src/praisonai-agents')
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
def get_weather(city: str) -> str:
@@ -106,7 +106,7 @@ def example_multi_agent():
agent=writer
)
agents = PraisonAIAgents(
agents = Agents(
agents=[researcher, writer],
tasks=[task1, task2],
process="sequential",
+297
View File
@@ -0,0 +1,297 @@
"""
Agent-Centric Autonomy Example.
Demonstrates the new agent-centric autonomy features in PraisonAI Agents:
- Agent(autonomy=True) for enabling autonomy
- Progressive escalation stages (DIRECT, HEURISTIC, PLANNED, AUTONOMOUS)
- Doom loop detection and recovery
- Verification hooks
- Subagent delegation
"""
from praisonaiagents import Agent
def example_basic_autonomy():
"""Basic autonomy example - simple question."""
print("\n" + "="*60)
print("Example 1: Basic Autonomy - Simple Question")
print("="*60)
agent = Agent(
instructions="You are a helpful coding assistant.",
autonomy=True,
)
# Analyze a simple question
prompt = "What is Python?"
signals = agent.analyze_prompt(prompt)
stage = agent.get_recommended_stage(prompt)
print(f"Prompt: {prompt}")
print(f"Detected signals: {signals}")
print(f"Recommended stage: {stage}")
# For simple questions, stage should be "direct"
assert stage == "direct", f"Expected 'direct', got '{stage}'"
print("✓ Simple question correctly identified as DIRECT stage")
def example_file_reference():
"""File reference detection example."""
print("\n" + "="*60)
print("Example 2: File Reference Detection")
print("="*60)
agent = Agent(
instructions="You are a code reviewer.",
autonomy=True,
)
prompt = "Read the main.py file and explain what it does"
signals = agent.analyze_prompt(prompt)
stage = agent.get_recommended_stage(prompt)
print(f"Prompt: {prompt}")
print(f"Detected signals: {signals}")
print(f"Recommended stage: {stage}")
assert "file_references" in signals, "Expected file_references signal"
assert stage == "heuristic", f"Expected 'heuristic', got '{stage}'"
print("✓ File reference correctly identified as HEURISTIC stage")
def example_edit_intent():
"""Edit intent detection example."""
print("\n" + "="*60)
print("Example 3: Edit Intent Detection")
print("="*60)
agent = Agent(
instructions="You are a code editor.",
autonomy=True,
)
prompt = "Edit the config.py file to add logging support"
signals = agent.analyze_prompt(prompt)
stage = agent.get_recommended_stage(prompt)
print(f"Prompt: {prompt}")
print(f"Detected signals: {signals}")
print(f"Recommended stage: {stage}")
assert "edit_intent" in signals, "Expected edit_intent signal"
assert stage == "planned", f"Expected 'planned', got '{stage}'"
print("✓ Edit intent correctly identified as PLANNED stage")
def example_multi_step_task():
"""Multi-step task detection example."""
print("\n" + "="*60)
print("Example 4: Multi-Step Task Detection")
print("="*60)
agent = Agent(
instructions="You are a software architect.",
autonomy=True,
)
prompt = """First analyze the codebase structure,
then refactor the authentication module,
and finally add comprehensive tests."""
signals = agent.analyze_prompt(prompt)
stage = agent.get_recommended_stage(prompt)
print(f"Prompt: {prompt[:50]}...")
print(f"Detected signals: {signals}")
print(f"Recommended stage: {stage}")
assert "multi_step" in signals, "Expected multi_step signal"
assert stage == "autonomous", f"Expected 'autonomous', got '{stage}'"
print("✓ Multi-step task correctly identified as AUTONOMOUS stage")
def example_custom_config():
"""Custom autonomy configuration example."""
print("\n" + "="*60)
print("Example 5: Custom Autonomy Configuration")
print("="*60)
agent = Agent(
instructions="You are a careful code reviewer.",
autonomy={
"max_iterations": 50,
"doom_loop_threshold": 5,
"auto_escalate": True,
"checkpoint_on_write": True,
},
)
print(f"Autonomy enabled: {agent.autonomy_enabled}")
print(f"Max iterations: {agent.autonomy_config.get('max_iterations')}")
print(f"Doom loop threshold: {agent.autonomy_config.get('doom_loop_threshold')}")
print(f"Checkpoint on write: {agent.autonomy_config.get('checkpoint_on_write')}")
assert agent.autonomy_enabled == True
assert agent.autonomy_config.get("max_iterations") == 50
print("✓ Custom configuration applied correctly")
def example_doom_loop_detection():
"""Doom loop detection example."""
print("\n" + "="*60)
print("Example 6: Doom Loop Detection")
print("="*60)
agent = Agent(
instructions="You are a test agent.",
autonomy={"doom_loop_threshold": 3},
)
# Simulate repeated actions
print("Simulating repeated actions...")
for i in range(3):
agent._record_action("read_file", {"path": "test.py"}, "content", True)
print(f" Action {i+1}: read_file test.py")
is_doom_loop = agent._is_doom_loop()
print(f"Doom loop detected: {is_doom_loop}")
assert is_doom_loop == True, "Expected doom loop to be detected"
print("✓ Doom loop correctly detected after repeated actions")
# Reset and verify
agent._reset_doom_loop()
is_doom_loop_after_reset = agent._is_doom_loop()
print(f"After reset: {is_doom_loop_after_reset}")
assert is_doom_loop_after_reset == False
print("✓ Doom loop correctly reset")
def example_verification_hooks():
"""Verification hooks example."""
print("\n" + "="*60)
print("Example 7: Verification Hooks")
print("="*60)
# Create a simple mock verification hook
class MockTestRunner:
name = "pytest"
def run(self, context=None):
return {"success": True, "output": "All 10 tests passed"}
agent = Agent(
instructions="You are a test-driven developer.",
autonomy=True,
verification_hooks=[MockTestRunner()],
)
print(f"Verification hooks registered: {len(agent._verification_hooks)}")
# Run verification hooks
results = agent._run_verification_hooks()
print(f"Verification results: {results}")
assert len(results) == 1
assert results[0]["success"] == True
print("✓ Verification hooks executed correctly")
def example_subagent_delegation():
"""Subagent delegation example."""
print("\n" + "="*60)
print("Example 8: Subagent Delegation")
print("="*60)
agent = Agent(
instructions="You are a project manager.",
autonomy=True,
)
# Check delegate method exists
assert hasattr(agent, "delegate"), "Agent should have delegate method"
assert callable(agent.delegate), "delegate should be callable"
# Check _create_subagent method
subagent = agent._create_subagent("explorer")
print(f"Created subagent: {subagent.name}")
print(f"Subagent instructions: {subagent.instructions[:50]}...")
assert "subagent_explorer" in subagent.name
print("✓ Subagent delegation works correctly")
def example_explorer_profile():
"""Explorer profile example."""
print("\n" + "="*60)
print("Example 9: Explorer Profile (Read-Only)")
print("="*60)
from praisonaiagents.agents.profiles import BUILTIN_PROFILES, get_profile
explorer = get_profile("explorer")
print(f"Profile name: {explorer.name}")
print(f"Description: {explorer.description[:60]}...")
print(f"Tools: {explorer.tools}")
print(f"Read-only metadata: {explorer.metadata.get('read_only')}")
# Verify explorer is read-only
write_tools = {"write_file", "delete_file", "bash", "shell"}
has_write_tools = any(tool in write_tools for tool in explorer.tools)
assert not has_write_tools, "Explorer should not have write tools"
assert explorer.metadata.get("read_only") == True
print("✓ Explorer profile is correctly read-only")
def example_run_autonomous_method():
"""run_autonomous method example (without actual LLM call)."""
print("\n" + "="*60)
print("Example 10: run_autonomous Method")
print("="*60)
agent = Agent(
instructions="You are a helpful assistant.",
autonomy=True,
)
# Verify method exists
assert hasattr(agent, "run_autonomous"), "Agent should have run_autonomous method"
assert callable(agent.run_autonomous), "run_autonomous should be callable"
print("✓ run_autonomous method exists and is callable")
print(" (Skipping actual execution to avoid LLM API calls)")
def main():
"""Run all examples."""
print("\n" + "#"*60)
print("# Agent-Centric Autonomy Examples")
print("#"*60)
try:
example_basic_autonomy()
example_file_reference()
example_edit_intent()
example_multi_step_task()
example_custom_config()
example_doom_loop_detection()
example_verification_hooks()
example_subagent_delegation()
example_explorer_profile()
example_run_autonomous_method()
print("\n" + "="*60)
print("All examples completed successfully! ✓")
print("="*60)
except Exception as e:
print(f"\n❌ Error: {e}")
raise
if __name__ == "__main__":
main()
@@ -5,7 +5,7 @@ This example demonstrates hierarchical agent patterns using PraisonAI
for structured task delegation and coordinated workflows.
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from praisonaiagents.tools import internet_search
print("=== AutoAgents Hierarchical Generation Example ===\n")
@@ -17,9 +17,7 @@ manager_agent = Agent(
goal="Coordinate hierarchical task execution and manage team workflow",
backstory="Experienced manager who excels at breaking down complex projects and coordinating teams",
tools=[internet_search],
allow_delegation=True,
verbose=True
)
allow_delegation=True)
# Create specialist agent for execution
research_agent = Agent(
@@ -27,9 +25,7 @@ research_agent = Agent(
role="Market Research Analyst",
goal="Conduct thorough market research and analysis",
backstory="Expert researcher skilled in market analysis and competitive intelligence",
tools=[internet_search],
verbose=True
)
tools=[internet_search])
# Hierarchical task structure
planning_task = Task(
@@ -58,11 +54,10 @@ execution_task = Task(
)
# Run hierarchical workflow
agents_system = PraisonAIAgents(
agents_system = Agents(
agents=[manager_agent, research_agent],
tasks=[planning_task, execution_task],
process="sequential",
verbose=True
process="sequential", output="verbose"
)
print("Starting hierarchical agent generation...")
+9 -10
View File
@@ -5,7 +5,7 @@ This example demonstrates a comprehensive code agent that can write, analyze, de
It includes self-reflection for improving code quality and multiple specialized tools.
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from praisonaiagents.tools import CodeInterpreter
# Initialize code execution tool
@@ -32,9 +32,9 @@ code_writer = Agent(
goal="Write clean, efficient, and well-documented Python code",
backstory="You are a senior Python developer with 10+ years of experience in writing production-quality code.",
instructions="Write Python code that follows PEP 8 standards, includes proper error handling, and has clear documentation.",
self_reflect=True,
min_reflect=2,
max_reflect=5
reflection=True,
)
code_reviewer = Agent(
@@ -84,11 +84,10 @@ def demonstrate_code_agent():
)
# Create workflow
workflow = PraisonAIAgents(
workflow = Agents(
agents=[code_writer, code_reviewer, code_executor],
tasks=[write_task, review_task, execute_task],
process="sequential",
verbose=True
process="sequential", output="verbose"
)
# Run the workflow
@@ -108,9 +107,9 @@ general_code_agent = Agent(
4. Refactor code for better performance
5. Add tests and documentation""",
tools=[code_interpreter, analyze_code_complexity],
self_reflect=True,
min_reflect=1,
max_reflect=3
reflection=True,
)
if __name__ == "__main__":
+4 -6
View File
@@ -31,10 +31,8 @@ def demonstrate_basic_context_generation():
# Create a ContextAgent using the factory function
context_agent = create_context_agent(
llm="gpt-5-nano",
name="Basic Context Engineer",
verbose=True
)
llm="gpt-4o-mini",
name="Basic Context Engineer")
# Example 1: Analyze codebase patterns
print("\n📁 Example 1: Analyzing Codebase Patterns")
@@ -133,7 +131,7 @@ def demonstrate_context_agent_as_tool():
print("=" * 60)
# Create ContextAgent
context_agent = create_context_agent(llm="gpt-5-nano")
context_agent = create_context_agent(llm="gpt-4o-mini")
# Example scenario: Product planning workflow
project_path = str(project_root)
@@ -186,7 +184,7 @@ def show_context_engineering_benefits():
# Context Engineering approach
print("\n✅ Context Engineering Approach:")
context_agent = create_context_agent(llm="gpt-5-nano")
context_agent = create_context_agent(llm="gpt-4o-mini")
# Generate comprehensive context
analysis = context_agent.analyze_codebase_patterns(str(project_root))
+5 -5
View File
@@ -4,7 +4,7 @@ from praisonaiagents.tools import duckduckgo
agent = Agent(instructions="You are a Image Analysis Agent", tools=[duckduckgo])
agent.start("I want to go London next week, find me a good hotel and flight")
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
# Create Image Analysis Agent
image_agent = Agent(
@@ -13,8 +13,8 @@ image_agent = Agent(
goal="Analyze images and videos to extract meaningful information",
backstory="""You are an expert in computer vision and image analysis.
You excel at describing images, detecting objects, and understanding visual content.""",
llm="gpt-5-nano",
self_reflect=False
llm="gpt-4o-mini",
reflection=False
)
# 1. Task with Image URL
@@ -35,8 +35,8 @@ task2 = Task(
images=["image.jpg"]
)
# Create PraisonAIAgents instance
agents = PraisonAIAgents(
# Create Agents instance
agents = Agents(
agents=[image_agent],
tasks=[task1, task2],
process="sequential",
@@ -4,7 +4,7 @@ from praisonaiagents.tools import duckduckgo
agent = Agent(instructions="You are a Image Analysis Agent", tools=[duckduckgo])
agent.start("I want to go London next week, find me a good hotel and flight")
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
# Create Image Analysis Agent
image_agent = Agent(
@@ -13,8 +13,8 @@ image_agent = Agent(
goal="Analyze images and videos to extract meaningful information",
backstory="""You are an expert in computer vision and image analysis.
You excel at describing images, detecting objects, and understanding visual content.""",
llm="gpt-5-nano",
self_reflect=False
llm="gpt-4o-mini",
reflection=False
)
# 1. Task with Image URL
@@ -35,8 +35,8 @@ task2 = Task(
images=["image.jpg"]
)
# Create PraisonAIAgents instance
agents = PraisonAIAgents(
# Create Agents instance
agents = Agents(
agents=[image_agent],
tasks=[task1, task2],
process="sequential",
+12 -13
View File
@@ -5,7 +5,7 @@ This example demonstrates a mathematical problem-solving agent that can handle v
including calculations, equation solving, and step-by-step explanations.
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from typing import Dict, Any
import math
import re
@@ -65,9 +65,9 @@ calculation_agent = Agent(
backstory="You are an expert mathematician who can handle complex calculations with precision.",
instructions="Perform the requested calculations step by step, showing your work clearly.",
tools=[basic_calculator, solve_quadratic, calculate_statistics],
self_reflect=True,
min_reflect=1,
max_reflect=2
reflection=True,
)
problem_solver_agent = Agent(
@@ -81,9 +81,9 @@ problem_solver_agent = Agent(
3. Determine what needs to be found
4. Show each step of the solution
5. Verify the answer makes sense""",
self_reflect=True,
min_reflect=2,
max_reflect=4
reflection=True,
)
math_tutor_agent = Agent(
@@ -123,11 +123,10 @@ def solve_math_problem_workflow():
context=[solve_task]
)
workflow = PraisonAIAgents(
workflow = Agents(
agents=[problem_solver_agent, calculation_agent, math_tutor_agent],
tasks=[understand_task, solve_task, verify_task],
process="sequential",
verbose=True
process="sequential", output="verbose"
)
return workflow.start()
@@ -148,9 +147,9 @@ math_assistant = Agent(
Always show your work step by step.""",
tools=[basic_calculator, solve_quadratic, calculate_statistics],
self_reflect=True,
min_reflect=1,
max_reflect=3
reflection=True,
)
if __name__ == "__main__":
+19 -30
View File
@@ -6,7 +6,7 @@ agent-based selection for cost optimization and performance.
"""
import os
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from praisonaiagents.agent import RouterAgent
from praisonaiagents.llm.model_router import ModelRouter, ModelProfile, TaskComplexity
@@ -21,9 +21,8 @@ def example_auto_routing():
role="Adaptive Research Assistant",
goal="Research topics using the most appropriate AI model",
backstory="I analyze task complexity and route to the best model",
models=["gpt-5-nano", "gpt-5-nano", "claude-3-5-sonnet-20241022"],
routing_strategy="auto", # Automatic model selection
verbose=True
models=["gpt-4o-mini", "gpt-4o-mini", "claude-3-5-sonnet-20241022"],
routing_strategy="auto" # Automatic model selection
)
# Create tasks with different complexity levels
@@ -49,11 +48,10 @@ def example_auto_routing():
)
# Run the tasks
agents = PraisonAIAgents(
agents = Agents(
agents=[research_agent],
tasks=[simple_task, moderate_task, complex_task],
process="sequential",
verbose=True
process="sequential", output="verbose"
)
results = agents.start()
@@ -81,13 +79,11 @@ def example_cost_optimized_workflow():
goal="Analyze data efficiently while minimizing costs",
models={
"gemini/gemini-1.5-flash": {},
"gpt-5-nano": {},
"gpt-4o-mini": {},
"deepseek-chat": {}
},
model_router=cost_router,
routing_strategy="cost-optimized",
verbose=True
)
routing_strategy="cost-optimized")
writer = RouterAgent(
name="Quality Writer",
@@ -95,10 +91,9 @@ def example_cost_optimized_workflow():
goal="Create high-quality content",
models={
"claude-3-5-sonnet-20241022": {},
"gpt-5-nano": {}
"gpt-4o-mini": {}
},
routing_strategy="performance-optimized", # Prefer better models
verbose=True
routing_strategy="performance-optimized" # Prefer better models
)
# Create workflow tasks
@@ -118,11 +113,10 @@ def example_cost_optimized_workflow():
)
# Run workflow
workflow = PraisonAIAgents(
workflow = Agents(
agents=[analyzer, writer],
tasks=[analysis_task, writing_task],
process="sequential",
verbose=True
process="sequential", output="verbose"
)
results = workflow.start()
@@ -144,8 +138,8 @@ def example_auto_agents_multi_provider():
auto_agents = AutoAgents(
instructions="Create a market research report on electric vehicles. Include data analysis, competitor analysis, and future projections.",
max_agents=3,
llm="gpt-5-nano", # Default model for agent generation
verbose=True
llm="gpt-4o-mini", # Default model for agent generation
output="verbose"
)
# After agents are created, upgrade them to multi-model agents
@@ -158,10 +152,8 @@ def example_auto_agents_multi_provider():
goal=agent.goal,
backstory=agent.backstory,
tools=agent.tools,
models=["gpt-5-nano", "gemini/gemini-1.5-flash", "claude-3-haiku-20240307", "gpt-5-nano"],
routing_strategy="auto",
verbose=True
)
models=["gpt-4o-mini", "gemini/gemini-1.5-flash", "claude-3-haiku-20240307", "gpt-4o-mini"],
routing_strategy="auto")
multi_model_agents.append(multi_agent)
# Update the agents in the AutoAgents instance
@@ -192,7 +184,7 @@ def example_custom_routing():
# Create custom model profiles for specific use cases
custom_models = [
ModelProfile(
name="gpt-5-nano",
name="gpt-4o-mini",
provider="openai",
complexity_range=(TaskComplexity.SIMPLE, TaskComplexity.VERY_COMPLEX),
cost_per_1k_tokens=0.0075,
@@ -227,9 +219,7 @@ def example_custom_routing():
backstory="I'm an expert coder who knows when to use different AI models",
model_router=custom_router,
routing_strategy="auto",
tools=[search_web],
verbose=True
)
tools=[search_web])
# Create coding tasks
tasks = [
@@ -249,11 +239,10 @@ def example_custom_routing():
]
# Run tasks
agents = PraisonAIAgents(
agents = Agents(
agents=[coder],
tasks=tasks,
process="sequential",
verbose=True
process="sequential", output="verbose"
)
results = agents.start()
+3 -2
View File
@@ -1,10 +1,11 @@
from praisonaiagents import Agent, Tools
from praisonaiagents import Agent, Tools, ReflectionConfig
from praisonaiagents.tools import execute_code, analyze_code, format_code, lint_code, disassemble_code # Code Tools
from praisonaiagents.tools import execute_command, list_processes, kill_process, get_system_info # Shell Tools
from praisonaiagents.tools import duckduckgo # Web Search Tool
agent = Agent(
instructions="You are a Programming Agent", self_reflect=True, min_reflect=5, max_reflect=10,
instructions="You are a Programming Agent",
reflection=ReflectionConfig(min_iterations=5, max_iterations=10),
tools=[execute_code, analyze_code, format_code, lint_code, disassemble_code, execute_command, list_processes, kill_process, get_system_info, duckduckgo]
)
agent.start(
@@ -18,8 +18,8 @@ from praisonaiagents.tools import duckduckgo
# Define model options with cost considerations
# Using different models for different complexity levels
cheap_model = "gpt-5-nano" # For simple tasks
balanced_model = "gpt-5-nano" # For moderate complexity
cheap_model = "gpt-4o-mini" # For simple tasks
balanced_model = "gpt-4o-mini" # For moderate complexity
premium_model = "claude-3-5-sonnet-20241022" # For complex tasks
# Create RouterAgent with cost optimization strategy
@@ -38,9 +38,7 @@ router = RouterAgent(
like writing, analysis, or research, use the balanced model. For complex tasks requiring
deep reasoning, creativity, or specialized knowledge, use the premium model.""",
tools=[duckduckgo],
verbose=True
)
tools=[duckduckgo])
# Example tasks of varying complexity levels
@@ -76,7 +74,7 @@ print("\n" + "="*80)
print("COST OPTIMIZATION ROUTER DEMONSTRATION COMPLETED")
print("="*80)
print("The RouterAgent automatically selected appropriate models based on task complexity:")
print("- Simple factual questions → Cheaper model (gpt-5-nano)")
print("- Moderate analysis tasks → Balanced model (gpt-5-nano)")
print("- Simple factual questions → Cheaper model (gpt-4o-mini)")
print("- Moderate analysis tasks → Balanced model (gpt-4o-mini)")
print("- Complex strategic analysis → Premium model (claude-3-5-sonnet)")
print("- This approach optimizes costs while maintaining quality for each task type")
+5 -5
View File
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
# Create Video Analysis Agent
video_agent = Agent(
@@ -7,8 +7,8 @@ video_agent = Agent(
goal="Analyze images and videos to extract meaningful information",
backstory="""You are an expert in computer vision and image analysis.
You excel at describing images, detecting objects, and understanding visual content.""",
llm="gpt-5-nano",
self_reflect=False
llm="gpt-4o-mini",
reflection=False
)
# Task with Video File
@@ -24,8 +24,8 @@ task1 = Task(
images=["video.mp4"]
)
# Create PraisonAIAgents instance
agents = PraisonAIAgents(
# Create Agents instance
agents = Agents(
agents=[video_agent],
tasks=[task1],
process="sequential",
+4 -4
View File
@@ -1,12 +1,12 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from praisonaiagents.tools import wiki_search, wiki_summary, wiki_page, wiki_random, wiki_language
agent = Agent(
instructions="You are a Wikipedia Agent",
tools=[wiki_search, wiki_summary, wiki_page, wiki_random, wiki_language],
self_reflect=True,
min_reflect=3,
max_reflect=5,
reflection=True,
)
agent.start(
"What is the history of AI?"
+1 -1
View File
@@ -2,7 +2,7 @@ from praisonaiagents import Agent, MCP
qa_agent = Agent(
instructions="""You are a Question Answering Agent.""",
llm="openai/gpt-5-nano",
llm="openai/gpt-4o-mini",
tools=MCP("http://localhost:8080/agents/sse")
)
+3 -3
View File
@@ -2,17 +2,17 @@ from praisonaiagents import Agent
weather_agent = Agent(
instructions="""You are a weather agent that can provide weather information for a given city.""",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
stock_agent = Agent(
instructions="""You are a stock market agent that can provide information about stock prices and market trends.""",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
travel_agent = Agent(
instructions="""You are a travel agent that can provide recommendations for destinations, hotels, and activities.""",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
weather_agent.launch(path="/weather", port=3030)
@@ -32,7 +32,7 @@ import os
from pathlib import Path
# Import our market research system
from praisonaiagents import Agent, Task, PraisonAIAgents, Tools
from praisonaiagents import Agent, Task, Agents, Tools
# Create FastAPI app
app = FastAPI(
@@ -115,8 +115,7 @@ def create_market_research_agents(config: MarketResearchConfig):
Analyze market size, trends, growth drivers, and overall market dynamics for
{config.industry} in {config.geography}. Provide quantitative data and cite sources.
""",
tools=[Tools.internet_search],
verbose=False
tools=[Tools.internet_search], output="minimal"
)
if "competitive_analysis" in config.sections:
@@ -128,8 +127,7 @@ def create_market_research_agents(config: MarketResearchConfig):
Analyze {config.company}'s competitive position in {config.industry} within {config.geography}.
Identify key competitors, market share, and competitive advantages.
""",
tools=[Tools.internet_search],
verbose=False
tools=[Tools.internet_search], output="minimal"
)
if "financial_performance" in config.sections:
@@ -141,8 +139,7 @@ def create_market_research_agents(config: MarketResearchConfig):
Research {config.company}'s financial performance, revenue trends, profitability,
and compare with industry benchmarks.
""",
tools=[Tools.internet_search],
verbose=False
tools=[Tools.internet_search], output="minimal"
)
if "growth_opportunities" in config.sections:
@@ -154,8 +151,7 @@ def create_market_research_agents(config: MarketResearchConfig):
Identify emerging opportunities, market gaps, expansion possibilities for {config.company}
in {config.industry} within {config.geography}.
""",
tools=[Tools.internet_search],
verbose=False
tools=[Tools.internet_search], output="minimal"
)
if "risk_assessment" in config.sections:
@@ -167,8 +163,7 @@ def create_market_research_agents(config: MarketResearchConfig):
Identify and analyze potential risks, challenges, and threats facing {config.company}
in {config.geography}. Consider regulatory, competitive, and market risks.
""",
tools=[Tools.internet_search],
verbose=False
tools=[Tools.internet_search], output="minimal"
)
# Always include synthesizer
@@ -179,8 +174,7 @@ def create_market_research_agents(config: MarketResearchConfig):
instructions=f"""
Create a professional secondary market research report for {config.company}
in {config.industry} with clear structure, insights, and recommendations.
""",
verbose=False
""", output="minimal"
)
return agents
@@ -266,11 +260,10 @@ async def generate_market_research_report(job_id: str, config: MarketResearchCon
job_status[job_id]["message"] = "Setting up research workflow..."
# Create workflow
workflow = PraisonAIAgents(
workflow = Agents(
agents=list(agents.values()),
tasks=tasks,
process="sequential",
verbose=False
process="sequential", output="minimal"
)
job_status[job_id]["progress"] = 30
+1 -1
View File
@@ -2,7 +2,7 @@ from praisonaiagents import Agent, MCP
search_agent = Agent(
instructions="""You are a weather agent that can provide weather information for a given city.""",
llm="openai/gpt-5-nano",
llm="openai/gpt-4o-mini",
tools=MCP("http://localhost:8080/sse")
)
search_agent.launch(path="/weather", port=3030)
+1 -1
View File
@@ -1,4 +1,4 @@
from praisonaiagents import Agent
agent = Agent(instructions="""You are a helpful assistant.""", llm="gpt-5-nano")
agent = Agent(instructions="""You are a helpful assistant.""", llm="gpt-4o-mini")
agent.launch(path="/ask", port=3030)
@@ -0,0 +1,52 @@
"""
Basic Agent Autonomy Example.
This example demonstrates the core autonomy features with a real API call.
Run with: python 00_autonomy_basic.py
Requirements:
- OPENAI_API_KEY environment variable set
"""
from praisonaiagents import Agent
def main():
# Create an agent with autonomy enabled
agent = Agent(
instructions="You are a helpful assistant. Be concise.",
autonomy=True,
)
print("=" * 50)
print("Agent Autonomy Basic Example")
print("=" * 50)
# Test 1: Signal detection (no API call)
print("\n1. Signal Detection (fast, no API call):")
prompts = [
"What is Python?",
"Read src/main.py and explain it",
"Refactor the auth module and add tests",
]
for prompt in prompts:
stage = agent.get_recommended_stage(prompt)
print(f" '{prompt[:40]}...' → {stage}")
# Test 2: Simple chat with autonomy (real API call)
print("\n2. Simple Chat (real API call):")
response = agent.chat("What is 2 + 2? Answer with just the number.")
print(f" Response: {response}")
# Test 3: Check autonomy status
print("\n3. Autonomy Status:")
print(f" Enabled: {agent.autonomy_enabled}")
print(f" Config: {agent.autonomy_config}")
print("\n" + "=" * 50)
print("✓ Basic autonomy example completed successfully!")
print("=" * 50)
if __name__ == "__main__":
main()
@@ -0,0 +1,94 @@
"""
Advanced Agent Autonomy Example.
Demonstrates advanced autonomy features:
- Custom configuration
- Doom loop detection
- Verification hooks
- Subagent delegation
Run with: python 01_autonomy_advanced.py
Requirements:
- OPENAI_API_KEY environment variable set
"""
from praisonaiagents import Agent
from praisonaiagents.agents.profiles import get_profile, BUILTIN_PROFILES
def main():
print("=" * 60)
print("Agent Autonomy Advanced Example")
print("=" * 60)
# 1. Custom autonomy configuration
print("\n1. Custom Autonomy Configuration:")
agent = Agent(
instructions="You are a coding assistant.",
autonomy={
"max_iterations": 30,
"doom_loop_threshold": 5,
"auto_escalate": True,
},
)
print(f" Max iterations: {agent.autonomy_config.get('max_iterations')}")
print(f" Doom loop threshold: {agent.autonomy_config.get('doom_loop_threshold')}")
print(f" Auto escalate: {agent.autonomy_config.get('auto_escalate')}")
# 2. Doom loop detection
print("\n2. Doom Loop Detection:")
for i in range(5):
agent._record_action("read_file", {"path": "test.py"}, "content", True)
is_doom = agent._is_doom_loop()
print(f" After 5 repeated actions: doom_loop={is_doom}")
agent._reset_doom_loop()
is_doom_after = agent._is_doom_loop()
print(f" After reset: doom_loop={is_doom_after}")
# 3. Verification hooks
print("\n3. Verification Hooks:")
class MockTestRunner:
name = "pytest"
def run(self, context=None):
return {"success": True, "output": "5 tests passed"}
agent_with_hooks = Agent(
instructions="You are a test-driven developer.",
autonomy=True,
verification_hooks=[MockTestRunner()],
)
results = agent_with_hooks._run_verification_hooks()
print(f" Hook results: {results}")
# 4. Subagent delegation
print("\n4. Subagent Delegation:")
subagent = agent._create_subagent("explorer")
print(f" Created subagent: {subagent.name}")
print(f" Profile: explorer (read-only)")
# 5. Explorer profile details
print("\n5. Explorer Profile:")
explorer = get_profile("explorer")
print(f" Tools: {explorer.tools}")
print(f" Read-only: {explorer.metadata.get('read_only')}")
print(f" Blocked tools: {explorer.metadata.get('blocked_tools')}")
# 6. Available profiles
print("\n6. Available Agent Profiles:")
for name in BUILTIN_PROFILES:
profile = get_profile(name)
if not profile.hidden:
print(f" - {name}: {profile.description[:50]}...")
print("\n" + "=" * 60)
print("✓ Advanced autonomy example completed successfully!")
print("=" * 60)
if __name__ == "__main__":
main()
@@ -0,0 +1,51 @@
"""
run_autonomous() Example.
Demonstrates the unified autonomous loop with real API calls.
Run with: python 02_run_autonomous.py
Requirements:
- OPENAI_API_KEY environment variable set
"""
from praisonaiagents import Agent
def main():
print("=" * 60)
print("run_autonomous() Example")
print("=" * 60)
# Create agent with autonomy
agent = Agent(
instructions="You are a helpful assistant. When asked a question, provide a clear answer and say 'Task completed' at the end.",
autonomy={"max_iterations": 5},
)
# Test 1: Simple task (should complete in 1 iteration)
print("\n1. Simple Task (direct stage):")
result = agent.run_autonomous(
"What is the capital of France? Say 'Task completed' after answering.",
max_iterations=3,
)
print(f" Success: {result.success}")
print(f" Completion reason: {result.completion_reason}")
print(f" Iterations: {result.iterations}")
print(f" Stage: {result.stage}")
print(f" Duration: {result.duration_seconds:.2f}s")
print(f" Output: {result.output[:100]}...")
# Test 2: Check result attributes
print("\n2. Result Attributes:")
print(f" Has actions: {len(result.actions) > 0}")
print(f" Has error: {result.error is not None}")
print("\n" + "=" * 60)
print("✓ run_autonomous() example completed successfully!")
print("=" * 60)
if __name__ == "__main__":
main()
@@ -0,0 +1,110 @@
"""
Verification Hooks Example.
Demonstrates how to use verification hooks with Agent autonomy.
Run with: python 03_verification_hooks.py
Requirements:
- OPENAI_API_KEY environment variable set
"""
from praisonaiagents import Agent
# Agent-centric quickstart
agent = Agent(
name="VerificationHooksDemo",
instructions="You are a helpful assistant demonstrating verification hooks."
)
# --- Advanced: Low-level Verification API ---
from praisonaiagents.hooks.verification import (
VerificationHook,
VerificationResult,
BaseVerificationHook,
CommandVerificationHook,
)
class SimpleTestHook(BaseVerificationHook):
"""A simple test verification hook."""
name = "simple_test"
def _execute(self, context=None):
return VerificationResult(
success=True,
output="All tests passed!",
details={"tests_run": 5, "passed": 5, "failed": 0},
)
class LintHook(BaseVerificationHook):
"""A lint verification hook."""
name = "lint"
def _execute(self, context=None):
return VerificationResult(
success=True,
output="No lint errors found",
details={"files_checked": 10, "errors": 0, "warnings": 2},
)
def main():
print("=" * 60)
print("Verification Hooks Example")
print("=" * 60)
# 1. Create hooks
print("\n1. Creating Verification Hooks:")
test_hook = SimpleTestHook()
lint_hook = LintHook()
print(f" - {test_hook.name}: {type(test_hook).__name__}")
print(f" - {lint_hook.name}: {type(lint_hook).__name__}")
# 2. Run hooks manually
print("\n2. Running Hooks Manually:")
test_result = test_hook.run()
lint_result = lint_hook.run()
print(f" Test result: success={test_result.success}, output='{test_result.output}'")
print(f" Lint result: success={lint_result.success}, output='{lint_result.output}'")
# 3. Create agent with verification hooks
print("\n3. Agent with Verification Hooks:")
agent = Agent(
instructions="You are a test-driven developer.",
autonomy=True,
verification_hooks=[test_hook, lint_hook],
)
print(f" Hooks registered: {len(agent._verification_hooks)}")
# 4. Run verification hooks through agent
print("\n4. Running Hooks Through Agent:")
results = agent._run_verification_hooks()
for result in results:
print(f" - {result['hook']}: success={result['success']}")
# 5. CommandVerificationHook example (without running)
print("\n5. CommandVerificationHook (structure only):")
cmd_hook = CommandVerificationHook(
name="pytest",
command=["python", "-m", "pytest", "-v"],
timeout=30.0,
)
print(f" Name: {cmd_hook.name}")
print(f" Command: {cmd_hook.command}")
print(f" Timeout: {cmd_hook.timeout_seconds}s")
print("\n" + "=" * 60)
print("✓ Verification hooks example completed successfully!")
print("=" * 60)
if __name__ == "__main__":
main()
+3 -3
View File
@@ -16,7 +16,7 @@ Environment Variables:
import cv2
import os
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
def capture_and_analyze():
@@ -28,7 +28,7 @@ def capture_and_analyze():
role="Camera Feed Analyzer",
goal="Analyze camera captures in real-time",
backstory="Expert in real-time visual analysis with computer vision skills",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
# Initialize camera (0 for default camera)
@@ -58,7 +58,7 @@ def capture_and_analyze():
)
# Run analysis
agents = PraisonAIAgents(
agents = Agents(
agents=[vision_agent],
tasks=[task],
process="sequential",
+3 -3
View File
@@ -18,7 +18,7 @@ import cv2
import time
import os
from datetime import datetime
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
class ContinuousCameraMonitor:
@@ -42,7 +42,7 @@ class ContinuousCameraMonitor:
role="Security Camera Analyst",
goal="Monitor camera feeds for activities and security events",
backstory="Expert security analyst with computer vision skills specialized in real-time monitoring",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
def capture_frame(self, cap):
@@ -71,7 +71,7 @@ class ContinuousCameraMonitor:
images=[image_path]
)
agents = PraisonAIAgents(
agents = Agents(
agents=[self.vision_agent],
tasks=[task],
process="sequential",
+5 -5
View File
@@ -17,7 +17,7 @@ Environment Variables:
import cv2
import os
from datetime import datetime
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
def capture_camera_frame(camera_id=0):
@@ -57,7 +57,7 @@ def analyze_with_multiple_agents():
role="Security Specialist",
goal="Identify security threats and suspicious activities",
backstory="Expert security analyst with years of experience in surveillance and threat detection",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
object_detector = Agent(
@@ -65,7 +65,7 @@ def analyze_with_multiple_agents():
role="Object Recognition Specialist",
goal="Identify and catalog all objects in the scene",
backstory="Computer vision expert specializing in object detection and classification",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
scene_analyst = Agent(
@@ -73,7 +73,7 @@ def analyze_with_multiple_agents():
role="Scene Understanding Expert",
goal="Provide comprehensive scene analysis and context",
backstory="Environmental analyst expert in understanding spatial relationships and scene context",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
# Create specialized tasks
@@ -114,7 +114,7 @@ def analyze_with_multiple_agents():
)
# Run multi-agent analysis
agents = PraisonAIAgents(
agents = Agents(
agents=[security_agent, object_detector, scene_analyst],
tasks=[security_task, object_task, scene_task],
process="parallel", # Run all analyses in parallel
@@ -18,7 +18,7 @@ import cv2
import os
import time
from datetime import datetime
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
class CameraVideoAnalyzer:
@@ -39,7 +39,7 @@ class CameraVideoAnalyzer:
role="Video Content Analyzer",
goal="Analyze video content for activities, objects, and events",
backstory="Expert in video analysis and temporal event detection with computer vision expertise",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
def record_video_segment(self, duration_seconds=10):
@@ -131,7 +131,7 @@ class CameraVideoAnalyzer:
images=[video_path] # PraisonAI supports video files in images parameter
)
agents = PraisonAIAgents(
agents = Agents(
agents=[self.video_agent],
tasks=[task],
process="sequential",
+1 -1
View File
@@ -13,7 +13,7 @@ async def main():
workspace=".",
output_format="json",
force=True,
model="gpt-5",
model="gpt-4o",
stream_partial=True
)
+3 -3
View File
@@ -27,9 +27,9 @@ pdf_agent = Agent(
instructions="Read the provided PDF document and answer questions about its content. Be specific and cite relevant sections when possible.",
knowledge=["document.pdf"], # Replace with your PDF file path
knowledge_config=config,
self_reflect=True,
min_reflect=1,
max_reflect=3
reflection=True,
)
# Example usage
@@ -24,7 +24,7 @@ from pathlib import Path
project_root = Path(__file__).parent.parent.parent.parent / "src" / "praisonai-agents"
sys.path.insert(0, str(project_root))
from praisonaiagents import Agent, Task, PraisonAIAgents, create_context_agent
from praisonaiagents import Agent, Task, Agents, create_context_agent
class ContextEngineeringWorkflow:
"""
@@ -34,7 +34,7 @@ class ContextEngineeringWorkflow:
to achieve higher success rates in AI-driven development tasks.
"""
def __init__(self, project_path: str, llm: str = "gpt-5-nano"):
def __init__(self, project_path: str, llm: str = "gpt-4o-mini"):
self.project_path = project_path
self.llm = llm
self.context_data = {}
@@ -59,16 +59,12 @@ class ContextEngineeringWorkflow:
4. Define success metrics
5. Specify technical constraints
""",
llm=self.llm,
verbose=True
)
llm=self.llm)
# 2. Context Engineering Agent - Generates comprehensive context
self.context_engineer = create_context_agent(
llm=self.llm,
name="Context Engineering Specialist",
verbose=True
)
name="Context Engineering Specialist")
# 3. Software Architect Agent - Designs implementation using context
self.architect = Agent(
@@ -87,9 +83,7 @@ class ContextEngineeringWorkflow:
4. Consider security and performance implications
5. Document architectural decisions and rationale
""",
llm=self.llm,
verbose=True
)
llm=self.llm)
# 4. Senior Developer Agent - Implements using context-enhanced guidance
self.developer = Agent(
@@ -107,9 +101,7 @@ class ContextEngineeringWorkflow:
4. Write clean, maintainable code
5. Follow the implementation blueprint exactly
""",
llm=self.llm,
verbose=True
)
llm=self.llm)
# 5. QA Engineer Agent - Validates using context-generated criteria
self.qa_engineer = Agent(
@@ -127,9 +119,7 @@ class ContextEngineeringWorkflow:
4. Check for security vulnerabilities
5. Validate performance requirements
""",
llm=self.llm,
verbose=True
)
llm=self.llm)
def run_context_engineering_workflow(self, feature_request: str):
"""
@@ -312,7 +302,7 @@ class ContextEngineeringWorkflow:
print("\n⚙️ Executing Context Engineering Workflow")
print("-" * 50)
agents_workflow = PraisonAIAgents(
agents_workflow = Agents(
agents=[
self.product_manager,
self.architect,
@@ -325,8 +315,7 @@ class ContextEngineeringWorkflow:
implementation_task,
qa_task
],
process="sequential",
verbose=True
process="sequential", output="verbose"
)
# Execute workflow
@@ -382,7 +371,7 @@ def run_example_workflow():
project_path = str(project_root)
workflow = ContextEngineeringWorkflow(
project_path=project_path,
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
# Example feature request
@@ -446,7 +435,7 @@ if __name__ == "__main__":
print("(Note: Full execution requires complete environment setup)")
project_path = str(project_root)
workflow = ContextEngineeringWorkflow(project_path, "gpt-5-nano")
workflow = ContextEngineeringWorkflow(project_path, "gpt-4o-mini")
print(f"\n✅ Workflow Setup Complete:")
print(f" • Project path: {project_path}")
@@ -11,7 +11,7 @@ Author: Generated for GitHub Issue #23
"""
import os
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from praisonaiagents.tools import read_csv, write_csv
def create_sample_urls_csv():
@@ -68,7 +68,7 @@ def method_1_simple_loop():
goal="Process each task from the CSV file efficiently",
backstory="Expert at handling repetitive tasks from CSV data",
instructions="Process each task thoroughly and provide detailed output",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
# Create a loop task that will automatically process each CSV row
@@ -81,7 +81,7 @@ def method_1_simple_loop():
)
# Initialize the agents system
agents = PraisonAIAgents(
agents = Agents(
agents=[loop_agent],
tasks=[loop_task],
process="workflow",
@@ -127,7 +127,7 @@ def method_2_manual_csv():
4. Create a summary of the processing results
5. Save results to a new CSV file if needed
""",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
# Create a task for manual CSV processing
@@ -145,7 +145,7 @@ def method_2_manual_csv():
)
# Initialize and run
agents = PraisonAIAgents(
agents = Agents(
agents=[csv_agent],
tasks=[csv_task],
process="sequential"
@@ -187,7 +187,7 @@ def method_3_url_processing():
4. Create a detailed analysis report
5. Save results with analysis data
""",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
# Create the URL processing task
@@ -211,7 +211,7 @@ def method_3_url_processing():
)
# Initialize and run
agents = PraisonAIAgents(
agents = Agents(
agents=[url_agent],
tasks=[url_task],
process="sequential"
@@ -264,7 +264,7 @@ def method_4_advanced_processing():
4. Create separate files for valid and invalid data
5. Provide detailed validation reports
""",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
# Create a processing agent
@@ -281,7 +281,7 @@ def method_4_advanced_processing():
3. Generate comprehensive results
4. Create detailed output reports
""",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
# Create validation task
@@ -319,7 +319,7 @@ def method_4_advanced_processing():
)
# Initialize and run with sequential processing
agents = PraisonAIAgents(
agents = Agents(
agents=[validator_agent, processor_agent],
tasks=[validation_task, processing_task],
process="sequential" # Validation must complete before processing
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
import logging
import os
@@ -24,9 +24,7 @@ knowledge_agent = Agent(
goal="Store and retrieve knowledge efficiently",
backstory="Expert in managing and utilizing stored knowledge",
knowledge=["sample.pdf"],
knowledge_config=config,
verbose=True
)
knowledge_config=config)
# Define a task for the agent
knowledge_task = Task(
@@ -37,7 +35,7 @@ knowledge_task = Task(
)
# Create and start the agents
agents = PraisonAIAgents(
agents = Agents(
agents=[knowledge_agent],
tasks=[knowledge_task],
process="sequential",
@@ -140,7 +140,7 @@ def main():
instructions="You are an AI research assistant. Use your knowledge base to answer questions about artificial intelligence topics.",
knowledge=sample_documents, # Will use default knowledge config
knowledge_config=rerank_config, # Use reranking config
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
print("7. Agent with reranking knowledge:")
+3 -3
View File
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
# Define the configuration for the Knowledge instance
config = {
@@ -16,7 +16,7 @@ rag_agent = Agent(
name="RAG Agent",
role="Information Specialist",
goal="Retrieve knowledge efficiently",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
# Define a task for the agent
@@ -29,7 +29,7 @@ rag_task = Task(
)
# Build Agents
agents = PraisonAIAgents(
agents = Agents(
agents=[rag_agent],
tasks=[rag_task],
user_id="user1"
@@ -8,7 +8,7 @@ This example demonstrates how to extract and utilize reasoning patterns in Prais
- Decision rationale extraction
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from typing import Dict, List, Tuple
from pydantic import BaseModel
import json
@@ -49,9 +49,9 @@ cot_agent = Agent(
6. Assess your confidence level
Always make your thinking process explicit and transparent.""",
self_reflect=True,
min_reflect=2,
max_reflect=3
reflection=True,
)
# Example 2: Reasoning Extractor Agent
@@ -100,11 +100,10 @@ def demonstrate_reasoning_extraction():
context=[problem_task]
)
workflow = PraisonAIAgents(
workflow = Agents(
agents=[cot_agent, reasoning_extractor],
tasks=[problem_task, extract_task],
process="sequential",
verbose=True
process="sequential", output="verbose"
)
return workflow.start()
@@ -184,11 +183,10 @@ def advanced_reasoning_patterns():
context=[deductive_task, inductive_task, abductive_task]
)
workflow = PraisonAIAgents(
workflow = Agents(
agents=[cot_agent, reasoning_validator],
tasks=[deductive_task, inductive_task, abductive_task, validation_task],
process="sequential",
verbose=True
process="sequential", output="verbose"
)
return workflow.start()
@@ -224,7 +222,7 @@ pattern_reasoning_agent = Agent(
3. Counterfactual: Explore what-if scenarios
4. Systems thinking: Consider interconnections
5. Probabilistic: Assess likelihoods""",
self_reflect=True
reflection=True
)
# Example 8: Interactive Reasoning Extraction
@@ -1,8 +1,8 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
agent = Agent(
instructions="You are a loop agent that creating a loop of tasks.",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
task = Task(
@@ -12,7 +12,7 @@ task = Task(
input_file="tasks.csv"
)
agents = PraisonAIAgents(
agents = Agents(
agents=[agent],
tasks=[task],
process="workflow",
+10 -14
View File
@@ -8,7 +8,7 @@ This example demonstrates various routing patterns in PraisonAI including:
- Dynamic task selection
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from typing import Dict, Any
import random
@@ -77,11 +77,10 @@ def conditional_routing_example():
agent=processor_agent
)
workflow = PraisonAIAgents(
workflow = Agents(
agents=[analyzer_agent, processor_agent],
tasks=[analysis_task, premium_task, standard_task, basic_task],
process="workflow",
verbose=True
process="workflow", output="verbose"
)
return workflow.start()
@@ -110,11 +109,10 @@ def loop_pattern_example():
agent=validator_agent
)
workflow = PraisonAIAgents(
workflow = Agents(
agents=[processor_agent, validator_agent],
tasks=[process_task, finalize_task],
process="workflow",
verbose=True
process="workflow", output="verbose"
)
return workflow.start()
@@ -185,11 +183,10 @@ def decision_tree_example():
all_tasks = [categorize_task, technical_task, business_task] + terminal_tasks
workflow = PraisonAIAgents(
workflow = Agents(
agents=[analyzer_agent, processor_agent],
tasks=all_tasks,
process="workflow",
verbose=True
process="workflow", output="verbose"
)
return workflow.start()
@@ -208,7 +205,7 @@ dynamic_router = Agent(
5. Current system load
Provide clear routing decisions with reasoning.""",
self_reflect=True
reflection=True
)
def dynamic_routing_example():
@@ -272,11 +269,10 @@ def dynamic_routing_example():
agent=processor_agent
)
workflow = PraisonAIAgents(
workflow = Agents(
agents=[dynamic_router, processor_agent],
tasks=[route_task, escalate_task, automate_task, review_task, standard_task],
process="workflow",
verbose=True
process="workflow", output="verbose"
)
return workflow.start()
@@ -8,7 +8,7 @@ This example demonstrates the self-reflection capabilities in PraisonAI agents:
- Impact on output quality
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from typing import Dict, Any
import time
@@ -24,9 +24,9 @@ basic_reflection_agent = Agent(
3. Identify areas for improvement
4. Revise and enhance
5. Repeat until satisfied""",
self_reflect=True, # Enable self-reflection
min_reflect=1, # Minimum 1 reflection iteration
max_reflect=3 # Maximum 3 reflection iterations
reflection=True, # Enable self-reflection
# Minimum 1 reflection iteration
# Maximum 3 reflection iterations
)
# Example 2: Advanced Reflection with Different Strategies
@@ -41,9 +41,9 @@ quality_focused_agent = Agent(
- Clarity of presentation
- Logical flow
- Supporting evidence""",
self_reflect=True,
min_reflect=3, # Higher minimum for quality
max_reflect=5 # Allow more iterations
reflection=True,
# Higher minimum for quality
# Allow more iterations
)
efficiency_balanced_agent = Agent(
@@ -52,9 +52,9 @@ efficiency_balanced_agent = Agent(
goal="Achieve good quality with reasonable reflection time",
backstory="You balance quality with efficiency, knowing when good enough is sufficient.",
instructions="Create good content with 1-2 rounds of improvement.",
self_reflect=True,
min_reflect=1,
max_reflect=2 # Limit iterations for efficiency
reflection=True,
# Limit iterations for efficiency
)
# Example 3: No Reflection for Comparison
@@ -64,7 +64,7 @@ no_reflection_agent = Agent(
goal="Provide immediate responses without reflection",
backstory="You provide quick, first-draft responses.",
instructions="Respond immediately with your first thought.",
self_reflect=False # Disable reflection
reflection=False # Disable reflection
)
# Demonstration function to show reflection in action
@@ -113,9 +113,9 @@ def reflection_workflow_example():
role="Initial content creator",
goal="Create first draft with self-improvement",
backstory="You create initial drafts and refine them through reflection.",
self_reflect=True,
min_reflect=2,
max_reflect=3
reflection=True,
)
# Reviewer without reflection (provides feedback)
@@ -124,7 +124,7 @@ def reflection_workflow_example():
role="Content reviewer",
goal="Provide constructive feedback",
backstory="You review content and provide specific improvement suggestions.",
self_reflect=False # No reflection needed for reviewing
reflection=False # No reflection needed for reviewing
)
# Final polisher with reflection
@@ -133,9 +133,9 @@ def reflection_workflow_example():
role="Final content polisher",
goal="Perfect the content based on feedback",
backstory="You take feedback and create the final, polished version.",
self_reflect=True,
min_reflect=1,
max_reflect=2
reflection=True,
)
# Workflow tasks
@@ -162,11 +162,10 @@ def reflection_workflow_example():
context=[create_task, review_task]
)
workflow = PraisonAIAgents(
workflow = Agents(
agents=[creator_agent, reviewer_agent, polisher_agent],
tasks=[create_task, review_task, polish_task],
process="sequential",
verbose=True
process="sequential", output="verbose"
)
return workflow.start()
@@ -202,9 +201,9 @@ custom_reflection_agent = Agent(
backstory="You adapt your reflection process based on content quality.",
instructions="""Create content and reflect on it.
Continue improving until quality standards are met or maximum iterations reached.""",
self_reflect=True,
min_reflect=1,
max_reflect=5
reflection=True,
)
# Example 6: Reflection Impact Analysis
@@ -5,7 +5,7 @@ This example demonstrates self-reflection capabilities using PraisonAI's
built-in reflection features for iterative improvement and quality optimization.
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from praisonaiagents.tools import internet_search
print("=== Self-Reflection Optimization Patterns Example ===\n")
@@ -17,12 +17,7 @@ reflection_agent = Agent(
goal="Demonstrate self-reflection and iterative improvement patterns",
backstory="Expert researcher with strong self-reflection and continuous improvement capabilities",
tools=[internet_search],
self_reflect=True,
min_reflect=2,
max_reflect=4,
reflect_llm="gpt-5-nano",
verbose=True
)
reflection=True)
# Create task that benefits from self-reflection
reflection_task = Task(
@@ -9,7 +9,7 @@ This is the simplest possible implementation for processing URLs from a CSV file
Author: Generated for GitHub Issue #23
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from praisonaiagents.tools import write_csv
def create_url_csv():
@@ -46,7 +46,7 @@ def main():
goal="Analyze each URL from the CSV list",
backstory="Expert at analyzing websites and URLs",
instructions="Analyze each URL and provide insights about the website",
llm="gpt-5-nano" # You can change this to any supported model
llm="gpt-4o-mini" # You can change this to any supported model
)
# Step 3: Create a task that will loop through the CSV
@@ -59,7 +59,7 @@ def main():
)
# Step 4: Run the agents
agents = PraisonAIAgents(
agents = Agents(
agents=[url_agent],
tasks=[url_task],
process="workflow",
@@ -4,7 +4,7 @@ Example: Using Latency Tracking Tool with PraisonAI
This example shows how to track latency without modifying core files.
"""
from praisonaiagents import Agent, PraisonAIAgents
from praisonaiagents import Agent, Agents
from latency_tracker_tool import (
latency_tracking_tool,
create_tracked_agent,
@@ -27,7 +27,7 @@ def example_with_tool():
role="Information Researcher",
goal="Research and provide information",
tools=[latency_tracking_tool], # Add our custom tool
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
# Start tracking manually
@@ -60,7 +60,7 @@ def example_with_wrappers():
name="Analyst",
role="Data Analyst",
goal="Analyze data and provide insights",
llm="gpt-5-nano",
llm="gpt-4o-mini",
request_id="request_2" # Unique request ID
)
@@ -83,7 +83,7 @@ def example_with_context_manager():
name="Calculator",
role="Math Expert",
goal="Solve mathematical problems",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
request_id = "request_3"
@@ -126,7 +126,7 @@ def example_mcp_tracking():
name="Assistant",
role="MCP Assistant",
goal="Handle MCP requests",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
plan = agent.chat("Plan: Search for Python documentation")
@@ -166,7 +166,7 @@ def example_concurrent_tracking():
name=f"Agent_{i}",
role="Concurrent Worker",
goal="Process requests concurrently",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
with tracker.track("processing", request_id):
@@ -5,7 +5,7 @@ This shows how to add latency tracking to an MCP server
without modifying PraisonAI core files.
"""
from praisonaiagents import Agent, PraisonAIAgents
from praisonaiagents import Agent, Agents
from hosted_server import HostedMCPServer # Import from local file
from latency_tracker_tool import tracker, get_latency_metrics
import json
@@ -56,7 +56,7 @@ def create_mcp_agent_with_tracking(request_id: str = "mcp_request"):
name="MCPAssistant",
role="MCP Request Handler",
goal="Handle MCP requests with latency tracking",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
return agent
@@ -23,7 +23,7 @@ def handle_mcp_request(query: str, request_id: str = "mcp_1"):
name="Assistant",
role="MCP Handler",
goal="Process requests",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
plan = f"I will search for information about {query}"
time.sleep(0.1) # Simulate planning time
+8 -12
View File
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from praisonaiagents.tools import cot_save, cot_upload_to_huggingface
from pydantic import BaseModel
import os
@@ -29,7 +29,7 @@ qa_generator = Agent(
role="Question Creator",
goal="Create challenging math and logic questions",
backstory="Expert in educational content creation",
llm="gpt-5-nano",
llm="gpt-4o-mini",
tools=[write_csv, count_questions]
)
@@ -38,9 +38,8 @@ total_questions_evaluator = Agent(
role="Total Questions Evaluator",
goal="Evaluate the total number of questions in qa_pairs.csv file",
backstory="Expert in evaluating the total number of questions in a file",
llm="gpt-5-nano",
tools=[count_questions],
verbose=False
llm="gpt-4o-mini",
tools=[count_questions], output="minimal"
)
cot_generator = Agent(
@@ -49,8 +48,7 @@ cot_generator = Agent(
goal="Generate and manage chain of thought solutions for Q&A pairs",
backstory="Expert in breaking down problems and generating detailed solution steps",
tools=[cot_save],
llm="gpt-5-nano",
verbose=False
llm="gpt-4o-mini", output="minimal"
)
upload_to_huggingface = Agent(
@@ -59,8 +57,7 @@ upload_to_huggingface = Agent(
goal="Upload the generated chain of thought solutions to a Huggingface dataset",
backstory="Expert in saving data to Huggingface",
tools=[cot_upload_to_huggingface],
llm="gpt-5-nano",
verbose=False
llm="gpt-4o-mini", output="minimal"
)
# Define tasks with workflow improvements
@@ -131,12 +128,11 @@ upload_to_huggingface_task = Task(
)
# Initialize workflow
agents = PraisonAIAgents(
agents = Agents(
agents=[qa_generator, total_questions_evaluator, cot_generator, upload_to_huggingface],
tasks=[generate_task, evaluate_total_questions_task, generate_cot_task, upload_to_huggingface_task],
process="workflow",
max_iter=30,
verbose=False
max_iter=30, output="minimal"
)
agents.start()
+192
View File
@@ -0,0 +1,192 @@
"""
Escalation Pipeline Example.
Demonstrates the progressive escalation from direct response to autonomous mode.
"""
import asyncio
from praisonaiagents.escalation import (
EscalationPipeline,
EscalationStage,
EscalationConfig,
EscalationTrigger,
DoomLoopDetector,
ObservabilityHooks,
EventType,
)
def demo_signal_detection():
"""Demonstrate signal detection from prompts."""
print("\n" + "="*60)
print("SIGNAL DETECTION DEMO")
print("="*60)
trigger = EscalationTrigger()
# Test various prompts
test_prompts = [
("What is Python?", "Simple question"),
("Read the file src/main.py", "File reference"),
("Refactor the authentication module and add tests", "Complex task"),
("First, analyze the code. Then, fix the bug. Finally, test it.", "Multi-step"),
]
for prompt, description in test_prompts:
signals = trigger.analyze(prompt)
stage = trigger.recommend_stage(signals)
print(f"\n📝 Prompt: {prompt[:50]}...")
print(f" Description: {description}")
print(f" Signals: {[s.value for s in signals]}")
print(f" Recommended Stage: {stage.name}")
def demo_doom_loop_detection():
"""Demonstrate doom loop detection."""
print("\n" + "="*60)
print("DOOM LOOP DETECTION DEMO")
print("="*60)
detector = DoomLoopDetector()
detector.start_session()
# Simulate repeated identical actions
print("\n🔄 Simulating repeated identical actions...")
for i in range(4):
detector.record_action(
action_type="read_file",
args={"path": "config.py"},
result="same content",
success=True,
)
if detector.is_doom_loop():
loop_type = detector.get_loop_type()
recovery = detector.get_recovery_action()
print(f" ⚠️ Doom loop detected after {i+1} actions!")
print(f" Type: {loop_type.value}")
print(f" Recovery: {recovery.value}")
break
else:
print(f" Action {i+1}: No loop detected")
# Show stats
stats = detector.get_stats()
print(f"\n📊 Stats: {stats}")
def demo_observability():
"""Demonstrate observability hooks."""
print("\n" + "="*60)
print("OBSERVABILITY DEMO")
print("="*60)
hooks = ObservabilityHooks()
# Register event handlers
def on_stage_change(event):
print(f" 📍 Stage changed: {event.data}")
def on_tool_call(event):
print(f" 🔧 Tool called: {event.data}")
hooks.on(EventType.STAGE_ESCALATE, on_stage_change)
hooks.on(EventType.TOOL_CALL_END, on_tool_call)
# Simulate events
print("\n🎯 Simulating execution events...")
hooks.start_execution("session_123")
hooks.set_stage(EscalationStage.DIRECT)
hooks.emit(EventType.STEP_START, {"step": 1})
hooks.emit(EventType.TOOL_CALL_END, {"tool": "read_file", "path": "main.py"})
hooks.emit(EventType.STEP_END, {"step": 1})
hooks.emit(EventType.STAGE_ESCALATE, {"from": "DIRECT", "to": "HEURISTIC"})
hooks.set_stage(EscalationStage.HEURISTIC)
hooks.end_execution(1500.0)
# Show summary
summary = hooks.get_summary()
print(f"\n📊 Summary: {summary}")
def demo_stage_descriptions():
"""Show descriptions of each escalation stage."""
print("\n" + "="*60)
print("ESCALATION STAGES")
print("="*60)
trigger = EscalationTrigger()
stages = [
(EscalationStage.DIRECT, "Simple questions, no tools needed"),
(EscalationStage.HEURISTIC, "File references, code blocks"),
(EscalationStage.PLANNED, "Edit intent, test/build tasks"),
(EscalationStage.AUTONOMOUS, "Multi-step, refactoring tasks"),
]
for stage, example in stages:
desc = trigger.get_stage_description(stage)
print(f"\n🎯 Stage {stage.value}: {stage.name}")
print(f" Description: {desc}")
print(f" Example use: {example}")
async def demo_pipeline_analysis():
"""Demonstrate pipeline analysis."""
print("\n" + "="*60)
print("PIPELINE ANALYSIS DEMO")
print("="*60)
config = EscalationConfig(
max_steps=20,
max_time_seconds=60,
auto_escalate=True,
)
pipeline = EscalationPipeline(config=config)
prompts = [
"What is a Python decorator?",
"Read the file src/auth.py and explain it",
"Refactor the user authentication to use JWT tokens",
]
for prompt in prompts:
stage = pipeline.analyze(prompt)
print(f"\n📝 Prompt: {prompt[:50]}...")
print(f" Recommended Stage: {stage.name}")
def main():
"""Run all demos."""
print("\n" + "="*60)
print("PRAISONAI ESCALATION PIPELINE DEMO")
print("="*60)
print("\nThis demo shows the progressive escalation system that")
print("automatically adjusts execution complexity based on task signals.")
# Run demos
demo_stage_descriptions()
demo_signal_detection()
demo_doom_loop_detection()
demo_observability()
asyncio.run(demo_pipeline_analysis())
print("\n" + "="*60)
print("DEMO COMPLETE")
print("="*60)
print("\nKey takeaways:")
print("1. Simple tasks stay at DIRECT stage (no tools)")
print("2. File references trigger HEURISTIC stage")
print("3. Edit/test tasks trigger PLANNED stage")
print("4. Multi-step/refactor tasks trigger AUTONOMOUS stage")
print("5. Doom loop detection prevents infinite loops")
print("6. Observability hooks enable monitoring")
if __name__ == "__main__":
main()
@@ -5,7 +5,7 @@ This example demonstrates callback systems using PraisonAI's built-in
callback functionality for monitoring agent interactions and tool usage.
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from praisonaiagents.display_callback import register_display_callback
from praisonaiagents.tools import internet_search
@@ -39,9 +39,7 @@ research_agent = Agent(
role="Information Researcher",
goal="Research topics and demonstrate callback monitoring",
backstory="Expert researcher that demonstrates callback functionality",
tools=[internet_search],
verbose=True
)
tools=[internet_search])
# Create a simple task
research_task = Task(
+7 -9
View File
@@ -1,7 +1,7 @@
import asyncio
import time
from typing import List, Dict
from praisonaiagents import Agent, Task, PraisonAIAgents, TaskOutput
from praisonaiagents import Agent, Task, Agents, TaskOutput
from praisonaiagents.main import (
display_error,
display_interaction,
@@ -98,8 +98,7 @@ sync_agent = Agent(
goal="Perform synchronous searches and return structured results",
backstory="Expert in sync operations and data organization",
tools=[sync_search_tool],
self_reflect=False,
verbose=True,
reflection=False,
markdown=True
)
@@ -109,8 +108,7 @@ async_agent = Agent(
goal="Perform asynchronous searches and return structured results",
backstory="Expert in async operations and data organization",
tools=[async_search_tool],
self_reflect=False,
verbose=True,
reflection=False,
markdown=True
)
@@ -160,7 +158,7 @@ workflow_async_task = Task(
def run_sync_example():
"""Run synchronous example"""
display_instruction("\nRunning Synchronous Example...", console)
agents = PraisonAIAgents(
agents = Agents(
agents=[sync_agent],
tasks=[sync_task],
verbose=1,
@@ -172,7 +170,7 @@ def run_sync_example():
async def run_async_example():
"""Run asynchronous example"""
display_instruction("\nRunning Asynchronous Example...", console)
agents = PraisonAIAgents(
agents = Agents(
agents=[async_agent],
tasks=[async_task],
verbose=1,
@@ -184,7 +182,7 @@ async def run_async_example():
async def run_mixed_example():
"""Run mixed sync/async example"""
display_instruction("\nRunning Mixed Sync/Async Example...", console)
agents = PraisonAIAgents(
agents = Agents(
agents=[sync_agent, async_agent],
tasks=[sync_task, async_task],
verbose=1,
@@ -196,7 +194,7 @@ async def run_mixed_example():
async def run_workflow_example():
"""Run workflow example with both sync and async tasks"""
display_instruction("\nRunning Workflow Example...", console)
agents = PraisonAIAgents(
agents = Agents(
agents=[sync_agent, async_agent],
tasks=[workflow_sync_task, workflow_async_task],
verbose=1,
@@ -1,7 +1,7 @@
import asyncio
import time
from typing import List, Dict
from praisonaiagents import Agent, Task, PraisonAIAgents, TaskOutput
from praisonaiagents import Agent, Task, Agents, TaskOutput
from duckduckgo_search import DDGS
from pydantic import BaseModel
@@ -53,8 +53,7 @@ async_agent = Agent(
goal="Perform fast parallel searches with structured results",
backstory="Expert in efficient data retrieval and parallel search operations",
tools=[async_search_tool],
self_reflect=False,
verbose=True,
reflection=False,
markdown=True
)
@@ -63,8 +62,7 @@ summary_agent = Agent(
role="Research Synthesizer",
goal="Create concise summaries from multiple search results",
backstory="Expert in analyzing and synthesizing information from multiple sources",
self_reflect=True,
verbose=True,
reflection=True,
markdown=True
)
@@ -114,8 +112,8 @@ async def run_parallel_tasks():
context=parallel_tasks
)
# Create a single PraisonAIAgents instance with both agents
agents = PraisonAIAgents(
# Create a single Agents instance with both agents
agents = Agents(
agents=[async_agent, summary_agent],
tasks=parallel_tasks + [summary_task],
verbose=1,
@@ -1,7 +1,7 @@
import asyncio
import time
from typing import List, Dict
from praisonaiagents import Agent, Task, PraisonAIAgents, TaskOutput
from praisonaiagents import Agent, Task, Agents, TaskOutput
from duckduckgo_search import DDGS
from pydantic import BaseModel
@@ -60,8 +60,7 @@ async_agent = Agent(
goal="Perform fast and efficient asynchronous searches with structured results",
backstory="Expert in parallel search operations and data retrieval",
tools=[async_search_tool],
self_reflect=False,
verbose=True,
reflection=False,
markdown=True
)
@@ -72,8 +71,7 @@ summary_agent = Agent(
backstory="""Expert in analyzing and synthesizing information from multiple sources.
Skilled at identifying patterns, trends, and connections between different topics.
Specializes in creating clear, structured summaries that highlight key insights.""",
self_reflect=True, # Enable self-reflection for better summary quality
verbose=True,
reflection=True, # Enable self-reflection for better summary quality
markdown=True
)
@@ -103,7 +101,7 @@ async_task = Task(
async def run_single_task():
"""Run single async task"""
print("\nRunning Single Async Task...")
agents = PraisonAIAgents(
agents = Agents(
agents=[async_agent],
tasks=[async_task],
verbose=1,
@@ -187,7 +185,7 @@ Present the summary in a clear, structured format with sections for findings, pa
)
# First run parallel search tasks
agents = PraisonAIAgents(
agents = Agents(
agents=[async_agent],
tasks=parallel_tasks, # Only run search tasks first
verbose=1,
@@ -212,7 +210,7 @@ Present the summary in a clear, structured format with sections for findings, pa
summary_task.context = completed_tasks
# Run summarization task with summary agent
summary_agents = PraisonAIAgents(
summary_agents = Agents(
agents=[summary_agent], # Use summary agent for synthesis
tasks=[summary_task],
verbose=1,
@@ -22,7 +22,6 @@ agents = AutoAgents(
instructions="Search for information about AI Agents",
tools=[duckduckgo],
process="sequential", # or "hierarchical"
verbose=True,
max_agents=3 # Maximum number of agents to create
)
+3 -4
View File
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
import time
def get_environment_state():
@@ -86,11 +86,10 @@ feedback_task = Task(
)
# Create workflow manager
workflow = PraisonAIAgents(
workflow = Agents(
agents=[llm_caller, action_agent, feedback_agent],
tasks=[monitor_task, action_task, feedback_task],
process="workflow",
verbose=True
process="workflow", output="verbose"
)
def main():
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
import json
from e2b_code_interpreter import Sandbox
@@ -23,15 +23,15 @@ def code_interpreter(code: str):
code_agent = Agent(
name="code_agent",
llm="gpt-5-nano",
llm="gpt-4o-mini",
backstory="Expert in writing Python scripts",
self_reflect=False
reflection=False
)
execution_agent = Agent(
name="execution_agent",
llm="gpt-5-nano",
llm="gpt-4o-mini",
backstory="Expert in executing Python scripts",
self_reflect=False,
reflection=False,
tools=[code_interpreter]
)
@@ -46,5 +46,5 @@ execution_agent_task = Task(
agent=execution_agent
)
agents = PraisonAIAgents(agents=[code_agent, execution_agent], tasks=[code_agent_task, execution_agent_task])
agents = Agents(agents=[code_agent, execution_agent], tasks=[code_agent_task, execution_agent_task])
agents.start()
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
# Create generator and evaluator agents
generator = Agent(
@@ -53,11 +53,10 @@ evaluate_task = Task(
)
# Create workflow manager
workflow = PraisonAIAgents(
workflow = Agents(
agents=[generator, evaluator],
tasks=[generate_task, evaluate_task],
process="workflow",
verbose=True
process="workflow", output="verbose"
)
# Run optimization workflow
+8 -11
View File
@@ -1,7 +1,7 @@
from praisonaiagents import (
Agent,
Task,
PraisonAIAgents,
Agents,
error_logs,
register_display_callback,
sync_display_callbacks,
@@ -124,14 +124,13 @@ def main():
goal="Uncover cutting-edge developments in AI and data science",
backstory="""You are an expert at a technology research group,
skilled in identifying trends and analyzing complex data.""",
verbose=True,
allow_delegation=False,
tools=[internet_search_tool],
llm="gpt-5-nano",
llm="gpt-4o-mini",
markdown=True,
reflect_llm="gpt-5-nano",
min_reflect=2,
max_reflect=4
)
writer = Agent(
@@ -140,9 +139,8 @@ def main():
goal="Craft compelling content on tech advancements",
backstory="""You are a content strategist known for
making complex tech topics interesting and easy to understand.""",
verbose=True,
allow_delegation=True,
llm="gpt-5-nano",
llm="gpt-4o-mini",
tools=[],
markdown=True
)
@@ -190,12 +188,11 @@ def main():
)
# Create and run agents manager
agents = PraisonAIAgents(
agents = Agents(
agents=[researcher, writer],
tasks=[task1, task2, task3, task4],
verbose=True,
process="sequential",
manager_llm="gpt-5-nano"
manager_llm="gpt-4o-mini"
)
agents.start()
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from typing import List, Dict, Union
from duckduckgo_search import DDGS
from langchain_community.tools import YouTubeSearchTool
@@ -112,7 +112,7 @@ data_agent = Agent(
goal="Perform internet searches to collect relevant information.",
backstory="Expert in finding and organizing internet data from multiple sources.",
tools=[internet_search_tool, youtube_search_tool, wikipedia_search_tool],
self_reflect=False
reflection=False
)
# 3. Tasks
@@ -141,7 +141,7 @@ validate_task = Task(
)
# 4. Workflow
agents = PraisonAIAgents(
agents = Agents(
agents=[data_agent],
tasks=[collect_task, validate_task],
verbose=1,
+9 -11
View File
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents, error_logs
from praisonaiagents import Agent, Task, Agents, error_logs
from duckduckgo_search import DDGS
def my_callback(output):
@@ -36,14 +36,13 @@ researcher = Agent(
goal="Uncover cutting-edge developments in AI and data science",
backstory="""You are an expert at a technology research group,
skilled in identifying trends and analyzing complex data.""",
verbose=True,
allow_delegation=False,
tools=[internet_search_tool],
llm="gpt-5-nano",
llm="gpt-4o-mini",
markdown=True,
reflect_llm="gpt-5-nano",
min_reflect=2,
max_reflect=4
)
writer = Agent(
name="Writer",
@@ -51,9 +50,8 @@ writer = Agent(
goal="Craft compelling content on tech advancements",
backstory="""You are a content strategist known for
making complex tech topics interesting and easy to understand.""",
verbose=True,
allow_delegation=True,
llm="gpt-5-nano",
llm="gpt-4o-mini",
tools=[],
markdown=True
)
@@ -99,12 +97,12 @@ task4 = Task(
)
# Create and run agents manager
agents = PraisonAIAgents(
agents = Agents(
agents=[researcher, writer],
tasks=[task1, task2, task3, task4],
verbose=False,
output="minimal",
process="sequential", # "sequential" or "hierarchical"
manager_llm="gpt-5-nano"
manager_llm="gpt-4o-mini"
)
result = agents.start()
@@ -24,7 +24,7 @@ import sys
# Add the local development path to use the current implementation
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', '..', 'src', 'praisonai-agents'))
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from praisonaiagents.tools import python_tools, file_tools, shell_tools
from praisonaiagents.main import register_approval_callback
from praisonaiagents.approval import (
@@ -82,9 +82,7 @@ def main():
name="Security Demo Agent",
role="System Administrator",
goal="Demonstrate human approval for dangerous operations",
tools=[python_tools, file_tools, shell_tools],
verbose=True
)
tools=[python_tools, file_tools, shell_tools])
# Define tasks that will trigger approval prompts
tasks = [
@@ -111,10 +109,9 @@ def main():
]
# Create and run process
process = PraisonAIAgents(
process = Agents(
agents=[agent],
tasks=tasks,
verbose=True
tasks=tasks, output="verbose"
)
print("\n🚀 Starting process with approval-required operations...")
@@ -160,9 +157,7 @@ async def async_demo():
name="Async Demo Agent",
role="Async Operations Specialist",
goal="Demonstrate async approval workflow",
tools=[python_tools],
verbose=True
)
tools=[python_tools])
# This would trigger approval in async context
task = Task(
@@ -171,10 +166,9 @@ async def async_demo():
expected_output="Async execution result"
)
process = PraisonAIAgents(
process = Agents(
agents=[agent],
tasks=[task],
verbose=True
tasks=[task], output="verbose"
)
try:
+4 -5
View File
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from langchain_community.tools import YouTubeSearchTool
from langchain_community.utilities import WikipediaAPIWrapper
@@ -9,7 +9,7 @@ agent = Agent(
goal="Search for information from multiple sources",
backstory="I am an AI assistant that can search YouTube and Wikipedia.",
tools=[YouTubeSearchTool, WikipediaAPIWrapper],
self_reflect=False
reflection=False
)
# Create tasks to demonstrate both tools
@@ -21,10 +21,9 @@ task = Task(
)
# Create and start the workflow
agents = PraisonAIAgents(
agents = Agents(
agents=[agent],
tasks=[task],
verbose=True
tasks=[task], output="verbose"
)
agents.start()
+9 -10
View File
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
import logging
import os
@@ -25,7 +25,7 @@ def main():
# else:
# logger.info("Creating new memory database")
# Create task config (without memory config since it's moved to PraisonAIAgents)
# Create task config (without memory config since it's moved to Agents)
task_config = {}
# Create agents with different roles
@@ -33,14 +33,14 @@ def main():
role="Research Analyst",
goal="Research and document key information about topics",
backstory="Expert at analyzing and storing information in memory",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
retriever = Agent(
role="Information Retriever",
goal="Retrieve and verify stored information from memory",
backstory="Specialist in searching and validating information from memory",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
# Task 1: Process the facts
@@ -100,11 +100,10 @@ def main():
agent=retriever
)
# Initialize PraisonAIAgents with memory configuration
agents = PraisonAIAgents(
# Initialize Agents with memory configuration
agents = Agents(
agents=[researcher, retriever],
tasks=[store_task, verify_task, query_task, query_both_task],
verbose=True, # Use same verbose level as memory
tasks=[store_task, verify_task, query_task, query_both_task], # Use same verbose level as memory
memory=True,
embedder={
"provider": "openai",
@@ -114,10 +113,10 @@ def main():
}
)
# agents = PraisonAIAgents(
# agents = Agents(
# agents=[researcher, retriever],
# tasks=[store_task, verify_task, query_task, query_both_task],
# verbose=True, # Use same verbose level as memory
# # Use same verbose level as memory
# memory=True
# )
+5 -5
View File
@@ -1,4 +1,4 @@
from praisonaiagents.agents.agents import Agent, Task, PraisonAIAgents
from praisonaiagents.agents.agents import Agent, Task, Agents
from praisonaiagents.tools import duckduckgo
# Test facts
@@ -9,19 +9,19 @@ fact3 = "The year the first Josinga was released is 2007"
fact_agent = Agent(
name="Fact Agent",
instructions="You are a fact agent, you store and retrieve facts in memory",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
research_agent = Agent(
name="Research Agent",
instructions="You are a research analyst, you research and document key points about topics",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
blog_agent = Agent(
name="Blog Agent",
instructions="You are a blog writer, you write a blog post about the research",
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
fact_task = Task(
@@ -54,7 +54,7 @@ blog_task = Task(
agent=blog_agent
)
agents = PraisonAIAgents(
agents = Agents(
agents=[fact_agent, research_agent, blog_agent],
tasks=[fact_task, research_task, research_task2, research_task3, research_task4, blog_task],
memory=True
+5 -5
View File
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
# Create Vision Analysis Agent
vision_agent = Agent(
@@ -7,8 +7,8 @@ vision_agent = Agent(
goal="Analyze images and videos to extract meaningful information",
backstory="""You are an expert in computer vision and image analysis.
You excel at describing images, detecting objects, and understanding visual content.""",
llm="gpt-5-nano",
self_reflect=False
llm="gpt-4o-mini",
reflection=False
)
# 1. Task with Image URL
@@ -42,8 +42,8 @@ task3 = Task(
images=["video.mp4"]
)
# Create PraisonAIAgents instance
agents = PraisonAIAgents(
# Create Agents instance
agents = Agents(
agents=[vision_agent],
tasks=[task1, task2, task3],
process="sequential",
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
import time
def get_time_check():
@@ -93,11 +93,10 @@ synthesize_task = Task(
)
# Create workflow manager
workflow = PraisonAIAgents(
workflow = Agents(
agents=[router, worker1, worker2, worker3, synthesizer],
tasks=[router_task, worker1_task, worker2_task, worker3_task, synthesize_task],
process="workflow",
verbose=True
process="workflow", output="verbose"
)
# Run orchestrated workflow
+2 -2
View File
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from datetime import datetime
import asyncio
@@ -75,7 +75,7 @@ aggregate_task = Task(
async def main():
# Create workflow manager
workflow = PraisonAIAgents(
workflow = Agents(
agents=[agent1, agent2, agent3, aggregator],
tasks=[task1, task2, task3, aggregate_task],
process="workflow"
+3 -3
View File
@@ -1,6 +1,6 @@
from praisonaiagents.agent import Agent
from praisonaiagents.task import Task
from praisonaiagents.agents import PraisonAIAgents
from praisonaiagents.agents import Agents
from typing import List, Dict
import time
@@ -64,11 +64,11 @@ final_task = Task(
)
# Create the workflow manager
workflow = PraisonAIAgents(
workflow = Agents(
agents=[agent1, agent2, agent3],
tasks=[initial_task, analysis_task, final_task],
process="workflow", # Use workflow process type
verbose=True
output="verbose"
)
# Run the workflow
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents, Tools
from praisonaiagents import Agent, Task, Agents, Tools
from pydantic import BaseModel
class AnalysisReport(BaseModel):
@@ -12,10 +12,9 @@ researcher = Agent(
role="Technology Research Analyst",
goal="Analyze and structure information about AI developments",
backstory="Expert analyst specializing in AI technology trends",
verbose=True,
llm="gpt-5-nano",
llm="gpt-4o-mini",
tools=[Tools.internet_search],
self_reflect=False
reflection=False
)
# Create an analyst agent
@@ -24,9 +23,8 @@ analyst = Agent(
role="Data Insights Specialist",
goal="Structure and analyze research findings",
backstory="Senior data analyst with expertise in pattern recognition",
verbose=True,
llm="gpt-5-nano",
self_reflect=False
llm="gpt-4o-mini",
reflection=False
)
# Define structured tasks
@@ -46,11 +44,11 @@ analysis_task = Task(
)
# Initialize and run agents
agents = PraisonAIAgents(
agents = Agents(
agents=[researcher, analyst],
tasks=[research_task, analysis_task],
process="sequential",
verbose=True
output="verbose"
)
result = agents.start()
print(result)
@@ -33,9 +33,9 @@ def get_structured_analysis(query: str, verbose: bool = True) -> AnalysisReport:
goal="Analyze topics and provide structured reports",
backstory="Expert at gathering information and providing structured analysis",
verbose=verbose,
self_reflect=True, # Enable self-reflection for better quality
reflection=True, # Enable self-reflection for better quality
markdown=True,
llm="gpt-5-nano"
llm="gpt-4o-mini"
)
# Create the analysis prompt
+4 -5
View File
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
import os
import requests
from typing import Any, Dict, List, Optional
@@ -94,7 +94,7 @@ agent = Agent(
goal="Search for information about 'AI Agents Framework'",
backstory="I am an AI assistant that can search GitHub.",
tools=[EXASearchTool],
self_reflect=False
reflection=False
)
# Create tasks to demonstrate both tools
@@ -106,10 +106,9 @@ task = Task(
)
# Create and start the workflow
agents = PraisonAIAgents(
agents = Agents(
agents=[agent],
tasks=[task],
verbose=True
tasks=[task], output="verbose"
)
agents.start()
+3 -3
View File
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from praisonaiagents.tools import get_article, get_news_sources, get_articles_from_source, get_trending_topics
# Create Wikipedia agent
@@ -8,7 +8,7 @@ news_agent = Agent(
goal="Collect and analyze news articles from various sources.",
backstory="Expert in news gathering and content analysis.",
tools=[get_article, get_news_sources, get_articles_from_source, get_trending_topics],
self_reflect=False
reflection=False
)
@@ -22,7 +22,7 @@ news_task = Task(
# Run agent
agents = PraisonAIAgents(
agents = Agents(
agents=[news_agent],
tasks=[news_task],
process="sequential"
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from typing import List, Dict
from duckduckgo_search import DDGS
@@ -35,7 +35,7 @@ data_agent = Agent(
goal="Perform internet searches to collect relevant information.",
backstory="Expert in finding and organising internet data.",
tools=[internet_search_tool],
self_reflect=False
reflection=False
)
# 3. Tasks
@@ -64,7 +64,7 @@ validate_task = Task(
)
# 4. AI Agents Workflow
agents = PraisonAIAgents(
agents = Agents(
agents=[data_agent],
tasks=[collect_task, validate_task],
verbose=1,
@@ -1,4 +1,4 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
import random
from typing import List, Dict, Union
import json
@@ -158,7 +158,7 @@ data_agent = Agent(
goal="Collect and validate data about individuals",
backstory="Expert in gathering and validating demographic data",
tools=[random_data_of_individuals],
self_reflect=False
reflection=False
)
process_agent = Agent(
@@ -167,7 +167,7 @@ process_agent = Agent(
goal="Process and categorize individual data",
backstory="Expert in data analysis and categorization",
tools=[process_data_of_individuals],
self_reflect=False
reflection=False
)
# Modify tasks to use Pydantic models
@@ -224,8 +224,8 @@ process_task = Task(
output_json=ProcessedPerson
)
# Create PraisonAIAgents instance with workflow process
workflow = PraisonAIAgents(
# Create Agents instance with workflow process
workflow = Agents(
agents=[data_agent, process_agent],
tasks=[collect_task, validate_task, process_task],
verbose=1,
@@ -12,7 +12,7 @@ Features demonstrated:
- Quality assurance workflows
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from praisonaiagents.task import TaskOutput
from typing import Tuple, Any
import re
@@ -125,7 +125,7 @@ email_task = Task(
description="Write a business proposal for a new software project, including contact information",
expected_output="Professional business proposal with contact email addresses",
agent=email_agent,
guardrail=email_format_guardrail,
guardrails=email_format_guardrail,
max_retries=3
)
@@ -135,7 +135,7 @@ content_task = Task(
description="Write an analysis of cloud computing benefits for small businesses",
expected_output="Comprehensive analysis of cloud computing benefits (minimum 50 words)",
agent=content_agent,
guardrail=word_count_guardrail,
guardrails=word_count_guardrail,
max_retries=3
)
@@ -145,7 +145,7 @@ business_task = Task(
description="Provide advice on handling difficult client relationships",
expected_output="Professional advice maintaining appropriate business tone",
agent=business_agent,
guardrail=professional_tone_guardrail,
guardrails=professional_tone_guardrail,
max_retries=3
)
@@ -155,7 +155,7 @@ research_task = Task(
description="Analyze the impact of artificial intelligence on job markets",
expected_output="Research-backed analysis with citations and evidence",
agent=research_agent,
guardrail=factual_accuracy_guardrail,
guardrails=factual_accuracy_guardrail,
max_retries=3
)
@@ -176,15 +176,14 @@ llm_guardrail_task = Task(
agent=llm_guardrail_agent,
# LLM-based guardrail using natural language validation
guardrail="Validate that this marketing content is: 1) Factually accurate with no false claims, 2) Compliant with advertising standards (no misleading statements), 3) Engaging and professional in tone, 4) Includes a clear call-to-action. If any criteria are not met, explain what needs to be improved.",
guardrails="Validate that this marketing content is: 1) Factually accurate with no false claims, 2) Compliant with advertising standards (no misleading statements), 3) Engaging and professional in tone, 4) Includes a clear call-to-action. If any criteria are not met, explain what needs to be improved.",
max_retries=2
)
# Execute tasks with guardrails
agents = PraisonAIAgents(
agents = Agents(
agents=[email_agent, content_agent, business_agent, research_agent, llm_guardrail_agent],
tasks=[email_task, content_task, business_task, research_task, llm_guardrail_task],
verbose=True
tasks=[email_task, content_task, business_task, research_task, llm_guardrail_task], output="verbose"
)
print("="*80)
@@ -5,7 +5,7 @@ This example demonstrates production guardrail patterns using PraisonAI's
built-in validation and safety mechanisms for secure agent operations.
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from praisonaiagents.tools import internet_search
print("=== Production Guardrails Patterns Example ===\n")
@@ -38,10 +38,8 @@ production_agent = Agent(
goal="Generate safe, compliant content for production use",
backstory="Production-ready agent with built-in safety and compliance checks",
tools=[internet_search],
guardrail=content_guardrail,
max_retries=3,
verbose=True
)
guardrails=content_guardrail,
max_retries=3)
# Create secure task with validation
secure_task = Task(
@@ -0,0 +1,182 @@
"""
Unified Handoff Configuration Example.
Demonstrates the new HandoffConfig features:
- Context policies (full, summary, none, last_n)
- Cycle detection
- Max depth limiting
- Timeout configuration
- Programmatic handoff via Agent.handoff_to()
"""
from praisonaiagents import Agent, handoff, HandoffConfig, ContextPolicy
# Create specialized agents
research_agent = Agent(
name="Research Agent",
role="Research Specialist",
goal="Conduct thorough research on topics",
backstory="Expert researcher with access to comprehensive knowledge bases."
)
writer_agent = Agent(
name="Writer Agent",
role="Content Writer",
goal="Create well-written content based on research",
backstory="Professional writer skilled in creating engaging content."
)
editor_agent = Agent(
name="Editor Agent",
role="Content Editor",
goal="Review and polish written content",
backstory="Experienced editor with keen eye for quality and clarity."
)
# Create coordinator with handoffs using new config options
coordinator = Agent(
name="Coordinator",
role="Project Coordinator",
goal="Coordinate tasks between specialized agents",
backstory="Expert at understanding requirements and delegating to the right specialist.",
instructions="""Analyze the request and delegate to the appropriate specialist:
- For research tasks, use transfer_to_research_agent
- For writing tasks, use transfer_to_writer_agent
- For editing tasks, use transfer_to_editor_agent
Always explain your delegation decision.""",
handoffs=[
# Handoff with summary context policy (default - safe)
handoff(
research_agent,
context_policy="summary",
timeout_seconds=120,
max_depth=5,
),
# Handoff with full context
handoff(
writer_agent,
context_policy="full",
timeout_seconds=180,
),
# Handoff with custom config
handoff(
editor_agent,
config=HandoffConfig(
context_policy=ContextPolicy.LAST_N,
max_context_messages=5,
detect_cycles=True,
max_depth=3,
),
),
]
)
def demo_llm_driven_handoff():
"""Demonstrate LLM-driven handoff via tool calls."""
print("=" * 60)
print("Demo 1: LLM-Driven Handoff")
print("=" * 60)
response = coordinator.chat("I need to research the latest trends in AI agents")
print(f"\nFinal Response:\n{response}")
def demo_programmatic_handoff():
"""Demonstrate programmatic handoff via Agent.handoff_to()."""
print("\n" + "=" * 60)
print("Demo 2: Programmatic Handoff (Agent.handoff_to)")
print("=" * 60)
# Create a simple agent for programmatic handoff
source_agent = Agent(
name="Source Agent",
role="Task Initiator",
goal="Initiate tasks and hand them off",
)
target_agent = Agent(
name="Target Agent",
role="Task Executor",
goal="Execute delegated tasks",
)
# Programmatic handoff with custom config
config = HandoffConfig(
context_policy=ContextPolicy.SUMMARY,
timeout_seconds=60,
detect_cycles=True,
)
result = source_agent.handoff_to(
target_agent,
prompt="Summarize the key benefits of multi-agent systems",
config=config,
)
print(f"\nHandoff Result:")
print(f" Success: {result.success}")
print(f" Target: {result.target_agent}")
print(f" Duration: {result.duration_seconds:.2f}s")
if result.response:
print(f" Response: {result.response[:200]}...")
def demo_handoff_chain():
"""Demonstrate a handoff chain with depth tracking."""
print("\n" + "=" * 60)
print("Demo 3: Handoff Chain")
print("=" * 60)
# Create a chain: Analyst -> Researcher -> Writer
analyst = Agent(
name="Analyst",
role="Requirements Analyst",
goal="Analyze requirements and delegate research",
)
researcher = Agent(
name="Researcher",
role="Research Specialist",
goal="Research topics and provide findings",
)
writer = Agent(
name="Writer",
role="Technical Writer",
goal="Write documentation based on research",
)
# Chain: Analyst hands off to Researcher
result1 = analyst.handoff_to(
researcher,
prompt="Research best practices for agent handoffs",
config=HandoffConfig(max_depth=5),
)
print(f"\nStep 1 - Analyst -> Researcher:")
print(f" Success: {result1.success}")
print(f" Depth: {result1.handoff_depth}")
# Researcher hands off to Writer
if result1.success:
result2 = researcher.handoff_to(
writer,
prompt=f"Write documentation based on: {result1.response[:100] if result1.response else 'research findings'}",
config=HandoffConfig(max_depth=5),
)
print(f"\nStep 2 - Researcher -> Writer:")
print(f" Success: {result2.success}")
print(f" Depth: {result2.handoff_depth}")
if __name__ == "__main__":
print("\n🤝 Unified Handoff Configuration Demo\n")
# Run demos
demo_llm_driven_handoff()
demo_programmatic_handoff()
demo_handoff_chain()
print("\n" + "=" * 60)
print("All demos completed!")
print("=" * 60)
@@ -5,7 +5,7 @@ This example demonstrates an advanced system that dynamically creates agents and
based on user inputs, including conditional logic, custom configurations, and file output.
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from praisonaiagents.tools import duckduckgo
import os
from typing import Dict, List
@@ -16,7 +16,7 @@ class DynamicAgentSystem:
def __init__(self):
self.user_preferences = {}
self.llm_config = {
"model": os.getenv("LLM_MODEL", "gpt-5-nano"),
"model": os.getenv("LLM_MODEL", "gpt-4o-mini"),
"temperature": 0.7
}
@@ -40,7 +40,7 @@ class DynamicAgentSystem:
goal=f"Conduct {inputs['depth']} research on {inputs['topic']} in {inputs['language']}",
backstory=f"Multilingual expert specializing in {inputs['depth']} analysis",
tools=[duckduckgo],
self_reflect=inputs['depth'] == 'detailed',
reflection=inputs['depth'] == 'detailed',
llm=self.llm_config
)
agents.append(research_agent)
@@ -137,7 +137,7 @@ class DynamicAgentSystem:
# Run the system
print(f"\n🚀 Starting {process} analysis for '{inputs['topic']}'...")
praison_agents = PraisonAIAgents(
praison_agents = Agents(
agents=agents,
tasks=tasks,
process=process,
@@ -6,14 +6,14 @@ to create flexible agent configurations that can be pre-configured via environme
"""
import os
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
# Set via environment or use default
default_topic = os.getenv("RESEARCH_TOPIC", "AI trends")
user_topic = input(f"Topic to research [{default_topic}]: ") or default_topic
# Get other configuration from environment
llm_model = os.getenv("LLM_MODEL", "gpt-5-nano")
llm_model = os.getenv("LLM_MODEL", "gpt-4o-mini")
temperature = float(os.getenv("LLM_TEMPERATURE", "0.7"))
max_retries = int(os.getenv("MAX_RETRIES", "3"))
@@ -46,7 +46,7 @@ print(f" - Max Retries: {max_retries}")
print("\n" + "="*50 + "\n")
# Run agents
agents = PraisonAIAgents(agents=[agent], tasks=[task])
agents = Agents(agents=[agent], tasks=[task])
result = agents.start()
# Save to environment-specified location if provided
@@ -5,7 +5,7 @@ This example shows how to use multiple user inputs to create a more sophisticate
agent system with tools, multiple agents, and context passing between tasks.
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
from praisonaiagents.tools import duckduckgo
# Get multiple inputs
@@ -19,7 +19,7 @@ search_agent = Agent(
goal=f"Find {num_results} relevant results about {topic}",
backstory="Expert in internet research and data collection",
tools=[duckduckgo],
self_reflect=False
reflection=False
)
# Create analysis agent
@@ -47,7 +47,7 @@ analyze_task = Task(
)
# Run sequential process
agents = PraisonAIAgents(
agents = Agents(
agents=[search_agent, analysis_agent],
tasks=[search_task, analyze_task],
process="sequential"
+2 -2
View File
@@ -5,7 +5,7 @@ This example demonstrates how to use basic user input to create dynamic agents a
The user is prompted for a search query, and the agent searches for information about it.
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import Agent, Task, Agents
# Get user input
user_query = input("What would you like to search for? ")
@@ -26,5 +26,5 @@ task = Task(
)
# Run
agents = PraisonAIAgents(agents=[agent], tasks=[task])
agents = Agents(agents=[agent], tasks=[task])
agents.start()
Loaded 100 of 524 files, more files were not shown because too many files have changed in this diff. Show more