Compare commits

...
1 Commits
Author SHA1 Message Date
Mervin Praison 92397615f8 Release v2.3.65 2025-12-20 21:49:04 +00:00
16 changed files with 766 additions and 58 deletions

No files matched your search

+24 -25
View File
@@ -267,9 +267,7 @@ npm install praisonai
| ↳ Git Integration | [Example](examples/python/cli/git_integration_example.py) | [📖](https://docs.praison.ai/docs/cli/git-integration) |
| ↳ Sandbox Execution | [Example](examples/python/cli/sandbox_execution_example.py) | [📖](https://docs.praison.ai/docs/cli/sandbox-execution) |
| **⏰ 24/7 Scheduling** | | |
| ↳ Agent Scheduler | [Example](examples/python/scheduled_agents/news_checker_live.py) | [📖](https://docs.praison.ai/scheduler/overview) |
| ↳ Scheduled News Checker | [Example](examples/python/scheduled_agents/news_checker_agent.py) | [📖](https://docs.praison.ai/scheduler/examples) |
| ↳ Simple Scheduler Test | [Example](examples/python/scheduled_agents/simple_test.py) | [📖](https://docs.praison.ai/scheduler/quickstart) |
| ↳ Agent Scheduler | [Example](examples/python/scheduled_agents/news_checker_live.py) | [📖](https://docs.praison.ai/docs/cli/scheduler) |
---
@@ -1248,32 +1246,33 @@ praisonai "Complex task" --metrics --planning
### Scheduler CLI:
```bash
# Schedule an agent to run periodically (24/7 operation)
praisonai schedule --interval hourly --task "Check latest AI news"
# Schedule an agent using agents.yaml
praisonai schedule agents.yaml
# Schedule with custom interval (every 30 minutes)
praisonai schedule --interval "*/30m" --task "Monitor system status"
# Override interval from CLI
praisonai schedule agents.yaml --interval "*/30m"
# Schedule with YAML config
praisonai schedule --config scheduler_config.yaml
# Schedule with immediate execution
praisonai schedule --interval daily --task "Generate report" --run-now
# Schedule with retry configuration
praisonai schedule --interval "*/6h" --task "Collect data" --max-retries 5
# View scheduler help
praisonai schedule --help
# With verbose logging
praisonai schedule agents.yaml --verbose
```
**Scheduler Intervals:**
- `hourly` - Every hour
- `daily` - Every 24 hours
- `*/30m` - Every 30 minutes
- `*/6h` - Every 6 hours
- `*/5s` - Every 5 seconds (for testing)
- `3600` - Custom seconds
**agents.yaml with schedule:**
```yaml
framework: praisonai
agents:
- name: "AI News Monitor"
instructions: "Search and summarize AI news"
tools:
- search_tool
task: "Search for latest AI news"
schedule:
interval: "hourly" # hourly, daily, */30m, */6h
max_retries: 3
run_immediately: true
```
### Image Processing CLI:
```bash
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.64" \
"praisonai>=2.3.65" \
"praisonai[chat]" \
"embedchain[github,youtube]"
+1 -1
View File
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.64" \
"praisonai>=2.3.65" \
"praisonai[ui]" \
"praisonai[chat]" \
"praisonai[realtime]" \
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.64" \
"praisonai>=2.3.65" \
"praisonai[ui]" \
"praisonai[crewai]"
@@ -0,0 +1,23 @@
# Example: Data Collection Agent with 30-minute Schedule
# Collects and analyzes data every 30 minutes
framework: praisonai
agents:
- name: "Data Collector"
role: "Data Analysis Specialist"
goal: "Collect and analyze market data at regular intervals"
instructions: |
You are a data analysis specialist focused on collecting and analyzing
market trends and metrics. Provide actionable insights from the data.
tools:
- search_tool
verbose: false
task: "Collect latest market data and identify key trends"
# Schedule: Every 30 minutes with immediate execution
schedule:
interval: "*/30m"
max_retries: 5
run_immediately: true
@@ -0,0 +1,53 @@
# Example: 24/7 AI News Monitor with Scheduler
# This agent runs continuously at regular intervals to monitor AI news
framework: praisonai
agents:
- name: "AI News Monitor"
role: "Technology News Analyst"
goal: "Monitor and summarize the latest AI developments"
instructions: |
You are an expert technology news analyst specializing in AI and machine learning.
Your job is to search for the latest AI news, identify the most important developments,
and provide clear, concise summaries with sources.
Focus on:
- Major AI product launches
- Research breakthroughs
- Industry trends
- Policy and regulation updates
tools:
- search_tool
verbose: true
# Task to execute on each scheduled run
task: |
Search for the latest AI news from the past 24 hours.
Provide the top 3 most important stories with:
1. Headline
2. Brief summary (2-3 sentences)
3. Source URL
4. Why it's significant
# Schedule configuration
schedule:
# Interval options:
# - "hourly" - Every hour
# - "daily" - Every 24 hours
# - "*/30m" - Every 30 minutes
# - "*/6h" - Every 6 hours
# - "3600" - Custom seconds
interval: "hourly"
# Maximum retry attempts if execution fails
max_retries: 3
# Run immediately when scheduler starts (before first scheduled interval)
run_immediately: true
# Optional: Variables for customization
variables:
news_category: "AI and Machine Learning"
region: "Global"
time_range: "24 hours"
+24 -25
View File
@@ -267,9 +267,7 @@ npm install praisonai
| ↳ Git Integration | [Example](examples/python/cli/git_integration_example.py) | [📖](https://docs.praison.ai/docs/cli/git-integration) |
| ↳ Sandbox Execution | [Example](examples/python/cli/sandbox_execution_example.py) | [📖](https://docs.praison.ai/docs/cli/sandbox-execution) |
| **⏰ 24/7 Scheduling** | | |
| ↳ Agent Scheduler | [Example](examples/python/scheduled_agents/news_checker_live.py) | [📖](https://docs.praison.ai/scheduler/overview) |
| ↳ Scheduled News Checker | [Example](examples/python/scheduled_agents/news_checker_agent.py) | [📖](https://docs.praison.ai/scheduler/examples) |
| ↳ Simple Scheduler Test | [Example](examples/python/scheduled_agents/simple_test.py) | [📖](https://docs.praison.ai/scheduler/quickstart) |
| ↳ Agent Scheduler | [Example](examples/python/scheduled_agents/news_checker_live.py) | [📖](https://docs.praison.ai/docs/cli/scheduler) |
---
@@ -1248,32 +1246,33 @@ praisonai "Complex task" --metrics --planning
### Scheduler CLI:
```bash
# Schedule an agent to run periodically (24/7 operation)
praisonai schedule --interval hourly --task "Check latest AI news"
# Schedule an agent using agents.yaml
praisonai schedule agents.yaml
# Schedule with custom interval (every 30 minutes)
praisonai schedule --interval "*/30m" --task "Monitor system status"
# Override interval from CLI
praisonai schedule agents.yaml --interval "*/30m"
# Schedule with YAML config
praisonai schedule --config scheduler_config.yaml
# Schedule with immediate execution
praisonai schedule --interval daily --task "Generate report" --run-now
# Schedule with retry configuration
praisonai schedule --interval "*/6h" --task "Collect data" --max-retries 5
# View scheduler help
praisonai schedule --help
# With verbose logging
praisonai schedule agents.yaml --verbose
```
**Scheduler Intervals:**
- `hourly` - Every hour
- `daily` - Every 24 hours
- `*/30m` - Every 30 minutes
- `*/6h` - Every 6 hours
- `*/5s` - Every 5 seconds (for testing)
- `3600` - Custom seconds
**agents.yaml with schedule:**
```yaml
framework: praisonai
agents:
- name: "AI News Monitor"
instructions: "Search and summarize AI news"
tools:
- search_tool
task: "Search for latest AI news"
schedule:
interval: "hourly" # hourly, daily, */30m, */6h
max_retries: 3
run_immediately: true
```
### Image Processing CLI:
```bash
+2 -2
View File
@@ -3,8 +3,8 @@ class Praisonai < Formula
desc "AI tools for various AI applications"
homepage "https://github.com/MervinPraison/PraisonAI"
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.64.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.64.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.65.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.65.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
@@ -0,0 +1,182 @@
"""
Agent Scheduler CLI Feature.
Provides CLI commands for scheduling agents to run 24/7 at regular intervals.
"""
import logging
logger = logging.getLogger(__name__)
class AgentSchedulerHandler:
"""Handler for agent scheduler CLI commands."""
@staticmethod
def handle_schedule_command(args) -> int:
"""
Handle the schedule command for running agents periodically.
Args:
args: Parsed command-line arguments
Returns:
Exit code (0 for success, 1 for error)
"""
try:
from praisonai.scheduler import AgentScheduler
except ImportError:
print("Error: praisonai.scheduler module not found")
print("Please ensure PraisonAI is properly installed")
return 1
# Get YAML file path (default to agents.yaml)
yaml_path = getattr(args, 'schedule_yaml', None) or 'agents.yaml'
# Get overrides from CLI
interval_override = getattr(args, 'schedule_interval', None)
max_retries_override = getattr(args, 'schedule_max_retries', None)
verbose = getattr(args, 'verbose', False)
# Set up logging
if verbose:
logging.basicConfig(
level=logging.INFO,
format='[%(asctime)s] %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
try:
# Create scheduler from YAML
print(f"🤖 Loading agent configuration from: {yaml_path}")
scheduler = AgentScheduler.from_yaml(
yaml_path=yaml_path,
interval_override=interval_override,
max_retries_override=max_retries_override
)
# Display configuration
agent_name = getattr(scheduler.agent, 'name', 'Agent')
print(f"\n{'='*60}")
print(f"🤖 Starting 24/7 {agent_name}")
print(f"{'='*60}")
print(f"Task: {scheduler.task[:80]}{'...' if len(scheduler.task) > 80 else ''}")
# Get schedule info
schedule_config = scheduler._yaml_schedule_config
interval = interval_override or schedule_config.get('interval', 'hourly')
max_retries = max_retries_override or schedule_config.get('max_retries', 3)
print(f"Schedule: {interval}")
print(f"Max Retries: {max_retries}")
print(f"{'='*60}\n")
# Start scheduler
print("⏰ Starting scheduler... (Press Ctrl+C to stop)\n")
scheduler.start_from_yaml_config()
# Keep running until interrupted
try:
while scheduler.is_running:
import time
time.sleep(1)
except KeyboardInterrupt:
print("\n\n🛑 Stopping scheduler...")
scheduler.stop()
# Display final stats
stats = scheduler.get_stats()
print("\n📊 Final Statistics:")
print(f" Total Executions: {stats['total_executions']}")
print(f" Successful: {stats['successful_executions']}")
print(f" Failed: {stats['failed_executions']}")
print(f" Success Rate: {stats['success_rate']:.1f}%")
print("\n✅ Agent stopped successfully\n")
return 0
except FileNotFoundError as e:
print(f"❌ Error: {e}")
print(f"\nMake sure {yaml_path} exists in the current directory.")
print("\nExample agents.yaml structure:")
print("""
framework: praisonai
agents:
- name: "AI News Monitor"
role: "News Analyst"
instructions: "Search and summarize AI news"
tools:
- search_tool
task: "Search for latest AI news"
schedule:
interval: "hourly"
max_retries: 3
run_immediately: true
""")
return 1
except ValueError as e:
print(f"❌ Configuration Error: {e}")
return 1
except Exception as e:
print(f"❌ Error: {e}")
if verbose:
import traceback
traceback.print_exc()
return 1
@staticmethod
def add_schedule_arguments(subparsers):
"""
Add schedule subcommand and arguments to parser.
Args:
subparsers: Subparsers object from argparse
"""
schedule_parser = subparsers.add_parser(
'schedule',
help='Schedule an agent to run continuously at regular intervals'
)
schedule_parser.add_argument(
'schedule_yaml',
nargs='?',
default='agents.yaml',
help='Path to agents.yaml file (default: agents.yaml)'
)
schedule_parser.add_argument(
'--interval',
dest='schedule_interval',
type=str,
help='Override schedule interval (e.g., "hourly", "*/30m", "daily")'
)
schedule_parser.add_argument(
'--max-retries',
dest='schedule_max_retries',
type=int,
help='Override maximum retry attempts (default: from YAML or 3)'
)
schedule_parser.add_argument(
'--verbose', '-v',
action='store_true',
help='Enable verbose logging'
)
schedule_parser.set_defaults(func=AgentSchedulerHandler.handle_schedule_command)
def setup_scheduler_cli(subparsers):
"""
Setup scheduler CLI commands.
Args:
subparsers: Subparsers object from argparse
"""
AgentSchedulerHandler.add_schedule_arguments(subparsers)
+10 -1
View File
@@ -223,7 +223,12 @@ class PraisonAI:
file_input = self.read_file_if_provided(getattr(args, 'file', None))
if args.command:
if args.command.startswith("tests.test") or args.command.startswith("tests/test"): # Argument used for testing purposes
# Handle schedule command
if args.command == "schedule":
from praisonai.cli.features.agent_scheduler import AgentSchedulerHandler
exit_code = AgentSchedulerHandler.handle_schedule_command(args)
sys.exit(exit_code)
elif args.command.startswith("tests.test") or args.command.startswith("tests/test"): # Argument used for testing purposes
print("test")
return "test"
else:
@@ -710,6 +715,10 @@ class PraisonAI:
parser.add_argument("--external-agent", type=str, choices=["claude", "gemini", "codex", "cursor"],
help="Use external AI CLI tool (claude, gemini, codex, cursor)")
# Agent Scheduler - for schedule command
parser.add_argument("--interval", dest="schedule_interval", type=str, help="Schedule interval (e.g., 'hourly', '*/30m', 'daily')")
parser.add_argument("--max-retries", dest="schedule_max_retries", type=int, help="Maximum retry attempts for scheduled execution")
# If we're in a test environment, parse with empty args to avoid pytest interference
if in_test_env:
args, unknown_args = parser.parse_known_args([])
@@ -0,0 +1,54 @@
"""
Simple CLI entry point for praisonai schedule command.
This provides a standalone entry point that can be called from main.py
"""
import sys
from praisonai.cli.features.agent_scheduler import AgentSchedulerHandler
def main():
"""Main entry point for schedule CLI command."""
import argparse
parser = argparse.ArgumentParser(
description='Schedule an agent to run continuously at regular intervals'
)
parser.add_argument(
'schedule_yaml',
nargs='?',
default='agents.yaml',
help='Path to agents.yaml file (default: agents.yaml)'
)
parser.add_argument(
'--interval',
dest='schedule_interval',
type=str,
help='Override schedule interval (e.g., "hourly", "*/30m", "daily")'
)
parser.add_argument(
'--max-retries',
dest='schedule_max_retries',
type=int,
help='Override maximum retry attempts (default: from YAML or 3)'
)
parser.add_argument(
'--verbose', '-v',
action='store_true',
help='Enable verbose logging'
)
args = parser.parse_args()
# Call the handler
exit_code = AgentSchedulerHandler.handle_schedule_command(args)
sys.exit(exit_code)
if __name__ == '__main__':
main()
+1 -1
View File
@@ -57,7 +57,7 @@ class CloudDeployer:
file.write("FROM python:3.11-slim\n")
file.write("WORKDIR /app\n")
file.write("COPY . .\n")
file.write("RUN pip install flask praisonai==2.3.64 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==2.3.65 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
@@ -230,6 +230,103 @@ class AgentScheduler:
except Exception as e:
logger.error(f"One-time execution failed: {e}")
raise
@classmethod
def from_yaml(
cls,
yaml_path: str,
interval_override: Optional[str] = None,
max_retries_override: Optional[int] = None,
on_success: Optional[Callable] = None,
on_failure: Optional[Callable] = None
) -> 'AgentScheduler':
"""
Create AgentScheduler from agents.yaml file.
Args:
yaml_path: Path to agents.yaml file
interval_override: Override schedule interval from YAML
max_retries_override: Override max_retries from YAML
on_success: Callback function on successful execution
on_failure: Callback function on failed execution
Returns:
Configured AgentScheduler instance
Example:
scheduler = AgentScheduler.from_yaml("agents.yaml")
scheduler.start()
Example agents.yaml:
framework: praisonai
agents:
- name: "AI News Monitor"
role: "News Analyst"
instructions: "Search and summarize AI news"
tools:
- search_tool
task: "Search for latest AI news"
schedule:
interval: "hourly"
max_retries: 3
run_immediately: true
"""
from .yaml_loader import load_agent_yaml_with_schedule, create_agent_from_config
# Load configuration from YAML
agent_config, schedule_config = load_agent_yaml_with_schedule(yaml_path)
# Create agent from config
agent = create_agent_from_config(agent_config)
# Get task
task = agent_config.get('task', '')
if not task:
raise ValueError("No task specified in YAML file")
# Create scheduler
scheduler = cls(
agent=agent,
task=task,
config=agent_config,
on_success=on_success,
on_failure=on_failure
)
# Store schedule config for auto-start
scheduler._yaml_schedule_config = schedule_config
scheduler._interval_override = interval_override
scheduler._max_retries_override = max_retries_override
return scheduler
def start_from_yaml_config(self) -> bool:
"""
Start scheduler using configuration from YAML file.
Must be called after from_yaml() class method.
Returns:
True if scheduler started successfully
"""
if not hasattr(self, '_yaml_schedule_config'):
raise RuntimeError("start_from_yaml_config() can only be called after from_yaml()")
schedule_config = self._yaml_schedule_config
# Use overrides if provided, otherwise use YAML config
interval = self._interval_override or schedule_config.get('interval', 'hourly')
max_retries = self._max_retries_override or schedule_config.get('max_retries', 3)
run_immediately = schedule_config.get('run_immediately', False)
return self.start(
schedule_expr=interval,
max_retries=max_retries,
run_immediately=run_immediately
)
def create_agent_scheduler(
@@ -0,0 +1,197 @@
"""
YAML loader for agent scheduler configuration.
Loads agent and schedule configuration from agents.yaml files.
"""
import yaml
import logging
from typing import Dict, Any, Tuple
from pathlib import Path
logger = logging.getLogger(__name__)
def load_agent_yaml_with_schedule(yaml_path: str) -> Tuple[Dict[str, Any], Dict[str, Any]]:
"""
Load agents.yaml file and extract agent and schedule configuration.
Args:
yaml_path: Path to agents.yaml file
Returns:
Tuple of (agent_config, schedule_config)
Raises:
FileNotFoundError: If YAML file doesn't exist
ValueError: If YAML is invalid or missing required fields
Example YAML structure:
framework: praisonai
agents:
- name: "AI News Monitor"
role: "News Analyst"
goal: "Monitor AI news"
instructions: "Search and summarize"
tools:
- search_tool
verbose: true
task: "Search for latest AI news"
schedule:
interval: "hourly"
max_retries: 3
run_immediately: true
"""
yaml_path = Path(yaml_path)
if not yaml_path.exists():
raise FileNotFoundError(f"YAML file not found: {yaml_path}")
try:
with open(yaml_path, 'r') as f:
config = yaml.safe_load(f)
except yaml.YAMLError as e:
raise ValueError(f"Invalid YAML file: {e}")
if not config:
raise ValueError("Empty YAML file")
# Extract agent configuration
agent_config = {}
# Get agents list (can be single agent or multiple)
agents = config.get('agents', [])
if not agents:
raise ValueError("No agents defined in YAML file")
# Use first agent for scheduling (can be extended for multi-agent later)
if isinstance(agents, list):
agent_config = agents[0]
else:
agent_config = agents
# Get task
task = config.get('task', '')
if not task:
# Try to get from agent's tasks
if 'tasks' in agent_config and agent_config['tasks']:
first_task = list(agent_config['tasks'].values())[0]
task = first_task.get('description', '')
agent_config['task'] = task
# Get framework (default to praisonai)
agent_config['framework'] = config.get('framework', 'praisonai')
# Extract schedule configuration (optional)
schedule_config = config.get('schedule', {})
# Set defaults for schedule if not provided
if not schedule_config:
schedule_config = {
'interval': 'hourly',
'max_retries': 3,
'run_immediately': False
}
else:
# Ensure required fields have defaults
schedule_config.setdefault('interval', 'hourly')
schedule_config.setdefault('max_retries', 3)
schedule_config.setdefault('run_immediately', False)
logger.info(f"Loaded agent '{agent_config.get('name', 'Unknown')}' with schedule interval '{schedule_config['interval']}'")
return agent_config, schedule_config
def create_agent_from_config(agent_config: Dict[str, Any]) -> Any:
"""
Create a PraisonAI Agent instance from configuration.
Args:
agent_config: Agent configuration dictionary
Returns:
Agent instance
Raises:
ImportError: If praisonaiagents is not installed
"""
try:
from praisonaiagents import Agent
except ImportError:
raise ImportError("praisonaiagents package is required. Install with: pip install praisonaiagents")
# Extract agent parameters
name = agent_config.get('name', 'Scheduled Agent')
role = agent_config.get('role', '')
goal = agent_config.get('goal', '')
instructions = agent_config.get('instructions', agent_config.get('backstory', ''))
verbose = agent_config.get('verbose', False)
# Handle tools
tools = []
tool_names = agent_config.get('tools', [])
if tool_names:
# Try to import tools
for tool_name in tool_names:
try:
# Try common tool imports
if tool_name == 'search_tool' or tool_name == 'InternetSearchTool':
try:
from tools import search_tool
tools.append(search_tool)
except ImportError:
logger.warning(f"Could not import {tool_name}, skipping")
# Add more tool imports as needed
except Exception as e:
logger.warning(f"Error loading tool {tool_name}: {e}")
# Create agent
agent = Agent(
name=name,
role=role,
goal=goal,
instructions=instructions,
tools=tools,
verbose=verbose
)
return agent
def validate_schedule_config(schedule_config: Dict[str, Any]) -> None:
"""
Validate schedule configuration.
Args:
schedule_config: Schedule configuration dictionary
Raises:
ValueError: If configuration is invalid
"""
required_fields = ['interval']
for field in required_fields:
if field not in schedule_config:
raise ValueError(f"Missing required schedule field: {field}")
# Validate interval format
interval = schedule_config['interval']
valid_intervals = ['hourly', 'daily']
if interval not in valid_intervals:
# Check if it's a custom format (*/Nm, */Nh, */Ns, or plain number)
if not (interval.startswith('*/') or interval.isdigit()):
raise ValueError(
f"Invalid interval format: {interval}. "
f"Use 'hourly', 'daily', '*/30m', '*/6h', or seconds as number"
)
# Validate max_retries
max_retries = schedule_config.get('max_retries', 3)
if not isinstance(max_retries, int) or max_retries < 0:
raise ValueError(f"max_retries must be a non-negative integer, got: {max_retries}")
+1 -1
View File
@@ -1 +1 @@
__version__ = "2.3.64"
__version__ = "2.3.65"
+95
View File
@@ -0,0 +1,95 @@
#!/usr/bin/env python3
"""
Quick test script for scheduler with agents.yaml integration.
"""
import os
import sys
# Add parent directory to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src/praisonai'))
def test_yaml_loader():
"""Test YAML loader functionality."""
print("Testing YAML loader...")
from praisonai.scheduler.yaml_loader import load_agent_yaml_with_schedule
# Test with example YAML
yaml_path = "examples/cookbooks/yaml/news_monitor_scheduled.yaml"
try:
agent_config, schedule_config = load_agent_yaml_with_schedule(yaml_path)
print(f"✅ Agent config loaded: {agent_config.get('name', 'Unknown')}")
print(f"✅ Schedule interval: {schedule_config.get('interval', 'Not set')}")
print(f"✅ Max retries: {schedule_config.get('max_retries', 'Not set')}")
return True
except Exception as e:
print(f"❌ Error: {e}")
return False
def test_scheduler_from_yaml():
"""Test AgentScheduler.from_yaml() method."""
print("\nTesting AgentScheduler.from_yaml()...")
from praisonai.scheduler import AgentScheduler
yaml_path = "examples/cookbooks/yaml/news_monitor_scheduled.yaml"
try:
scheduler = AgentScheduler.from_yaml(yaml_path)
print(f"✅ Scheduler created successfully")
print(f"✅ Agent: {getattr(scheduler.agent, 'name', 'Unknown')}")
print(f"✅ Task: {scheduler.task[:50]}...")
# Test that we can access schedule config
if hasattr(scheduler, '_yaml_schedule_config'):
print(f"✅ Schedule config attached")
return True
except Exception as e:
print(f"❌ Error: {e}")
import traceback
traceback.print_exc()
return False
def main():
"""Run all tests."""
print("="*60)
print("Scheduler Integration Tests")
print("="*60)
results = []
# Test 1: YAML Loader
results.append(("YAML Loader", test_yaml_loader()))
# Test 2: Scheduler from YAML
results.append(("Scheduler from YAML", test_scheduler_from_yaml()))
# Summary
print("\n" + "="*60)
print("Test Summary:")
print("="*60)
for name, passed in results:
status = "✅ PASS" if passed else "❌ FAIL"
print(f"{status} - {name}")
all_passed = all(passed for _, passed in results)
if all_passed:
print("\n🎉 All tests passed!")
return 0
else:
print("\n⚠️ Some tests failed")
return 1
if __name__ == "__main__":
sys.exit(main())