Skip to content

Add parallel workflow execution #761

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Bryan (Bryan-Roe) merged 3 commits into from
Aug 3, 2025
Merged

Add parallel workflow execution #761

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4f8778e
Add parallel workflow support to local agent demo
Bryan-Roe Jul 27, 2025
1225568
Update demo_local_agents.py
Bryan-Roe Aug 3, 2025
155b027
Update demo_local_agents.py
Bryan-Roe Aug 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 68 additions & 6 deletions demo_local_agents.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
#!/usr/bin/env python3
"""
Demonstration of local AI agents and capabilities
Demonstration of local AI agents and capabilities.
Includes parallel workflow support.

Copyright (c) 2025 Bryan Roe
Licensed under the MIT License
Expand Down Expand Up @@ -124,6 +125,44 @@ async def process_workflow(self, workflow: list) -> dict:

return results

async def process_workflow_parallel(self, workflow: list) -> dict:

@sourcery-ai sourcery-ai Bot Jul 27, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (complexity): Consider merging the sequential and parallel workflow methods into a single function with a parallel flag and reusing the workflow definition in the demo.

Suggested change
async def process_workflow_parallel(self, workflow: list) -> dict:
You can collapse the two nearly identical methods and demo branches by:
1. Merging `process_workflow` and `process_workflow_parallel` into one `process_workflow` with a `parallel: bool` flag.
2. Extracting the demo workflow definition once and re-using it for both sequential and parallel runs.
Example refactor:
```python
class AGIWorkflowOrchestrator:
async def process_workflow(self, workflow: list, parallel: bool = False) -> dict:
"""Process workflow tasks, either sequentially or in parallel."""
results = {}
tasks, names = [], []
for step in workflow:
agent = step["agent"]
fn = step["function"]
params = step.get("params", {})
try:
function = self.kernel.get_function(agent, fn)
if parallel:
tasks.append(asyncio.create_task(function.invoke(self.kernel, **params)))
names.append(f"{agent}_{fn}")
logger.info(f"🚀 Scheduled: {agent}.{fn}")
else:
res = await function.invoke(self.kernel, **params)
results[f"{agent}_{fn}"] = str(res)
logger.info(f"✅ Completed: {agent}.{fn}")
except Exception as e:
logger.error(f"❌ Failed {agent}.{fn} - {e}")
results[f"{agent}_{fn}"] = f"Error: {e}"
if parallel and tasks:
completed = await asyncio.gather(*tasks, return_exceptions=True)
for name, res in zip(names, completed):
if isinstance(res, Exception):
logger.error(f"❌ Failed: {name} - {res}")
results[name] = f"Error: {res}"
else:
logger.info(f"✅ Completed: {name}")
results[name] = str(res)
return results

Then in your demo:

async def demo_local_agents():
    demo_workflow = [
        {"agent": "monitor_agent", "function": "monitor_performance"},
        {"agent": "file_agent",    "function": "file_operation", "params": {"operation":"create","filename":"test.txt"}},
        {"agent": "chat_agent",    "function": "process_request", "params": {"request":"status"}},
        {"agent": "file_agent",    "function": "file_operation", "params": {"operation":"read","filename":"test.txt"}},
    ]

    # inside your menu:
    if choice in ("5", "6"):
        parallel = (choice == "6")
        mode = "in parallel" if parallel else "sequentially"
        print(f"🔄 Running workflow {mode}...")
        results = await orchestrator.process_workflow(demo_workflow, parallel=parallel)
        for step, res in results.items():
            print(f"  ✅ {step}: {res}")

This removes duplication in both the executor and the demo.

"""
Process workflow tasks concurrently across agents.

Unlike sequential execution, tasks are executed concurrently without any dependency ordering.
This means that tasks are scheduled to run simultaneously, and their execution order is not guaranteed.
Developers should be aware of potential race conditions if tasks depend on shared resources or have interdependencies.
Ensure that workflows are designed to avoid conflicts and unintended behavior due to concurrency.
"""
results = {}
tasks = []
task_names = []

for step in workflow:
agent_id = step.get("agent")
function_name = step.get("function")
params = step.get("params", {})

try:
function = self.kernel.get_function(agent_id, function_name)
tasks.append(asyncio.create_task(function.invoke(self.kernel, **params)))
task_names.append(f"{agent_id}_{function_name}")
logger.info(f"🚀 Scheduled: {agent_id}.{function_name}")
except Exception as e:
logger.error(f"❌ Failed to schedule {agent_id}.{function_name} - {e}")
results[f"{agent_id}_{function_name}"] = f"Error: {e}"

completed = await asyncio.gather(*tasks, return_exceptions=True)
for name, res in zip(task_names, completed):
Comment on lines +154 to +156

@sourcery-ai sourcery-ai Bot Jul 27, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion: Potential issue if all tasks fail to schedule.

Consider adding a warning or log message if no tasks are scheduled, as this may indicate a configuration issue and the results will only reflect scheduling errors.

Suggested change
completed = await asyncio.gather(*tasks, return_exceptions=True)
for name, res in zip(task_names, completed):
if not tasks:
logger.warning("⚠️ No tasks were scheduled. This may indicate a configuration issue. Results will only reflect scheduling errors.")
completed = await asyncio.gather(*tasks, return_exceptions=True)
for name, res in zip(task_names, completed):

if isinstance(res, Exception):
logger.error(f"❌ Failed: {name} - {res}")
results[name] = f"Error: {res}"
else:
logger.info(f"✅ Completed: {name}")
results[name] = str(res)

return results

async def demo_local_agents():

@sourcery-ai sourcery-ai Bot Jul 27, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (code-quality): Low code quality found in demo_local_agents - 24% (low-code-quality)


ExplanationThe quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.

How can you solve this?

It might be worth refactoring this function to make it shorter and more readable.

  • Reduce the function length by extracting pieces of functionality out into
    their own functions. This is the most important thing you can do - ideally a
    function should be less than 10 lines.
  • Reduce nesting, perhaps by introducing guard clauses to return early.
  • Ensure that variables are tightly scoped, so that code using related concepts
    sits together within the function rather than being scattered.

"""Demonstrate local AGI agents"""
print("🚀 Starting Local AGI Agent Demo")
Expand All @@ -150,10 +189,11 @@ async def demo_local_agents():
print("2. File operations")
print("3. System monitoring")
print("4. Agent information")
print("5. Run workflow")
print("6. Exit")
print("5. Run workflow (sequential)")
print("6. Run workflow (parallel)")
print("7. Exit")

choice = input("\nSelect option (1-6): ").strip()
choice = input("\nSelect option (1-7): ").strip()

if choice == "1":
message = input("Enter message: ")
Expand Down Expand Up @@ -183,20 +223,42 @@ async def demo_local_agents():
print("❌ Agent not found")

elif choice == "5":
# Demo workflow
# Demo workflow (sequential)
workflow = [
{"agent": "monitor_agent", "function": "monitor_performance"},
{"agent": "file_agent", "function": "file_operation", "params": {"operation": "create", "filename": "test.txt"}},
{"agent": "chat_agent", "function": "process_request", "params": {"request": "status"}},
{"agent": "file_agent", "function": "file_operation", "params": {"operation": "read", "filename": "test.txt"}}
]

print("🔄 Running workflow...")
print("🔄 Running workflow sequentially...")
results = await orchestrator.process_workflow(workflow)
for step, result in results.items():
print(f" ✅ {step}: {result}")

elif choice == "6":
# Run the same demo workflow in parallel
# Split workflow into two groups to avoid race conditions
initial_workflow = [
{"agent": "monitor_agent", "function": "monitor_performance"},
{"agent": "file_agent", "function": "file_operation", "params": {"operation": "create", "filename": "test.txt"}},
{"agent": "chat_agent", "function": "process_request", "params": {"request": "status"}}
]

dependent_workflow = [
{"agent": "file_agent", "function": "file_operation", "params": {"operation": "read", "filename": "test.txt"}}
]

print("🔄 Running initial workflow in parallel...")
initial_results = await orchestrator.process_workflow_parallel(initial_workflow)
for step, result in initial_results.items():
print(f" ✅ {step}: {result}")

print("🔄 Running dependent workflow sequentially...")
dependent_results = await orchestrator.process_workflow(dependent_workflow)
for step, result in dependent_results.items():
print(f" ✅ {step}: {result}")
elif choice == "7":
print("👋 Goodbye!")
break
else:
Expand Down
Loading