Introduction
In this tutorial, we'll explore how to use asynchronous subagents in the Hermes Agent framework from Nous Research. Asynchronous subagents allow you to delegate tasks to background agents without blocking the main conversation thread. This is particularly useful for complex workflows where you want to keep the user experience smooth while performing time-consuming operations.
This tutorial will guide you through setting up the environment, creating a basic asynchronous delegation system, and understanding how to manage background tasks effectively.
Prerequisites
Before starting this tutorial, you should have:
- A basic understanding of Python programming
- Python 3.7 or higher installed on your system
- Access to the Hermes Agent framework (either through GitHub or a local installation)
- Basic knowledge of how to use command-line tools
Step-by-Step Instructions
1. Setting Up Your Environment
First, we need to create a new Python environment to work with. This ensures we don't interfere with other projects on your system.
python -m venv hermes_env
source hermes_env/bin/activate # On Windows: hermes_env\Scripts\activate
Next, install the required dependencies:
pip install hermes-agent
This command installs the Hermes Agent framework along with its dependencies. The framework provides the core tools we'll use for asynchronous delegation.
2. Creating a Basic Agent Setup
Now, let's create a simple Python script to initialize our agent. This script will set up the basic framework for handling asynchronous tasks.
import asyncio
from hermes_agent import Agent
# Initialize the main agent
agent = Agent(name="MainAgent")
async def main():
print("Main agent initialized successfully!")
# We'll add more functionality here
if __name__ == "__main__":
asyncio.run(main())
This code creates a basic agent structure. The Agent class is the core component of Hermes that handles task delegation. We're using asyncio because asynchronous operations are at the heart of this functionality.
3. Implementing the Async Delegation Tool
Let's now implement the core functionality for spawning asynchronous subagents. We'll create a function that can delegate tasks to background agents:
import asyncio
from hermes_agent import Agent
async def spawn_async_task(agent, task_description):
"""Spawn an asynchronous subagent to handle a task"""
# Create a new subagent
subagent = Agent(name=f"SubAgent_{len(agent.subagents) + 1}")
# Add the task to the subagent
subagent.add_task(task_description)
# Start the subagent in the background
task = asyncio.create_task(subagent.run())
print(f"Task '{task_description}' delegated to subagent {subagent.name}")
return task
async def main():
agent = Agent(name="MainAgent")
# Spawn a few tasks
task1 = await spawn_async_task(agent, "Research latest AI trends")
task2 = await spawn_async_task(agent, "Generate report summary")
print("Tasks delegated, main agent continues working...")
# Wait for tasks to complete
await asyncio.gather(task1, task2)
print("All tasks completed!")
if __name__ == "__main__":
asyncio.run(main())
Here, we're creating a spawn_async_task function that creates a new subagent for each task. The asyncio.create_task function ensures that the subagent runs in the background without blocking the main thread. This is the key concept behind asynchronous delegation.
4. Adding Task Monitoring and Control
Next, we'll enhance our system to include monitoring capabilities. This allows us to check on the status of our background tasks:
import asyncio
from hermes_agent import Agent
class TaskManager:
def __init__(self):
self.tasks = {}
self.agent = Agent(name="MainAgent")
async def spawn_task(self, task_description):
subagent = Agent(name=f"SubAgent_{len(self.tasks) + 1}")
subagent.add_task(task_description)
# Store the task with its ID
task_id = f"task_{len(self.tasks) + 1}"
task = asyncio.create_task(subagent.run())
self.tasks[task_id] = {
'subagent': subagent,
'task': task,
'status': 'running'
}
print(f"Task '{task_description}' started with ID {task_id}")
return task_id
def check_task_status(self, task_id):
if task_id in self.tasks:
return self.tasks[task_id]['status']
return "Task not found"
async def wait_for_task(self, task_id):
if task_id in self.tasks:
await self.tasks[task_id]['task']
self.tasks[task_id]['status'] = 'completed'
print(f"Task {task_id} completed!")
else:
print(f"Task {task_id} not found")
async def main():
task_manager = TaskManager()
# Spawn tasks
task1_id = await task_manager.spawn_task("Research latest AI trends")
task2_id = await task_manager.spawn_task("Generate report summary")
# Check status
print(f"Task {task1_id} status: {task_manager.check_task_status(task1_id)}")
# Wait for completion
await task_manager.wait_for_task(task1_id)
await task_manager.wait_for_task(task2_id)
print("All tasks completed successfully!")
if __name__ == "__main__":
asyncio.run(main())
This enhanced version introduces a TaskManager class that keeps track of all tasks. This is important because it allows you to monitor and control background processes without losing track of them. The status tracking helps in debugging and managing complex workflows.
5. Handling Task Results and Collection
Finally, let's add functionality to collect results from our subagents:
import asyncio
from hermes_agent import Agent
async def spawn_and_collect_task(agent, task_description):
"""Spawn a task and collect its result"""
subagent = Agent(name=f"SubAgent_{len(agent.subagents) + 1}")
subagent.add_task(task_description)
# Run the task in background
task = asyncio.create_task(subagent.run())
# Wait for completion
await task
# Collect the result
result = subagent.get_result()
print(f"Result from {task_description}: {result}")
return result
async def main():
agent = Agent(name="MainAgent")
# Spawn and collect results
result1 = await spawn_and_collect_task(agent, "Research latest AI trends")
result2 = await spawn_and_collect_task(agent, "Generate report summary")
print("All results collected successfully!")
print(f"Results: {result1}, {result2}")
if __name__ == "__main__":
asyncio.run(main())
This final version demonstrates how to collect results from background tasks. The key concept here is that while tasks run asynchronously, we can still retrieve their outputs when needed. This makes the system both efficient and functional.
Summary
In this tutorial, we've learned how to implement asynchronous subagents using the Hermes Agent framework. We started with basic setup and gradually built up to a complete system that can spawn, monitor, and collect results from background tasks. The core benefit of this approach is that it allows the main agent to continue working while background tasks are processed, significantly improving user experience.
The asynchronous delegation system we've built is particularly useful for complex workflows where multiple operations need to happen simultaneously. By understanding how to implement these patterns, you can create more responsive and efficient AI systems.
