Handoffs

Handoffs 是由 OpenAI 提出的工作流模式，通过调用子智能体的方式来完成目标任务。 在 AgentScope 中通过工具调用的方式实现 handoffs 非常简单。首先，我们创建一个函数来允许协调者动态创建子智能体。

Handoffs 示例

Orchestrator: {
    "type": "tool_use",
    "name": "create_worker",
    "input": {
        "task_description": "在 Python 中编写并执行一个打印 'Hello, World!' 的简单脚本。"
    },
    "id": "call_353c0b403368442f944ec3"
}
Worker: {
    "type": "tool_use",
    "name": "execute_python_code",
    "input": {
        "code": "print('Hello, World!')",
        "timeout": 5
    },
    "id": "call_1c84478bfc7b4dd8800d0b"
}
system: {
    "type": "tool_result",
    "id": "call_1c84478bfc7b4dd8800d0b",
    "name": "execute_python_code",
    "output": [
        {
            "type": "text",
            "text": "<returncode>0</returncode><stdout>Hello, World!\n</stdout><stderr></stderr>"
        }
    ]
}
Worker: 成功执行了 Python 脚本，它打印出了 'Hello, World!'。
system: {
    "type": "tool_result",
    "id": "call_353c0b403368442f944ec3",
    "name": "create_worker",
    "output": [
        {
            "type": "text",
            "text": "成功执行了 Python 脚本，它打印出了 'Hello, World!'。"
        }
    ]
}
Orchestrator: 成功执行了 Python 脚本，它打印出了 '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,
)


# 创建子智能体的工具函数
async def create_worker(
    task_description: str,
) -> ToolResponse:
    """创建一个子智能体来完成给定的任务。子智能体配备了 Python 执行工具。

    Args:
        task_description (``str``):
            子智能体要完成的任务描述。
    """
    # 为子智能体智能体配备一些工具
    toolkit = Toolkit()
    toolkit.register_tool_function(execute_python_code)

    # 创建子智能体智能体
    worker = ReActAgent(
        name="Worker",
        sys_prompt="你是一个智能体。你的目标是完成给定的任务。",
        model=DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.environ["DASHSCOPE_API_KEY"],
            stream=False,
        ),
        formatter=DashScopeChatFormatter(),
        toolkit=toolkit,
    )
    # 让子智能体完成任务
    res = await worker(Msg("user", task_description, "user"))
    return ToolResponse(
        content=res.get_content_blocks("text"),
    )


async def run_handoffs() -> None:
    """交接工作流示例。"""
    # 初始化协调者智能体
    toolkit = Toolkit()
    toolkit.register_tool_function(create_worker)

    orchestrator = ReActAgent(
        name="Orchestrator",
        sys_prompt="你是一个协调者智能体。你的目标是通过将任务分解为更小的任务并创建子智能体来完成它们，从而完成给定的任务。",
        model=DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.environ["DASHSCOPE_API_KEY"],
            stream=False,
        ),
        memory=InMemoryMemory(),
        formatter=DashScopeChatFormatter(),
        toolkit=toolkit,
    )

    # 任务描述
    task_description = "在 Python 中执行 hello world"

    # 创建子智能体来完成任务
    await orchestrator(Msg("user", task_description, "user"))


asyncio.run(run_handoffs())