Kye Gomez
3 months ago
parent
e43e4d8f73
commit
30b94c0d6b
@ -1,496 +1,415 @@
|
|||||||
# CronJob
|
# CronJob
|
||||||
|
|
||||||
A wrapper class that turns any callable (including Swarms agents) into a scheduled cron job. This class provides functionality to schedule and run tasks at specified intervals using the schedule library with cron-style scheduling.
|
A wrapper class that turns any callable (including Swarms agents) into a scheduled cron job using the schedule library with cron-style scheduling.
|
||||||
|
|
||||||
## Overview
|
Full Path `from swarms.structs.cron_job`
|
||||||
|
|
||||||
The CronJob class allows you to:
|
## Class Definition
|
||||||
|
|
||||||
- Schedule any callable or Swarms Agent to run at specified intervals
|
```python
|
||||||
|
class CronJob:
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
agent: Optional[Union[Any, Callable]] = None,
|
||||||
|
interval: Optional[str] = None,
|
||||||
|
job_id: Optional[str] = None,
|
||||||
|
callback: Optional[Callable[[Any, str, dict], Any]] = None,
|
||||||
|
) -> None
|
||||||
|
```
|
||||||
|
|
||||||
- Support for seconds, minutes, and hours intervals
|
## Constructor Parameters
|
||||||
|
|
||||||
- Run tasks in a separate thread
|
| Parameter | Type | Default | Description |
|
||||||
|
|-----------|------|---------|-------------|
|
||||||
|
| `agent` | `Optional[Union[Any, Callable]]` | `None` | The Swarms Agent instance or callable to be scheduled |
|
||||||
|
| `interval` | `Optional[str]` | `None` | Interval string in format "Xunit" (e.g., "5seconds", "10minutes", "1hour") |
|
||||||
|
| `job_id` | `Optional[str]` | `None` | Unique identifier for the job. Auto-generated if not provided |
|
||||||
|
| `callback` | `Optional[Callable[[Any, str, dict], Any]]` | `None` | Function to customize output processing |
|
||||||
|
|
||||||
- Handle graceful start/stop of scheduled jobs
|
## Instance Attributes
|
||||||
|
|
||||||
- Manage multiple concurrent scheduled jobs
|
| Attribute | Type | Description |
|
||||||
|
|-----------|------|-------------|
|
||||||
|
| `agent` | `Union[Any, Callable]` | The scheduled agent or callable |
|
||||||
|
| `interval` | `str` | The scheduling interval string |
|
||||||
|
| `job_id` | `str` | Unique job identifier |
|
||||||
|
| `is_running` | `bool` | Current execution status |
|
||||||
|
| `thread` | `Optional[threading.Thread]` | Background execution thread |
|
||||||
|
| `schedule` | `schedule.Scheduler` | Internal scheduler instance |
|
||||||
|
| `callback` | `Optional[Callable[[Any, str, dict], Any]]` | Output customization function |
|
||||||
|
| `execution_count` | `int` | Number of executions completed |
|
||||||
|
| `start_time` | `Optional[float]` | Job start timestamp |
|
||||||
|
|
||||||
## Architecture
|
## Methods
|
||||||
|
|
||||||
```mermaid
|
### `run(task: str, **kwargs) -> Any`
|
||||||
graph TD
|
|
||||||
A[CronJob] --> B[Initialize]
|
|
||||||
B --> C[Parse Interval]
|
|
||||||
C --> D[Schedule Task]
|
|
||||||
D --> E[Run Job]
|
|
||||||
E --> F[Execute Task]
|
|
||||||
F --> G{Is Agent?}
|
|
||||||
G -->|Yes| H[Run Agent]
|
|
||||||
G -->|No| I[Run Callable]
|
|
||||||
H --> J[Handle Result]
|
|
||||||
I --> J
|
|
||||||
J --> K[Sleep]
|
|
||||||
K --> E
|
|
||||||
```
|
|
||||||
|
|
||||||
## Class Reference
|
Schedules and starts the cron job execution.
|
||||||
|
|
||||||
### Constructor
|
**Parameters:**
|
||||||
|
- `task` (`str`): Task string to be executed by the agent
|
||||||
|
- `**kwargs` (`dict`): Additional parameters passed to agent's run method
|
||||||
|
|
||||||
```python
|
**Returns:**
|
||||||
def __init__(
|
- `Any`: Result of the cron job execution
|
||||||
agent: Optional[Union[Agent, Callable]] = None,
|
|
||||||
interval: Optional[str] = None,
|
**Raises:**
|
||||||
job_id: Optional[str] = None
|
- `CronJobConfigError`: If agent or interval is not configured
|
||||||
)
|
- `CronJobExecutionError`: If task execution fails
|
||||||
```
|
|
||||||
|
### `__call__(task: str, **kwargs) -> Any`
|
||||||
|
|
||||||
|
Callable interface for the CronJob instance.
|
||||||
|
|
||||||
|
**Parameters:**
|
||||||
|
- `task` (`str`): Task string to be executed
|
||||||
|
- `**kwargs` (`dict`): Additional parameters passed to agent's run method
|
||||||
|
|
||||||
|
**Returns:**
|
||||||
|
- `Any`: Result of the task execution
|
||||||
|
|
||||||
|
### `start() -> None`
|
||||||
|
|
||||||
|
Starts the scheduled job in a separate thread.
|
||||||
|
|
||||||
|
**Raises:**
|
||||||
|
- `CronJobExecutionError`: If the job fails to start
|
||||||
|
|
||||||
|
### `stop() -> None`
|
||||||
|
|
||||||
|
Stops the scheduled job gracefully.
|
||||||
|
|
||||||
|
**Raises:**
|
||||||
|
- `CronJobExecutionError`: If the job fails to stop properly
|
||||||
|
|
||||||
| Parameter | Type | Description | Required |
|
### `set_callback(callback: Callable[[Any, str, dict], Any]) -> None`
|
||||||
|-----------|------|-------------|-----------|
|
|
||||||
| agent | Agent or Callable | The Swarms Agent instance or callable to be scheduled | No |
|
|
||||||
| interval | str | The interval string (e.g., "5seconds", "10minutes", "1hour") | No |
|
|
||||||
| job_id | str | Unique identifier for the job. If not provided, one will be generated | No |
|
|
||||||
|
|
||||||
### Methods
|
Sets or updates the callback function for output customization.
|
||||||
|
|
||||||
#### run
|
**Parameters:**
|
||||||
|
- `callback` (`Callable[[Any, str, dict], Any]`): Function to customize output processing
|
||||||
|
|
||||||
|
### `get_execution_stats() -> Dict[str, Any]`
|
||||||
|
|
||||||
|
Retrieves execution statistics for the cron job.
|
||||||
|
|
||||||
|
**Returns:**
|
||||||
|
- `Dict[str, Any]`: Dictionary containing:
|
||||||
|
- `job_id` (`str`): Unique job identifier
|
||||||
|
- `is_running` (`bool`): Current execution status
|
||||||
|
- `execution_count` (`int`): Number of executions completed
|
||||||
|
- `start_time` (`Optional[float]`): Job start timestamp
|
||||||
|
- `uptime` (`float`): Seconds since job started
|
||||||
|
- `interval` (`str`): Scheduled execution interval
|
||||||
|
|
||||||
|
## Callback Function Signature
|
||||||
|
|
||||||
```python
|
```python
|
||||||
def run(task: str, **kwargs)
|
def callback_function(
|
||||||
|
output: Any, # Original output from the agent
|
||||||
|
task: str, # Task that was executed
|
||||||
|
metadata: dict # Job execution metadata
|
||||||
|
) -> Any: # Customized output (any type)
|
||||||
|
pass
|
||||||
```
|
```
|
||||||
|
|
||||||
| Parameter | Type | Description | Required |
|
### Callback Metadata Dictionary
|
||||||
|-----------|------|-------------|-----------|
|
|
||||||
| task | str | The task string to be executed by the agent | Yes |
|
| Key | Type | Description |
|
||||||
| **kwargs | dict | Additional parameters to pass to the agent's run method | No |
|
|-----|------|-------------|
|
||||||
|
| `job_id` | `str` | Unique job identifier |
|
||||||
|
| `timestamp` | `float` | Execution timestamp (Unix time) |
|
||||||
|
| `execution_count` | `int` | Sequential execution number |
|
||||||
|
| `task` | `str` | The task string that was executed |
|
||||||
|
| `kwargs` | `dict` | Additional parameters passed to agent |
|
||||||
|
| `start_time` | `Optional[float]` | Job start timestamp |
|
||||||
|
| `is_running` | `bool` | Current job status |
|
||||||
|
|
||||||
|
## Interval Format
|
||||||
|
|
||||||
|
The `interval` parameter accepts strings in the format `"Xunit"`:
|
||||||
|
|
||||||
|
| Unit | Examples | Description |
|
||||||
|
|------|----------|-------------|
|
||||||
|
| `seconds` | `"5seconds"`, `"30seconds"` | Execute every X seconds |
|
||||||
|
| `minutes` | `"1minute"`, `"15minutes"` | Execute every X minutes |
|
||||||
|
| `hours` | `"1hour"`, `"6hours"` | Execute every X hours |
|
||||||
|
|
||||||
|
## Exceptions
|
||||||
|
|
||||||
#### __call__
|
### `CronJobError`
|
||||||
|
Base exception class for all CronJob errors.
|
||||||
|
|
||||||
|
### `CronJobConfigError`
|
||||||
|
Raised for configuration errors (invalid agent, interval format, etc.).
|
||||||
|
|
||||||
|
### `CronJobScheduleError`
|
||||||
|
Raised for scheduling-related errors.
|
||||||
|
|
||||||
|
### `CronJobExecutionError`
|
||||||
|
Raised for execution-related errors (start/stop failures, task execution failures).
|
||||||
|
|
||||||
|
## Type Definitions
|
||||||
|
|
||||||
```python
|
```python
|
||||||
def __call__(task: str, **kwargs)
|
from typing import Any, Callable, Dict, Optional, Union
|
||||||
```
|
|
||||||
|
# Agent type can be any callable or object with run method
|
||||||
|
AgentType = Union[Any, Callable]
|
||||||
|
|
||||||
| Parameter | Type | Description | Required |
|
# Callback function signature
|
||||||
|-----------|------|-------------|-----------|
|
CallbackType = Callable[[Any, str, Dict[str, Any]], Any]
|
||||||
| task | str | The task string to be executed | Yes |
|
|
||||||
| **kwargs | dict | Additional parameters to pass to the agent's run method | No |
|
# Execution statistics return type
|
||||||
|
StatsType = Dict[str, Any]
|
||||||
|
```
|
||||||
|
|
||||||
## Examples
|
## Quick Start Examples
|
||||||
|
|
||||||
### Basic Usage with Swarms Agent
|
### Basic Usage
|
||||||
|
|
||||||
```python
|
```python
|
||||||
from swarms import Agent, CronJob
|
from swarms import Agent, CronJob
|
||||||
from loguru import logger
|
|
||||||
|
|
||||||
# Initialize the agent
|
# Simple agent cron job
|
||||||
agent = Agent(
|
agent = Agent(agent_name="MyAgent", ...)
|
||||||
agent_name="Quantitative-Trading-Agent",
|
cron_job = CronJob(agent=agent, interval="30seconds")
|
||||||
agent_description="Advanced quantitative trading and algorithmic analysis agent",
|
cron_job.run("Analyze market trends")
|
||||||
system_prompt="""You are an expert quantitative trading agent...""",
|
|
||||||
max_loops=1,
|
|
||||||
model_name="gpt-4.1",
|
|
||||||
dynamic_temperature_enabled=True,
|
|
||||||
output_type="str-all-except-first",
|
|
||||||
streaming_on=True,
|
|
||||||
print_on=True,
|
|
||||||
telemetry_enable=False,
|
|
||||||
)
|
|
||||||
|
|
||||||
# Create and run a cron job every 10 seconds
|
|
||||||
logger.info("Starting example cron job")
|
|
||||||
cron_job = CronJob(agent=agent, interval="10seconds")
|
|
||||||
cron_job.run(
|
|
||||||
task="What are the best top 3 etfs for gold coverage?"
|
|
||||||
)
|
|
||||||
```
|
```
|
||||||
|
|
||||||
### Using with a Custom Function
|
### With Custom Function
|
||||||
|
|
||||||
```python
|
```python
|
||||||
def custom_task(task: str):
|
def my_task(task: str) -> str:
|
||||||
print(f"Executing task: {task}")
|
return f"Completed: {task}"
|
||||||
return "Task completed"
|
|
||||||
|
|
||||||
# Create a cron job with a custom function
|
cron_job = CronJob(agent=my_task, interval="1minute")
|
||||||
cron_job = CronJob(
|
cron_job.run("Process data")
|
||||||
agent=custom_task,
|
|
||||||
interval="5minutes",
|
|
||||||
job_id="custom_task_job"
|
|
||||||
)
|
|
||||||
cron_job.run("Perform analysis")
|
|
||||||
```
|
```
|
||||||
|
|
||||||
|
### With Callback
|
||||||
|
|
||||||
### Cron Jobs With Multi-Agent Structures
|
```python
|
||||||
|
def callback(output, task, metadata):
|
||||||
|
return {"result": output, "count": metadata["execution_count"]}
|
||||||
|
|
||||||
You can also run Cron Jobs with multi-agent structures like `SequentialWorkflow`, `ConcurrentWorkflow`, `HiearchicalSwarm`, and other methods.
|
cron_job = CronJob(
|
||||||
|
agent=agent,
|
||||||
|
interval="30seconds",
|
||||||
|
callback=callback
|
||||||
|
)
|
||||||
|
```
|
||||||
|
|
||||||
- Just initialize the class as the agent parameter in the `CronJob(agent=swarm)`
|
|
||||||
|
|
||||||
- Input your arguments into the `.run(task: str)` method
|
## Full Examples
|
||||||
|
|
||||||
|
### Complete Agent with Callback
|
||||||
|
|
||||||
```python
|
```python
|
||||||
"""
|
from swarms import Agent, CronJob
|
||||||
Cryptocurrency Concurrent Multi-Agent Cron Job Example
|
from datetime import datetime
|
||||||
|
import json
|
||||||
|
|
||||||
This example demonstrates how to use ConcurrentWorkflow with CronJob to create
|
# Create agent
|
||||||
a powerful cryptocurrency tracking system. Each specialized agent analyzes a
|
agent = Agent(
|
||||||
specific cryptocurrency concurrently every minute.
|
agent_name="Financial-Analyst",
|
||||||
|
system_prompt="You are a financial analyst. Analyze market data and provide insights.",
|
||||||
|
model_name="gpt-4o-mini",
|
||||||
|
max_loops=1
|
||||||
|
)
|
||||||
|
|
||||||
Features:
|
# Advanced callback with monitoring
|
||||||
- ConcurrentWorkflow for parallel agent execution
|
class AdvancedCallback:
|
||||||
- CronJob scheduling for automated runs every 1 minute
|
def __init__(self):
|
||||||
- Each agent specializes in analyzing one specific cryptocurrency
|
self.history = []
|
||||||
- Real-time data fetching from CoinGecko API
|
self.error_count = 0
|
||||||
- Concurrent analysis of multiple cryptocurrencies
|
|
||||||
- Structured output with professional formatting
|
def __call__(self, output, task, metadata):
|
||||||
|
# Track execution
|
||||||
|
execution_data = {
|
||||||
|
"output": output,
|
||||||
|
"execution_id": metadata["execution_count"],
|
||||||
|
"timestamp": datetime.fromtimestamp(metadata["timestamp"]).isoformat(),
|
||||||
|
"task": task,
|
||||||
|
"job_id": metadata["job_id"],
|
||||||
|
"success": bool(output and "error" not in str(output).lower())
|
||||||
|
}
|
||||||
|
|
||||||
|
if not execution_data["success"]:
|
||||||
|
self.error_count += 1
|
||||||
|
|
||||||
|
self.history.append(execution_data)
|
||||||
|
|
||||||
|
# Keep only last 100 executions
|
||||||
|
if len(self.history) > 100:
|
||||||
|
self.history.pop(0)
|
||||||
|
|
||||||
|
return execution_data
|
||||||
|
|
||||||
|
def get_stats(self):
|
||||||
|
return {
|
||||||
|
"total_executions": len(self.history),
|
||||||
|
"error_count": self.error_count,
|
||||||
|
"success_rate": (len(self.history) - self.error_count) / len(self.history) if self.history else 0
|
||||||
|
}
|
||||||
|
|
||||||
|
# Use advanced callback
|
||||||
|
callback = AdvancedCallback()
|
||||||
|
cron_job = CronJob(
|
||||||
|
agent=agent,
|
||||||
|
interval="2minutes",
|
||||||
|
job_id="financial_analysis_job",
|
||||||
|
callback=callback
|
||||||
|
)
|
||||||
|
|
||||||
Architecture:
|
# Run the cron job
|
||||||
CronJob -> ConcurrentWorkflow -> [Bitcoin Agent, Ethereum Agent, Solana Agent, etc.] -> Parallel Analysis
|
try:
|
||||||
"""
|
cron_job.run("Analyze current market conditions and provide investment recommendations")
|
||||||
|
except KeyboardInterrupt:
|
||||||
|
cron_job.stop()
|
||||||
|
print("Final stats:", json.dumps(callback.get_stats(), indent=2))
|
||||||
|
```
|
||||||
|
|
||||||
from typing import List
|
### Multi-Agent Workflow with CronJob
|
||||||
from loguru import logger
|
|
||||||
|
|
||||||
|
```python
|
||||||
from swarms import Agent, CronJob, ConcurrentWorkflow
|
from swarms import Agent, CronJob, ConcurrentWorkflow
|
||||||
from swarms_tools import coin_gecko_coin_api
|
import json
|
||||||
|
|
||||||
|
# Create specialized agents
|
||||||
def create_crypto_specific_agents() -> List[Agent]:
|
bitcoin_agent = Agent(
|
||||||
"""
|
agent_name="Bitcoin-Analyst",
|
||||||
Creates agents that each specialize in analyzing a specific cryptocurrency.
|
system_prompt="You are a Bitcoin specialist. Focus only on Bitcoin analysis.",
|
||||||
|
model_name="gpt-4o-mini",
|
||||||
Returns:
|
max_loops=1
|
||||||
List[Agent]: List of cryptocurrency-specific Agent instances
|
)
|
||||||
"""
|
|
||||||
|
|
||||||
# Bitcoin Specialist Agent
|
|
||||||
bitcoin_agent = Agent(
|
|
||||||
agent_name="Bitcoin-Analyst",
|
|
||||||
agent_description="Expert analyst specializing exclusively in Bitcoin (BTC) analysis and market dynamics",
|
|
||||||
system_prompt="""You are a Bitcoin specialist and expert analyst. Your expertise includes:
|
|
||||||
|
|
||||||
BITCOIN SPECIALIZATION:
|
|
||||||
- Bitcoin's unique position as digital gold
|
|
||||||
- Bitcoin halving cycles and their market impact
|
|
||||||
- Bitcoin mining economics and hash rate analysis
|
|
||||||
- Lightning Network and Layer 2 developments
|
|
||||||
- Bitcoin adoption by institutions and countries
|
|
||||||
- Bitcoin's correlation with traditional markets
|
|
||||||
- Bitcoin technical analysis and on-chain metrics
|
|
||||||
- Bitcoin's role as a store of value and hedge against inflation
|
|
||||||
|
|
||||||
ANALYSIS FOCUS:
|
|
||||||
- Analyze ONLY Bitcoin data from the provided dataset
|
|
||||||
- Focus on Bitcoin-specific metrics and trends
|
|
||||||
- Consider Bitcoin's unique market dynamics
|
|
||||||
- Evaluate Bitcoin's dominance and market leadership
|
|
||||||
- Assess institutional adoption trends
|
|
||||||
- Monitor on-chain activity and network health
|
|
||||||
|
|
||||||
DELIVERABLES:
|
|
||||||
- Bitcoin-specific analysis and insights
|
|
||||||
- Price action assessment and predictions
|
|
||||||
- Market dominance analysis
|
|
||||||
- Institutional adoption impact
|
|
||||||
- Technical and fundamental outlook
|
|
||||||
- Risk factors specific to Bitcoin
|
|
||||||
|
|
||||||
Extract Bitcoin data from the provided dataset and provide comprehensive Bitcoin-focused analysis.""",
|
|
||||||
model_name="groq/moonshotai/kimi-k2-instruct",
|
|
||||||
max_loops=1,
|
|
||||||
dynamic_temperature_enabled=True,
|
|
||||||
streaming_on=False,
|
|
||||||
tools=[coin_gecko_coin_api],
|
|
||||||
)
|
|
||||||
|
|
||||||
# Ethereum Specialist Agent
|
|
||||||
ethereum_agent = Agent(
|
|
||||||
agent_name="Ethereum-Analyst",
|
|
||||||
agent_description="Expert analyst specializing exclusively in Ethereum (ETH) analysis and ecosystem development",
|
|
||||||
system_prompt="""You are an Ethereum specialist and expert analyst. Your expertise includes:
|
|
||||||
|
|
||||||
ETHEREUM SPECIALIZATION:
|
|
||||||
- Ethereum's smart contract platform and DeFi ecosystem
|
|
||||||
- Ethereum 2.0 transition and proof-of-stake mechanics
|
|
||||||
- Gas fees, network usage, and scalability solutions
|
|
||||||
- Layer 2 solutions (Arbitrum, Optimism, Polygon)
|
|
||||||
- DeFi protocols and TVL (Total Value Locked) analysis
|
|
||||||
- NFT markets and Ethereum's role in digital assets
|
|
||||||
- Developer activity and ecosystem growth
|
|
||||||
- EIP proposals and network upgrades
|
|
||||||
|
|
||||||
ANALYSIS FOCUS:
|
|
||||||
- Analyze ONLY Ethereum data from the provided dataset
|
|
||||||
- Focus on Ethereum's platform utility and network effects
|
|
||||||
- Evaluate DeFi ecosystem health and growth
|
|
||||||
- Assess Layer 2 adoption and scalability solutions
|
|
||||||
- Monitor network usage and gas fee trends
|
|
||||||
- Consider Ethereum's competitive position vs other smart contract platforms
|
|
||||||
|
|
||||||
DELIVERABLES:
|
|
||||||
- Ethereum-specific analysis and insights
|
|
||||||
- Platform utility and adoption metrics
|
|
||||||
- DeFi ecosystem impact assessment
|
|
||||||
- Network health and scalability evaluation
|
|
||||||
- Competitive positioning analysis
|
|
||||||
- Technical and fundamental outlook for ETH
|
|
||||||
|
|
||||||
Extract Ethereum data from the provided dataset and provide comprehensive Ethereum-focused analysis.""",
|
|
||||||
model_name="groq/moonshotai/kimi-k2-instruct",
|
|
||||||
max_loops=1,
|
|
||||||
dynamic_temperature_enabled=True,
|
|
||||||
streaming_on=False,
|
|
||||||
tools=[coin_gecko_coin_api],
|
|
||||||
)
|
|
||||||
|
|
||||||
# Solana Specialist Agent
|
|
||||||
solana_agent = Agent(
|
|
||||||
agent_name="Solana-Analyst",
|
|
||||||
agent_description="Expert analyst specializing exclusively in Solana (SOL) analysis and ecosystem development",
|
|
||||||
system_prompt="""You are a Solana specialist and expert analyst. Your expertise includes:
|
|
||||||
|
|
||||||
SOLANA SPECIALIZATION:
|
|
||||||
- Solana's high-performance blockchain architecture
|
|
||||||
- Proof-of-History consensus mechanism
|
|
||||||
- Solana's DeFi ecosystem and DEX platforms (Serum, Raydium)
|
|
||||||
- NFT marketplaces and creator economy on Solana
|
|
||||||
- Network outages and reliability concerns
|
|
||||||
- Developer ecosystem and Rust programming adoption
|
|
||||||
- Validator economics and network decentralization
|
|
||||||
- Cross-chain bridges and interoperability
|
|
||||||
|
|
||||||
ANALYSIS FOCUS:
|
|
||||||
- Analyze ONLY Solana data from the provided dataset
|
|
||||||
- Focus on Solana's performance and scalability advantages
|
|
||||||
- Evaluate network stability and uptime improvements
|
|
||||||
- Assess ecosystem growth and developer adoption
|
|
||||||
- Monitor DeFi and NFT activity on Solana
|
|
||||||
- Consider Solana's competitive position vs Ethereum
|
|
||||||
|
|
||||||
DELIVERABLES:
|
|
||||||
- Solana-specific analysis and insights
|
|
||||||
- Network performance and reliability assessment
|
|
||||||
- Ecosystem growth and adoption metrics
|
|
||||||
- DeFi and NFT market analysis
|
|
||||||
- Competitive advantages and challenges
|
|
||||||
- Technical and fundamental outlook for SOL
|
|
||||||
|
|
||||||
Extract Solana data from the provided dataset and provide comprehensive Solana-focused analysis.""",
|
|
||||||
model_name="groq/moonshotai/kimi-k2-instruct",
|
|
||||||
max_loops=1,
|
|
||||||
dynamic_temperature_enabled=True,
|
|
||||||
streaming_on=False,
|
|
||||||
tools=[coin_gecko_coin_api],
|
|
||||||
)
|
|
||||||
|
|
||||||
# Cardano Specialist Agent
|
|
||||||
cardano_agent = Agent(
|
|
||||||
agent_name="Cardano-Analyst",
|
|
||||||
agent_description="Expert analyst specializing exclusively in Cardano (ADA) analysis and research-driven development",
|
|
||||||
system_prompt="""You are a Cardano specialist and expert analyst. Your expertise includes:
|
|
||||||
|
|
||||||
CARDANO SPECIALIZATION:
|
|
||||||
- Cardano's research-driven development approach
|
|
||||||
- Ouroboros proof-of-stake consensus protocol
|
|
||||||
- Smart contract capabilities via Plutus and Marlowe
|
|
||||||
- Cardano's three-layer architecture (settlement, computation, control)
|
|
||||||
- Academic partnerships and peer-reviewed research
|
|
||||||
- Cardano ecosystem projects and DApp development
|
|
||||||
- Native tokens and Cardano's UTXO model
|
|
||||||
- Sustainability and treasury funding mechanisms
|
|
||||||
|
|
||||||
ANALYSIS FOCUS:
|
|
||||||
- Analyze ONLY Cardano data from the provided dataset
|
|
||||||
- Focus on Cardano's methodical development approach
|
|
||||||
- Evaluate smart contract adoption and ecosystem growth
|
|
||||||
- Assess academic partnerships and research contributions
|
|
||||||
- Monitor native token ecosystem development
|
|
||||||
- Consider Cardano's long-term roadmap and milestones
|
|
||||||
|
|
||||||
DELIVERABLES:
|
|
||||||
- Cardano-specific analysis and insights
|
|
||||||
- Development progress and milestone achievements
|
|
||||||
- Smart contract ecosystem evaluation
|
|
||||||
- Academic research impact assessment
|
|
||||||
- Native token and DApp adoption metrics
|
|
||||||
- Technical and fundamental outlook for ADA
|
|
||||||
|
|
||||||
Extract Cardano data from the provided dataset and provide comprehensive Cardano-focused analysis.""",
|
|
||||||
model_name="groq/moonshotai/kimi-k2-instruct",
|
|
||||||
max_loops=1,
|
|
||||||
dynamic_temperature_enabled=True,
|
|
||||||
streaming_on=False,
|
|
||||||
tools=[coin_gecko_coin_api],
|
|
||||||
)
|
|
||||||
|
|
||||||
# Binance Coin Specialist Agent
|
|
||||||
bnb_agent = Agent(
|
|
||||||
agent_name="BNB-Analyst",
|
|
||||||
agent_description="Expert analyst specializing exclusively in BNB analysis and Binance ecosystem dynamics",
|
|
||||||
system_prompt="""You are a BNB specialist and expert analyst. Your expertise includes:
|
|
||||||
|
|
||||||
BNB SPECIALIZATION:
|
|
||||||
- BNB's utility within the Binance ecosystem
|
|
||||||
- Binance Smart Chain (BSC) development and adoption
|
|
||||||
- BNB token burns and deflationary mechanics
|
|
||||||
- Binance exchange volume and market leadership
|
|
||||||
- BSC DeFi ecosystem and yield farming
|
|
||||||
- Cross-chain bridges and multi-chain strategies
|
|
||||||
- Regulatory challenges facing Binance globally
|
|
||||||
- BNB's role in transaction fee discounts and platform benefits
|
|
||||||
|
|
||||||
ANALYSIS FOCUS:
|
|
||||||
- Analyze ONLY BNB data from the provided dataset
|
|
||||||
- Focus on BNB's utility value and exchange benefits
|
|
||||||
- Evaluate BSC ecosystem growth and competition with Ethereum
|
|
||||||
- Assess token burn impact on supply and price
|
|
||||||
- Monitor Binance platform developments and regulations
|
|
||||||
- Consider BNB's centralized vs decentralized aspects
|
|
||||||
|
|
||||||
DELIVERABLES:
|
|
||||||
- BNB-specific analysis and insights
|
|
||||||
- Utility value and ecosystem benefits assessment
|
|
||||||
- BSC adoption and DeFi growth evaluation
|
|
||||||
- Token economics and burn mechanism impact
|
|
||||||
- Regulatory risk and compliance analysis
|
|
||||||
- Technical and fundamental outlook for BNB
|
|
||||||
|
|
||||||
Extract BNB data from the provided dataset and provide comprehensive BNB-focused analysis.""",
|
|
||||||
model_name="groq/moonshotai/kimi-k2-instruct",
|
|
||||||
max_loops=1,
|
|
||||||
dynamic_temperature_enabled=True,
|
|
||||||
streaming_on=False,
|
|
||||||
tools=[coin_gecko_coin_api],
|
|
||||||
)
|
|
||||||
|
|
||||||
# XRP Specialist Agent
|
|
||||||
xrp_agent = Agent(
|
|
||||||
agent_name="XRP-Analyst",
|
|
||||||
agent_description="Expert analyst specializing exclusively in XRP analysis and cross-border payment solutions",
|
|
||||||
system_prompt="""You are an XRP specialist and expert analyst. Your expertise includes:
|
|
||||||
|
|
||||||
XRP SPECIALIZATION:
|
|
||||||
- XRP's role in cross-border payments and remittances
|
|
||||||
- RippleNet adoption by financial institutions
|
|
||||||
- Central Bank Digital Currency (CBDC) partnerships
|
|
||||||
- Regulatory landscape and SEC lawsuit implications
|
|
||||||
- XRP Ledger's consensus mechanism and energy efficiency
|
|
||||||
- On-Demand Liquidity (ODL) usage and growth
|
|
||||||
- Competition with SWIFT and traditional payment rails
|
|
||||||
- Ripple's partnerships with banks and payment providers
|
|
||||||
|
|
||||||
ANALYSIS FOCUS:
|
|
||||||
- Analyze ONLY XRP data from the provided dataset
|
|
||||||
- Focus on XRP's utility in payments and remittances
|
|
||||||
- Evaluate RippleNet adoption and institutional partnerships
|
|
||||||
- Assess regulatory developments and legal clarity
|
|
||||||
- Monitor ODL usage and transaction volumes
|
|
||||||
- Consider XRP's competitive position in payments
|
|
||||||
|
|
||||||
DELIVERABLES:
|
|
||||||
- XRP-specific analysis and insights
|
|
||||||
- Payment utility and adoption assessment
|
|
||||||
- Regulatory landscape and legal developments
|
|
||||||
- Institutional partnership impact evaluation
|
|
||||||
- Cross-border payment market analysis
|
|
||||||
- Technical and fundamental outlook for XRP
|
|
||||||
|
|
||||||
Extract XRP data from the provided dataset and provide comprehensive XRP-focused analysis.""",
|
|
||||||
model_name="groq/moonshotai/kimi-k2-instruct",
|
|
||||||
max_loops=1,
|
|
||||||
dynamic_temperature_enabled=True,
|
|
||||||
streaming_on=False,
|
|
||||||
tools=[coin_gecko_coin_api],
|
|
||||||
)
|
|
||||||
|
|
||||||
return [
|
|
||||||
bitcoin_agent,
|
|
||||||
ethereum_agent,
|
|
||||||
solana_agent,
|
|
||||||
cardano_agent,
|
|
||||||
bnb_agent,
|
|
||||||
xrp_agent,
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
def create_crypto_workflow() -> ConcurrentWorkflow:
|
|
||||||
"""
|
|
||||||
Creates a ConcurrentWorkflow with cryptocurrency-specific analysis agents.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
ConcurrentWorkflow: Configured workflow for crypto analysis
|
|
||||||
"""
|
|
||||||
agents = create_crypto_specific_agents()
|
|
||||||
|
|
||||||
workflow = ConcurrentWorkflow(
|
|
||||||
name="Crypto-Specific-Analysis-Workflow",
|
|
||||||
description="Concurrent execution of cryptocurrency-specific analysis agents",
|
|
||||||
agents=agents,
|
|
||||||
max_loops=1,
|
|
||||||
)
|
|
||||||
|
|
||||||
return workflow
|
|
||||||
|
|
||||||
|
|
||||||
def create_crypto_cron_job() -> CronJob:
|
|
||||||
"""
|
|
||||||
Creates a CronJob that runs cryptocurrency-specific analysis every minute using ConcurrentWorkflow.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
CronJob: Configured cron job for automated crypto analysis
|
|
||||||
"""
|
|
||||||
# Create the concurrent workflow
|
|
||||||
workflow = create_crypto_workflow()
|
|
||||||
|
|
||||||
# Create the cron job
|
|
||||||
cron_job = CronJob(
|
|
||||||
agent=workflow, # Use the workflow as the agent
|
|
||||||
interval="5seconds", # Run every 1 minute
|
|
||||||
)
|
|
||||||
|
|
||||||
return cron_job
|
|
||||||
|
|
||||||
|
|
||||||
def main():
|
|
||||||
"""
|
|
||||||
Main function to run the cryptocurrency-specific concurrent analysis cron job.
|
|
||||||
"""
|
|
||||||
cron_job = create_crypto_cron_job()
|
|
||||||
|
|
||||||
prompt = """
|
|
||||||
|
|
||||||
Conduct a comprehensive analysis of your assigned cryptocurrency.
|
|
||||||
|
|
||||||
"""
|
|
||||||
|
|
||||||
# Start the cron job
|
ethereum_agent = Agent(
|
||||||
logger.info("🔄 Starting automated analysis loop...")
|
agent_name="Ethereum-Analyst",
|
||||||
logger.info("⏰ Press Ctrl+C to stop the cron job")
|
system_prompt="You are an Ethereum specialist. Focus only on Ethereum analysis.",
|
||||||
|
model_name="gpt-4o-mini",
|
||||||
|
max_loops=1
|
||||||
|
)
|
||||||
|
|
||||||
output = cron_job.run(task=prompt)
|
# Create concurrent workflow
|
||||||
print(output)
|
workflow = ConcurrentWorkflow(
|
||||||
|
name="Crypto-Analysis-Workflow",
|
||||||
|
agents=[bitcoin_agent, ethereum_agent],
|
||||||
|
max_loops=1
|
||||||
|
)
|
||||||
|
|
||||||
|
# Workflow callback
|
||||||
|
def workflow_callback(output, task, metadata):
|
||||||
|
"""Process multi-agent workflow output."""
|
||||||
|
return {
|
||||||
|
"workflow_results": output,
|
||||||
|
"execution_id": metadata["execution_count"],
|
||||||
|
"timestamp": metadata["timestamp"],
|
||||||
|
"agents_count": len(workflow.agents),
|
||||||
|
"task": task,
|
||||||
|
"metadata": {
|
||||||
|
"job_id": metadata["job_id"],
|
||||||
|
"uptime": metadata.get("uptime", 0)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
# Create workflow cron job
|
||||||
|
workflow_cron = CronJob(
|
||||||
|
agent=workflow,
|
||||||
|
interval="5minutes",
|
||||||
|
job_id="crypto_workflow_job",
|
||||||
|
callback=workflow_callback
|
||||||
|
)
|
||||||
|
|
||||||
if __name__ == "__main__":
|
# Run workflow cron job
|
||||||
main()
|
workflow_cron.run("Analyze your assigned cryptocurrency and provide market insights")
|
||||||
```
|
```
|
||||||
|
|
||||||
## Conclusion
|
### API Integration Example
|
||||||
|
|
||||||
The CronJob class provides a powerful way to schedule and automate tasks using Swarms Agents or custom functions. Key benefits include:
|
```python
|
||||||
|
import requests
|
||||||
|
from swarms import Agent, CronJob
|
||||||
|
import json
|
||||||
|
|
||||||
- Easy integration with Swarms Agents
|
# Create agent
|
||||||
|
agent = Agent(
|
||||||
|
agent_name="News-Analyst",
|
||||||
|
system_prompt="Analyze news and provide summaries.",
|
||||||
|
model_name="gpt-4o-mini",
|
||||||
|
max_loops=1
|
||||||
|
)
|
||||||
|
|
||||||
- Flexible interval scheduling
|
# API webhook callback
|
||||||
|
def api_callback(output, task, metadata):
|
||||||
|
"""Send results to external API."""
|
||||||
|
payload = {
|
||||||
|
"data": output,
|
||||||
|
"source": "swarms_cronjob",
|
||||||
|
"job_id": metadata["job_id"],
|
||||||
|
"execution_id": metadata["execution_count"],
|
||||||
|
"timestamp": metadata["timestamp"],
|
||||||
|
"task": task
|
||||||
|
}
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Send to webhook (replace with your URL)
|
||||||
|
response = requests.post(
|
||||||
|
"https://api.example.com/webhook",
|
||||||
|
json=payload,
|
||||||
|
timeout=30
|
||||||
|
)
|
||||||
|
|
||||||
|
return {
|
||||||
|
"original_output": output,
|
||||||
|
"api_status": "sent",
|
||||||
|
"api_response_code": response.status_code,
|
||||||
|
"execution_id": metadata["execution_count"]
|
||||||
|
}
|
||||||
|
except requests.RequestException as e:
|
||||||
|
return {
|
||||||
|
"original_output": output,
|
||||||
|
"api_status": "failed",
|
||||||
|
"error": str(e),
|
||||||
|
"execution_id": metadata["execution_count"]
|
||||||
|
}
|
||||||
|
|
||||||
|
# Database logging callback
|
||||||
|
def db_callback(output, task, metadata):
|
||||||
|
"""Log to database (pseudo-code)."""
|
||||||
|
# db.execute(
|
||||||
|
# "INSERT INTO cron_results (job_id, output, timestamp) VALUES (?, ?, ?)",
|
||||||
|
# (metadata["job_id"], output, metadata["timestamp"])
|
||||||
|
# )
|
||||||
|
|
||||||
|
return {
|
||||||
|
"output": output,
|
||||||
|
"logged_to_db": True,
|
||||||
|
"execution_id": metadata["execution_count"]
|
||||||
|
}
|
||||||
|
|
||||||
|
# Create cron job with API integration
|
||||||
|
api_cron_job = CronJob(
|
||||||
|
agent=agent,
|
||||||
|
interval="10minutes",
|
||||||
|
job_id="news_analysis_api_job",
|
||||||
|
callback=api_callback
|
||||||
|
)
|
||||||
|
|
||||||
- Thread-safe execution
|
# Dynamic callback switching example
|
||||||
|
db_cron_job = CronJob(
|
||||||
|
agent=agent,
|
||||||
|
interval="1hour",
|
||||||
|
job_id="news_analysis_db_job"
|
||||||
|
)
|
||||||
|
|
||||||
- Graceful error handling
|
# Start with API callback
|
||||||
|
db_cron_job.set_callback(api_callback)
|
||||||
|
|
||||||
- Simple API for task scheduling
|
# Later switch to database callback
|
||||||
|
# db_cron_job.set_callback(db_callback)
|
||||||
|
|
||||||
- Support for both agent and callable-based tasks
|
# Get execution statistics
|
||||||
|
stats = db_cron_job.get_execution_stats()
|
||||||
|
print(f"Job statistics: {json.dumps(stats, indent=2)}")
|
||||||
|
```
|
||||||
@ -0,0 +1,12 @@
|
|||||||
|
from swarms.agents.agent_judge import AgentJudge
|
||||||
|
|
||||||
|
# Initialize the agent judge
|
||||||
|
judge = AgentJudge(
|
||||||
|
agent_name="quality-judge", model_name="gpt-4", max_loops=2
|
||||||
|
)
|
||||||
|
|
||||||
|
# Example agent output to evaluate
|
||||||
|
agent_output = "The capital of France is Paris. The city is known for its famous Eiffel Tower and delicious croissants. The population is approximately 2.1 million people."
|
||||||
|
|
||||||
|
# Run evaluation with context building
|
||||||
|
evaluations = judge.run(task=agent_output)
|
||||||
@ -0,0 +1,21 @@
|
|||||||
|
from swarms.agents.agent_judge import AgentJudge
|
||||||
|
|
||||||
|
# Initialize the agent judge with custom evaluation criteria
|
||||||
|
judge = AgentJudge(
|
||||||
|
agent_name="technical-judge",
|
||||||
|
model_name="gpt-4",
|
||||||
|
max_loops=1,
|
||||||
|
evaluation_criteria={
|
||||||
|
"accuracy": 0.4,
|
||||||
|
"completeness": 0.3,
|
||||||
|
"clarity": 0.2,
|
||||||
|
"logic": 0.1,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
# Example technical agent output to evaluate
|
||||||
|
technical_output = "To solve the quadratic equation x² + 5x + 6 = 0, we can use the quadratic formula: x = (-b ± √(b² - 4ac)) / 2a. Here, a=1, b=5, c=6. Substituting: x = (-5 ± √(25 - 24)) / 2 = (-5 ± √1) / 2 = (-5 ± 1) / 2. So x = -2 or x = -3."
|
||||||
|
|
||||||
|
# Run evaluation with context building
|
||||||
|
evaluations = judge.run(task=technical_output)
|
||||||
|
print(evaluations)
|
||||||
@ -0,0 +1,20 @@
|
|||||||
|
from swarms.agents.agent_judge import AgentJudge
|
||||||
|
|
||||||
|
# Initialize the agent judge for creative content evaluation
|
||||||
|
judge = AgentJudge(
|
||||||
|
agent_name="creative-judge",
|
||||||
|
model_name="gpt-4",
|
||||||
|
max_loops=2,
|
||||||
|
evaluation_criteria={
|
||||||
|
"creativity": 0.4,
|
||||||
|
"originality": 0.3,
|
||||||
|
"engagement": 0.2,
|
||||||
|
"coherence": 0.1,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
# Example creative agent output to evaluate
|
||||||
|
creative_output = "The moon hung like a silver coin in the velvet sky, casting shadows that danced with the wind. Ancient trees whispered secrets to the stars, while time itself seemed to pause in reverence of this magical moment. The world held its breath, waiting for the next chapter of the eternal story."
|
||||||
|
|
||||||
|
# Run evaluation with context building
|
||||||
|
evaluations = judge.run(task=creative_output)
|
||||||
@ -0,0 +1,313 @@
|
|||||||
|
"""
|
||||||
|
Callback CronJob Example
|
||||||
|
|
||||||
|
This example demonstrates how to use the new callback functionality in CronJob
|
||||||
|
to customize output processing while the cron job is still running.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import json
|
||||||
|
import time
|
||||||
|
from datetime import datetime
|
||||||
|
from typing import Any, Dict
|
||||||
|
from loguru import logger
|
||||||
|
|
||||||
|
from swarms import Agent, CronJob
|
||||||
|
|
||||||
|
|
||||||
|
def create_sample_agent():
|
||||||
|
"""Create a sample agent for demonstration."""
|
||||||
|
return Agent(
|
||||||
|
agent_name="Sample-Analysis-Agent",
|
||||||
|
system_prompt="""You are a data analysis agent. Analyze the given data and provide insights.
|
||||||
|
Keep your responses concise and focused on key findings.""",
|
||||||
|
model_name="gpt-4o-mini",
|
||||||
|
max_loops=1,
|
||||||
|
print_on=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# Example 1: Simple output transformation callback
|
||||||
|
def transform_output_callback(output: Any, task: str, metadata: Dict) -> Dict:
|
||||||
|
"""Transform the agent output into a structured format.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
output: The original output from the agent
|
||||||
|
task: The task that was executed
|
||||||
|
metadata: Job metadata including execution count, timestamp, etc.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Dict: Transformed output with additional metadata
|
||||||
|
"""
|
||||||
|
return {
|
||||||
|
"original_output": output,
|
||||||
|
"transformed_at": datetime.fromtimestamp(metadata["timestamp"]).isoformat(),
|
||||||
|
"execution_number": metadata["execution_count"],
|
||||||
|
"task_executed": task,
|
||||||
|
"job_status": "running" if metadata["is_running"] else "stopped",
|
||||||
|
"uptime_seconds": metadata["uptime"] if metadata["start_time"] else 0
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
# Example 2: Output filtering and enhancement callback
|
||||||
|
def filter_and_enhance_callback(output: Any, task: str, metadata: Dict) -> Dict:
|
||||||
|
"""Filter and enhance the output based on execution count and content.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
output: The original output from the agent
|
||||||
|
task: The task that was executed
|
||||||
|
metadata: Job metadata
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Dict: Filtered and enhanced output
|
||||||
|
"""
|
||||||
|
# Only include outputs that contain certain keywords
|
||||||
|
if isinstance(output, str):
|
||||||
|
if any(keyword in output.lower() for keyword in ["important", "key", "significant", "trend"]):
|
||||||
|
enhanced_output = {
|
||||||
|
"content": output,
|
||||||
|
"priority": "high",
|
||||||
|
"execution_id": metadata["execution_count"],
|
||||||
|
"timestamp": metadata["timestamp"],
|
||||||
|
"analysis_type": "priority_content"
|
||||||
|
}
|
||||||
|
else:
|
||||||
|
enhanced_output = {
|
||||||
|
"content": output,
|
||||||
|
"priority": "normal",
|
||||||
|
"execution_id": metadata["execution_count"],
|
||||||
|
"timestamp": metadata["timestamp"],
|
||||||
|
"analysis_type": "standard_content"
|
||||||
|
}
|
||||||
|
else:
|
||||||
|
enhanced_output = {
|
||||||
|
"content": str(output),
|
||||||
|
"priority": "unknown",
|
||||||
|
"execution_id": metadata["execution_count"],
|
||||||
|
"timestamp": metadata["timestamp"],
|
||||||
|
"analysis_type": "non_string_content"
|
||||||
|
}
|
||||||
|
|
||||||
|
return enhanced_output
|
||||||
|
|
||||||
|
|
||||||
|
# Example 3: Real-time monitoring callback
|
||||||
|
class MonitoringCallback:
|
||||||
|
"""Callback class that provides real-time monitoring capabilities."""
|
||||||
|
|
||||||
|
def __init__(self):
|
||||||
|
self.output_history = []
|
||||||
|
self.error_count = 0
|
||||||
|
self.success_count = 0
|
||||||
|
self.last_execution_time = None
|
||||||
|
|
||||||
|
def __call__(self, output: Any, task: str, metadata: Dict) -> Dict:
|
||||||
|
"""Monitor and track execution metrics.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
output: The original output from the agent
|
||||||
|
task: The task that was executed
|
||||||
|
metadata: Job metadata
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Dict: Output with monitoring information
|
||||||
|
"""
|
||||||
|
current_time = time.time()
|
||||||
|
|
||||||
|
# Calculate execution time
|
||||||
|
if self.last_execution_time:
|
||||||
|
execution_time = current_time - self.last_execution_time
|
||||||
|
else:
|
||||||
|
execution_time = 0
|
||||||
|
|
||||||
|
self.last_execution_time = current_time
|
||||||
|
|
||||||
|
# Track success/error
|
||||||
|
if output and output != "Error":
|
||||||
|
self.success_count += 1
|
||||||
|
status = "success"
|
||||||
|
else:
|
||||||
|
self.error_count += 1
|
||||||
|
status = "error"
|
||||||
|
|
||||||
|
# Store in history (keep last 100)
|
||||||
|
monitoring_data = {
|
||||||
|
"output": output,
|
||||||
|
"status": status,
|
||||||
|
"execution_time": execution_time,
|
||||||
|
"execution_count": metadata["execution_count"],
|
||||||
|
"timestamp": metadata["timestamp"],
|
||||||
|
"task": task,
|
||||||
|
"metrics": {
|
||||||
|
"success_rate": self.success_count / (self.success_count + self.error_count),
|
||||||
|
"total_executions": self.success_count + self.error_count,
|
||||||
|
"error_count": self.error_count,
|
||||||
|
"success_count": self.success_count
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
self.output_history.append(monitoring_data)
|
||||||
|
if len(self.output_history) > 100:
|
||||||
|
self.output_history.pop(0)
|
||||||
|
|
||||||
|
return monitoring_data
|
||||||
|
|
||||||
|
def get_summary(self) -> Dict:
|
||||||
|
"""Get monitoring summary."""
|
||||||
|
return {
|
||||||
|
"total_executions": self.success_count + self.error_count,
|
||||||
|
"success_count": self.success_count,
|
||||||
|
"error_count": self.error_count,
|
||||||
|
"success_rate": self.success_count / (self.success_count + self.error_count) if (self.success_count + self.error_count) > 0 else 0,
|
||||||
|
"history_length": len(self.output_history),
|
||||||
|
"last_execution_time": self.last_execution_time
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
# Example 4: API integration callback
|
||||||
|
def api_webhook_callback(output: Any, task: str, metadata: Dict) -> Dict:
|
||||||
|
"""Callback that could send output to an external API.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
output: The original output from the agent
|
||||||
|
task: The task that was executed
|
||||||
|
metadata: Job metadata
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Dict: Output with API integration metadata
|
||||||
|
"""
|
||||||
|
# In a real implementation, you would send this to your API
|
||||||
|
api_payload = {
|
||||||
|
"data": output,
|
||||||
|
"source": "cron_job",
|
||||||
|
"job_id": metadata["job_id"],
|
||||||
|
"execution_id": metadata["execution_count"],
|
||||||
|
"timestamp": metadata["timestamp"],
|
||||||
|
"task": task
|
||||||
|
}
|
||||||
|
|
||||||
|
# Simulate API call (replace with actual HTTP request)
|
||||||
|
logger.info(f"Would send to API: {json.dumps(api_payload, indent=2)}")
|
||||||
|
|
||||||
|
return {
|
||||||
|
"output": output,
|
||||||
|
"api_status": "sent",
|
||||||
|
"api_payload": api_payload,
|
||||||
|
"execution_id": metadata["execution_count"]
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
"""Demonstrate different callback usage patterns."""
|
||||||
|
logger.info("🚀 Starting Callback CronJob Examples")
|
||||||
|
|
||||||
|
# Create the agent
|
||||||
|
agent = create_sample_agent()
|
||||||
|
|
||||||
|
# Example 1: Simple transformation callback
|
||||||
|
logger.info("📝 Example 1: Simple Output Transformation")
|
||||||
|
transform_cron = CronJob(
|
||||||
|
agent=agent,
|
||||||
|
interval="15seconds",
|
||||||
|
job_id="transform-example",
|
||||||
|
callback=transform_output_callback
|
||||||
|
)
|
||||||
|
|
||||||
|
# Example 2: Filtering and enhancement callback
|
||||||
|
logger.info("🔍 Example 2: Output Filtering and Enhancement")
|
||||||
|
filter_cron = CronJob(
|
||||||
|
agent=agent,
|
||||||
|
interval="20seconds",
|
||||||
|
job_id="filter-example",
|
||||||
|
callback=filter_and_enhance_callback
|
||||||
|
)
|
||||||
|
|
||||||
|
# Example 3: Monitoring callback
|
||||||
|
logger.info("📊 Example 3: Real-time Monitoring")
|
||||||
|
monitoring_callback = MonitoringCallback()
|
||||||
|
monitoring_cron = CronJob(
|
||||||
|
agent=agent,
|
||||||
|
interval="25seconds",
|
||||||
|
job_id="monitoring-example",
|
||||||
|
callback=monitoring_callback
|
||||||
|
)
|
||||||
|
|
||||||
|
# Example 4: API integration callback
|
||||||
|
logger.info("🌐 Example 4: API Integration")
|
||||||
|
api_cron = CronJob(
|
||||||
|
agent=agent,
|
||||||
|
interval="30seconds",
|
||||||
|
job_id="api-example",
|
||||||
|
callback=api_webhook_callback
|
||||||
|
)
|
||||||
|
|
||||||
|
# Start all cron jobs
|
||||||
|
logger.info("▶️ Starting all cron jobs...")
|
||||||
|
|
||||||
|
# Start them in separate threads to run concurrently
|
||||||
|
import threading
|
||||||
|
|
||||||
|
def run_cron(cron_job, task):
|
||||||
|
try:
|
||||||
|
cron_job.run(task=task)
|
||||||
|
except KeyboardInterrupt:
|
||||||
|
cron_job.stop()
|
||||||
|
|
||||||
|
# Start each cron job in its own thread
|
||||||
|
threads = []
|
||||||
|
tasks = [
|
||||||
|
"Analyze the current market trends and provide key insights",
|
||||||
|
"What are the most important factors affecting today's economy?",
|
||||||
|
"Provide a summary of recent technological developments",
|
||||||
|
"Analyze the impact of current events on business operations"
|
||||||
|
]
|
||||||
|
|
||||||
|
for i, (cron_job, task) in enumerate([
|
||||||
|
(transform_cron, tasks[0]),
|
||||||
|
(filter_cron, tasks[1]),
|
||||||
|
(monitoring_cron, tasks[2]),
|
||||||
|
(api_cron, tasks[3])
|
||||||
|
]):
|
||||||
|
thread = threading.Thread(
|
||||||
|
target=run_cron,
|
||||||
|
args=(cron_job, task),
|
||||||
|
daemon=True,
|
||||||
|
name=f"cron-thread-{i}"
|
||||||
|
)
|
||||||
|
thread.start()
|
||||||
|
threads.append(thread)
|
||||||
|
|
||||||
|
logger.info("✅ All cron jobs started successfully!")
|
||||||
|
logger.info("📊 Press Ctrl+C to stop and see monitoring summary")
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Let them run for a while
|
||||||
|
time.sleep(120) # Run for 2 minutes
|
||||||
|
|
||||||
|
# Show monitoring summary
|
||||||
|
logger.info("📈 Monitoring Summary:")
|
||||||
|
logger.info(json.dumps(monitoring_callback.get_summary(), indent=2))
|
||||||
|
|
||||||
|
# Show execution stats for each cron job
|
||||||
|
for cron_job, name in [
|
||||||
|
(transform_cron, "Transform"),
|
||||||
|
(filter_cron, "Filter"),
|
||||||
|
(monitoring_cron, "Monitoring"),
|
||||||
|
(api_cron, "API")
|
||||||
|
]:
|
||||||
|
stats = cron_job.get_execution_stats()
|
||||||
|
logger.info(f"{name} Cron Stats: {json.dumps(stats, indent=2)}")
|
||||||
|
|
||||||
|
except KeyboardInterrupt:
|
||||||
|
logger.info("⏹️ Stopping all cron jobs...")
|
||||||
|
|
||||||
|
# Stop all cron jobs
|
||||||
|
for cron_job in [transform_cron, filter_cron, monitoring_cron, api_cron]:
|
||||||
|
cron_job.stop()
|
||||||
|
|
||||||
|
# Show final monitoring summary
|
||||||
|
logger.info("📊 Final Monitoring Summary:")
|
||||||
|
logger.info(json.dumps(monitoring_callback.get_summary(), indent=2))
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
@ -0,0 +1,81 @@
|
|||||||
|
"""
|
||||||
|
Simple Callback CronJob Example
|
||||||
|
|
||||||
|
This example shows the basic usage of the new callback functionality
|
||||||
|
in CronJob to customize output while the job is running.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import json
|
||||||
|
import time
|
||||||
|
from datetime import datetime
|
||||||
|
from loguru import logger
|
||||||
|
|
||||||
|
from swarms import Agent, CronJob
|
||||||
|
|
||||||
|
|
||||||
|
def create_simple_agent():
|
||||||
|
"""Create a simple agent for demonstration."""
|
||||||
|
return Agent(
|
||||||
|
agent_name="Simple-Analysis-Agent",
|
||||||
|
system_prompt="You are a simple analysis agent. Provide brief insights on the given topic.",
|
||||||
|
model_name="gpt-4o-mini",
|
||||||
|
max_loops=1,
|
||||||
|
print_on=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def simple_callback(output, task, metadata):
|
||||||
|
"""Simple callback that adds metadata to the output.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
output: The original output from the agent
|
||||||
|
task: The task that was executed
|
||||||
|
metadata: Job metadata (execution count, timestamp, etc.)
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: Enhanced output with metadata
|
||||||
|
"""
|
||||||
|
return {
|
||||||
|
"agent_output": output,
|
||||||
|
"execution_number": metadata["execution_count"],
|
||||||
|
"timestamp": datetime.fromtimestamp(metadata["timestamp"]).isoformat(),
|
||||||
|
"task": task,
|
||||||
|
"job_id": metadata["job_id"]
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
"""Demonstrate basic callback usage."""
|
||||||
|
logger.info("🚀 Starting Simple Callback Example")
|
||||||
|
|
||||||
|
# Create agent and cron job with callback
|
||||||
|
agent = create_simple_agent()
|
||||||
|
|
||||||
|
cron_job = CronJob(
|
||||||
|
agent=agent,
|
||||||
|
interval="10seconds",
|
||||||
|
job_id="simple-callback-example",
|
||||||
|
callback=simple_callback
|
||||||
|
)
|
||||||
|
|
||||||
|
logger.info("▶️ Starting cron job with callback...")
|
||||||
|
logger.info("📝 The callback will enhance each output with metadata")
|
||||||
|
logger.info("⏹️ Press Ctrl+C to stop")
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Start the cron job
|
||||||
|
cron_job.run(
|
||||||
|
task="What are the key trends in artificial intelligence today?"
|
||||||
|
)
|
||||||
|
except KeyboardInterrupt:
|
||||||
|
logger.info("⏹️ Stopping cron job...")
|
||||||
|
cron_job.stop()
|
||||||
|
|
||||||
|
# Show execution statistics
|
||||||
|
stats = cron_job.get_execution_stats()
|
||||||
|
logger.info("📊 Final Statistics:")
|
||||||
|
logger.info(json.dumps(stats, indent=2))
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
Loading…
Reference in new issue