-
-
Notifications
You must be signed in to change notification settings - Fork 418
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
56b0afb
commit 315ac71
Showing
24 changed files
with
2,612 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,91 @@ | ||
from praisonaiagents.agent import Agent | ||
from praisonaiagents.task import Task | ||
from praisonaiagents.agents import PraisonAIAgents | ||
from typing import List, Dict | ||
import time | ||
|
||
def get_time_check(): | ||
current_time = int(time.time()) | ||
result = "even" if current_time % 2 == 0 else "odd" | ||
print(f"Time check: {current_time} is {result}") | ||
return result | ||
|
||
def create_prompt_chain(): | ||
# Create agents for each step in the chain | ||
agent1 = Agent( | ||
name="Time Checker", | ||
role="Time checker", | ||
goal="Check if the time is even or odd", | ||
instructions="Check if the time is even or odd", | ||
tools=[get_time_check] | ||
) | ||
|
||
agent2 = Agent( | ||
name="Advanced Analyzer", | ||
role="Advanced data analyzer", | ||
goal="Perform in-depth analysis of processed data", | ||
instructions="Analyze the processed data in detail" | ||
) | ||
|
||
agent3 = Agent( | ||
name="Final Processor", | ||
role="Final data processor", | ||
goal="Generate final output based on analysis", | ||
instructions="Create final output based on analyzed data" | ||
) | ||
|
||
# Create tasks for each step | ||
initial_task = Task( | ||
name="time_check", | ||
description="Getting time check and checking if it is even or odd", | ||
expected_output="Getting time check and checking if it is even or odd", | ||
agent=agent1, | ||
is_start=True, # Mark as the starting task | ||
task_type="decision", # This task will make a decision | ||
next_tasks=["advanced_analysis"], # Next task if condition passes | ||
condition={ | ||
"even": ["advanced_analysis"], # If passes, go to advanced analysis | ||
"odd": ["final_processing"] # If fails, exit the chain | ||
} | ||
) | ||
|
||
analysis_task = Task( | ||
name="advanced_analysis", | ||
description="Perform advanced analysis on the processed data", | ||
expected_output="Analyzed data ready for final processing", | ||
agent=agent2, | ||
next_tasks=["final_processing"] | ||
) | ||
|
||
final_task = Task( | ||
name="final_processing", | ||
description="Generate final output", | ||
expected_output="Final processed result", | ||
agent=agent3 | ||
) | ||
|
||
# Create the workflow manager | ||
workflow = PraisonAIAgents( | ||
agents=[agent1, agent2, agent3], | ||
tasks=[initial_task, analysis_task, final_task], | ||
process="workflow", # Use workflow process type | ||
verbose=True | ||
) | ||
|
||
return workflow | ||
|
||
def main(): | ||
# Create and run the prompt chain | ||
workflow = create_prompt_chain() | ||
|
||
# Run the workflow | ||
results = workflow.start() | ||
|
||
# Print results | ||
print("\nWorkflow Results:") | ||
for task_id, result in results["task_results"].items(): | ||
if result: | ||
print(f"Task {task_id}: {result.raw}") | ||
|
||
if __name__ == "__main__": | ||
main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,91 @@ | ||
from praisonaiagents.agent import Agent | ||
from praisonaiagents.task import Task | ||
from praisonaiagents.agents import PraisonAIAgents | ||
from typing import List, Dict | ||
import time | ||
|
||
def get_time_check(): | ||
current_time = int(time.time()) | ||
result = "even" if current_time % 2 == 0 else "odd" | ||
print(f"Time check: {current_time} is {result}") | ||
return result | ||
|
||
def create_prompt_chain(): | ||
# Create agents for each step in the chain | ||
agent1 = Agent( | ||
name="Time Checker", | ||
role="Time checker", | ||
goal="Check if the time is even or odd", | ||
instructions="Check if the time is even or odd", | ||
tools=[get_time_check] | ||
) | ||
|
||
agent2 = Agent( | ||
name="Advanced Analyzer", | ||
role="Advanced data analyzer", | ||
goal="Perform in-depth analysis of processed data", | ||
instructions="Analyze the processed data in detail" | ||
) | ||
|
||
agent3 = Agent( | ||
name="Final Processor", | ||
role="Final data processor", | ||
goal="Generate final output based on analysis", | ||
instructions="Create final output based on analyzed data" | ||
) | ||
|
||
# Create tasks for each step | ||
initial_task = Task( | ||
name="time_check", | ||
description="Getting time check and checking if it is even or odd", | ||
expected_output="Getting time check and checking if it is even or odd", | ||
agent=agent1, | ||
is_start=True, # Mark as the starting task | ||
task_type="decision", # This task will make a decision | ||
next_tasks=["advanced_analysis"], # Next task if condition passes | ||
condition={ | ||
"even": ["advanced_analysis"], # If passes, go to advanced analysis | ||
"odd": ["final_processing"] # If fails, exit the chain | ||
} | ||
) | ||
|
||
analysis_task = Task( | ||
name="advanced_analysis", | ||
description="Perform advanced analysis on the processed data", | ||
expected_output="Analyzed data ready for final processing", | ||
agent=agent2, | ||
next_tasks=["final_processing"] | ||
) | ||
|
||
final_task = Task( | ||
name="final_processing", | ||
description="Generate final output", | ||
expected_output="Final processed result", | ||
agent=agent3 | ||
) | ||
|
||
# Create the workflow manager | ||
workflow = PraisonAIAgents( | ||
agents=[agent1, agent2, agent3], | ||
tasks=[initial_task, analysis_task, final_task], | ||
process="workflow", # Use workflow process type | ||
verbose=True | ||
) | ||
|
||
return workflow | ||
|
||
def main(): | ||
# Create and run the prompt chain | ||
workflow = create_prompt_chain() | ||
|
||
# Run the workflow | ||
results = workflow.start() | ||
|
||
# Print results | ||
print("\nWorkflow Results:") | ||
for task_id, result in results["task_results"].items(): | ||
if result: | ||
print(f"Task {task_id}: {result.raw}") | ||
|
||
if __name__ == "__main__": | ||
main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,82 @@ | ||
from praisonaiagents.agent import Agent | ||
from praisonaiagents.task import Task | ||
from praisonaiagents.agents import PraisonAIAgents | ||
import time | ||
|
||
current_time = int(time.time()) | ||
result = "even" if current_time % 2 == 0 else "odd" | ||
print(f"Time check: {current_time} is {result}") | ||
|
||
# Create specialized agents | ||
router = Agent( | ||
name="Router", | ||
role="Input Router", | ||
goal="Evaluate input and determine routing path", | ||
instructions="Analyze input and decide whether to proceed or exit", | ||
tools=[get_time_check] | ||
) | ||
|
||
processor1 = Agent( | ||
name="Processor 1", | ||
role="Secondary Processor", | ||
goal="Process valid inputs that passed initial check", | ||
instructions="Process data that passed the routing check" | ||
) | ||
|
||
processor2 = Agent( | ||
name="Processor 2", | ||
role="Final Processor", | ||
goal="Perform final processing on validated data", | ||
instructions="Generate final output for processed data" | ||
) | ||
|
||
# Create tasks with routing logic | ||
routing_task = Task( | ||
name="initial_routing", | ||
description="check the time and return according to what is returned", | ||
expected_output="pass or fail based on what is returned", | ||
agent=router, | ||
is_start=True, | ||
task_type="decision", | ||
condition={ | ||
"pass": ["process_valid"], | ||
"fail": "exit" | ||
} | ||
) | ||
|
||
processing_task = Task( | ||
name="process_valid", | ||
description="Process validated input", | ||
expected_output="Processed data ready for final step", | ||
agent=processor1, | ||
next_tasks=["final_process"] | ||
) | ||
|
||
final_task = Task( | ||
name="final_process", | ||
description="Generate final output", | ||
expected_output="Final processed result", | ||
agent=processor2 | ||
) | ||
|
||
# Create and run workflow | ||
workflow = PraisonAIAgents( | ||
agents=[router, processor1, processor2], | ||
tasks=[routing_task, processing_task, final_task], | ||
process="workflow", | ||
verbose=True | ||
) | ||
|
||
print("\nStarting Routing Workflow...") | ||
print("=" * 50) | ||
|
||
results = workflow.start() | ||
|
||
print("\nWorkflow Results:") | ||
print("=" * 50) | ||
for task_id, result in results["task_results"].items(): | ||
if result: | ||
task_name = result.description | ||
print(f"\nTask: {task_name}") | ||
print(f"Result: {result.raw}") | ||
print("-" * 50) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,83 @@ | ||
from praisonaiagents.agent import Agent | ||
from praisonaiagents.task import Task | ||
from praisonaiagents.agents import PraisonAIAgents | ||
import time | ||
|
||
def get_time_check(): | ||
current_time = int(time.time()) | ||
result = "even" if current_time % 2 == 0 else "odd" | ||
print(f"Time check: {current_time} is {result}") | ||
return result | ||
|
||
# Create specialized agents | ||
router = Agent( | ||
name="Router", | ||
role="Input Router", | ||
goal="Evaluate input and determine routing path", | ||
instructions="Analyze input and decide whether to proceed or exit", | ||
tools=[get_time_check] | ||
) | ||
|
||
processor1 = Agent( | ||
name="Processor 1", | ||
role="Secondary Processor", | ||
goal="Process valid inputs that passed initial check", | ||
instructions="Process data that passed the routing check" | ||
) | ||
|
||
processor2 = Agent( | ||
name="Processor 2", | ||
role="Final Processor", | ||
goal="Perform final processing on validated data", | ||
instructions="Generate final output for processed data" | ||
) | ||
|
||
# Create tasks with routing logic | ||
routing_task = Task( | ||
name="initial_routing", | ||
description="check the time and return according to what is returned", | ||
expected_output="pass or fail based on what is returned", | ||
agent=router, | ||
is_start=True, | ||
task_type="decision", | ||
condition={ | ||
"pass": ["process_valid"], | ||
"fail": ["process_invalid"] | ||
} | ||
) | ||
|
||
processing_task = Task( | ||
name="process_valid", | ||
description="Process validated input", | ||
expected_output="Processed data ready for final step", | ||
agent=processor1, | ||
) | ||
|
||
final_task = Task( | ||
name="process_invalid", | ||
description="Generate final output", | ||
expected_output="Final processed result", | ||
agent=processor2 | ||
) | ||
|
||
# Create and run workflow | ||
workflow = PraisonAIAgents( | ||
agents=[router, processor1, processor2], | ||
tasks=[routing_task, processing_task, final_task], | ||
process="workflow", | ||
verbose=True | ||
) | ||
|
||
print("\nStarting Routing Workflow...") | ||
print("=" * 50) | ||
|
||
results = workflow.start() | ||
|
||
print("\nWorkflow Results:") | ||
print("=" * 50) | ||
for task_id, result in results["task_results"].items(): | ||
if result: | ||
task_name = result.description | ||
print(f"\nTask: {task_name}") | ||
print(f"Result: {result.raw}") | ||
print("-" * 50) |
Oops, something went wrong.