路由

在AgentScope中有两种方式实现路由，既简单又易于实现：

• 通过结构化输出进行路由

• 通过工具调用进行路由

提示

考虑目前智能体路由尚无统一标准/定义，我们遵循Building effective agents中的设定

通过结构化输出进行路由

通过这种方式，我们可以直接利用智能体的结构化输出来确定将消息路由到哪个智能体。

初始化一个路由智能体

import asyncio
import json
import os
from typing import Literal

from pydantic import BaseModel, Field

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 Toolkit, ToolResponse

router = ReActAgent(
    name="Router",
    sys_prompt="You're a routing agent. Your target is to route the user query to the right follow-up task.",
    model=DashScopeChatModel(
        model_name="qwen-max",
        api_key=os.environ["DASHSCOPE_API_KEY"],
        stream=False,
    ),
    formatter=DashScopeChatFormatter(),
)


# Use structured output to specify the routing task
class RoutingChoice(BaseModel):
    your_choice: Literal[
        "Content Generation",
        "Programming",
        "Information Retrieval",
        None,
    ] = Field(
        description="Choose the right follow-up task, and choose ``None`` if the task is too simple or no suitable task",
    )
    task_description: str | None = Field(
        description="The task description",
        default=None,
    )


async def example_router_explicit() -> None:
    """Example of explicit routing with structured output."""
    msg_user = Msg(
        "user",
        "Help me to write a poem",
        "user",
    )

    # Route the query
    msg_res = await router(
        msg_user,
        structured_model=RoutingChoice,
    )

    # The structured output is stored in the metadata field
    print("The structured output:")
    print(json.dumps(msg_res.metadata, indent=4, ensure_ascii=False))


asyncio.run(example_router_explicit())

Router: Of course, I can help you with that. Could you please tell me more about the theme or any specific elements you would like to include in your poem?
The structured output:
{
    "your_choice": "Content Generation",
    "task_description": null
}

通过工具调用进行路由

另一种方式是包装下游智能体作为工具函数，这样才能让路由智能体依据用户查询来决定调用哪个工具。

我们首先定义几个工具函数：

async def generate_python(demand: str) -> ToolResponse:
    """Generate Python code based on the demand.

    Args:
        demand (``str``):
            The demand for the Python code.
    """
    # An example demand agent
    python_agent = ReActAgent(
        name="PythonAgent",
        sys_prompt="You're a Python expert, your target is to generate Python code based on the demand.",
        model=DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.environ["DASHSCOPE_API_KEY"],
            stream=False,
        ),
        memory=InMemoryMemory(),
        formatter=DashScopeChatFormatter(),
        toolkit=Toolkit(),
    )
    msg_res = await python_agent(Msg("user", demand, "user"))

    return ToolResponse(
        content=msg_res.get_content_blocks("text"),
    )


# Fake some other tool functions for demonstration purposes
async def generate_poem(demand: str) -> ToolResponse:
    """Generate a poem based on the demand.

    Args:
        demand (``str``):
            The demand for the poem.
    """
    pass


async def web_search(query: str) -> ToolResponse:
    """Search the web for the query.

    Args:
        query (``str``):
            The query to search.
    """
    pass

接下来，我们定义一个路由智能体并为其配备上述工具函数。

toolkit = Toolkit()
toolkit.register_tool_function(generate_python)
toolkit.register_tool_function(generate_poem)
toolkit.register_tool_function(web_search)

# Initialize the routing agent with the toolkit
router_implicit = ReActAgent(
    name="Router",
    sys_prompt="You're a routing agent. Your target is to route the user query to the right follow-up task.",
    model=DashScopeChatModel(
        model_name="qwen-max",
        api_key=os.environ["DASHSCOPE_API_KEY"],
        stream=False,
    ),
    formatter=DashScopeChatFormatter(),
    toolkit=toolkit,
    memory=InMemoryMemory(),
)


async def example_router_implicit() -> None:
    """Example of implicit routing with tool calls."""
    msg_user = Msg(
        "user",
        "Help me to generate a quick sort function in Python",
        "user",
    )

    # Route the query
    await router_implicit(msg_user)


asyncio.run(example_router_implicit())

Router: {
    "type": "tool_use",
    "name": "generate_python",
    "input": {
        "demand": "a quick sort function"
    },
    "id": "call_650806e22efd476b8502ef"
}
PythonAgent: Here is a simple implementation of the quick sort algorithm in Python:
```python
def quick_sort(arr):
    if len(arr) <= 1:
        return arr
    else:
        pivot = arr[len(arr) // 2]
        left = [x for x in arr if x < pivot]
        middle = [x for x in arr if x == pivot]
        right = [x for x in arr if x > pivot]
        return quick_sort(left) + middle + quick_sort(right)
```
You can use this function to sort a list of numbers. For example, `quick_sort([3, 6, 8, 10, 1, 2, 1])` will return a sorted list `[1, 1, 2, 3, 6, 8, 10]`.
system: {
    "type": "tool_result",
    "id": "call_650806e22efd476b8502ef",
    "name": "generate_python",
    "output": [
        {
            "type": "text",
            "text": "Here is a simple implementation of the quick sort algorithm in Python:\n```python\ndef quick_sort(arr):\n    if len(arr) <= 1:\n        return arr\n    else:\n        pivot = arr[len(arr) // 2]\n        left = [x for x in arr if x < pivot]\n        middle = [x for x in arr if x == pivot]\n        right = [x for x in arr if x > pivot]\n        return quick_sort(left) + middle + quick_sort(right)\n```\nYou can use this function to sort a list of numbers. For example, `quick_sort([3, 6, 8, 10, 1, 2, 1])` will return a sorted list `[1, 1, 2, 3, 6, 8, 10]`."
        }
    ]
}
Router: Here is a simple implementation of the quick sort algorithm in Python:
```python
def quick_sort(arr):
    if len(arr) <= 1:
        return arr
    else:
        pivot = arr[len(arr) // 2]
        left = [x for x in arr if x < pivot]
        middle = [x for x in arr if x == pivot]
        right = [x for x in arr if x > pivot]
        return quick_sort(left) + middle + quick_sort(right)
```
You can use this function to sort a list of numbers. For example, `quick_sort([3, 6, 8, 10, 1, 2, 1])` will return a sorted list `[1, 1, 2, 3, 6, 8, 10]`.