Handoffs

Handoffs example

It’s very simple to implement the Orchestrator-Workers workflow with tool calls in AgentScope. First, we create a function to allow the orchestrator to create workers dynamically.

Orchestrator: {
    "type": "tool_use",
    "name": "create_worker",
    "input": {
        "task_description": "Print 'Hello, World!' using Python"
    },
    "id": "call_3f25507090c049d8ac452b"
}
Worker: {
    "type": "tool_use",
    "name": "execute_python_code",
    "input": {
        "code": "print('Hello, World!')",
        "timeout": 300
    },
    "id": "call_d7e402d1f5084230be201c"
}
system: {
    "type": "tool_result",
    "id": "call_d7e402d1f5084230be201c",
    "name": "execute_python_code",
    "output": [
        {
            "type": "text",
            "text": "<returncode>0</returncode><stdout>Hello, World!\n</stdout><stderr></stderr>"
        }
    ]
}
Worker: The Python code has been executed successfully and printed 'Hello, World!' as expected.
system: {
    "type": "tool_result",
    "id": "call_3f25507090c049d8ac452b",
    "name": "create_worker",
    "output": [
        {
            "type": "text",
            "text": "The Python code has been executed successfully and printed 'Hello, World!' as expected."
        }
    ]
}
Orchestrator: The worker has successfully executed the Python code and printed 'Hello, World!'.

import asyncio
import os

from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeChatFormatter
from agentscope.memory import InMemoryMemory
from agentscope.message import Msg
from agentscope.model import DashScopeChatModel
from agentscope.tool import (
    ToolResponse,
    Toolkit,
    execute_python_code,
)


# The tool function to create a worker
async def create_worker(
    task_description: str,
) -> ToolResponse:
    """Create a worker to finish the given task. The worker is equipped with python execution tool.

    Args:
        task_description (``str``):
            The description of the task to be finished by the worker.
    """
    # Equip the worker agent with some tools
    toolkit = Toolkit()
    toolkit.register_tool_function(execute_python_code)

    # Create a worker agent
    worker = ReActAgent(
        name="Worker",
        sys_prompt="You're a worker agent. Your target is to finish the given task.",
        model=DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.environ["DASHSCOPE_API_KEY"],
            stream=False,
        ),
        formatter=DashScopeChatFormatter(),
        toolkit=toolkit,
    )
    # Let the worker finish the task
    res = await worker(Msg("user", task_description, "user"))
    return ToolResponse(
        content=res.get_content_blocks("text"),
    )


async def run_handoffs() -> None:
    """Example of handoffs workflow."""
    # Initialize the orchestrator agent
    toolkit = Toolkit()
    toolkit.register_tool_function(create_worker)

    orchestrator = ReActAgent(
        name="Orchestrator",
        sys_prompt="You're an orchestrator agent. Your target is to finish the given task by decomposing it into smaller tasks and creating workers to finish them.",
        model=DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.environ["DASHSCOPE_API_KEY"],
            stream=False,
        ),
        memory=InMemoryMemory(),
        formatter=DashScopeChatFormatter(),
        toolkit=toolkit,
    )

    # The task description
    task_description = "Execute hello world in Python"

    # Create a worker to finish the task
    await orchestrator(Msg("user", task_description, "user"))


asyncio.run(run_handoffs())