Routing

There are two ways to implement routing in AgentScope, both simple and easy to implement:

• Routing by structured output

• Routing by tool calls

Tip

Considering there is no unified standard/definition for agent routing, we follow the setting in Building effective agents

Routing by Structured Output

By this way, we can directly use the structured output of the agent to determine which agent to route the message to.

Initialize a routing agent

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())

/home/runner/work/agentscope/agentscope/src/agentscope/model/_dashscope_model.py:182: DeprecationWarning: 'required' is not supported by DashScope API. It will be converted to 'auto'.
  warnings.warn(
Router: {
    "type": "tool_use",
    "name": "generate_response",
    "input": {
        "your_choice": "Content Generation",
        "task_description": "Write a poem as per the user's request"
    },
    "id": "call_cac74b7f648d46978e9498"
}
system: {
    "type": "tool_result",
    "id": "call_cac74b7f648d46978e9498",
    "name": "generate_response",
    "output": [
        {
            "type": "text",
            "text": "Successfully generated response."
        }
    ]
}
Router: Sure, I can help with that! Could you please provide me with some themes or specific elements you'd like to include in your poem? This could be anything from a particular mood, a story, nature, love, or any other ideas you have in mind.
The structured output:
{
    "your_choice": "Content Generation",
    "task_description": "Write a poem as per the user's request"
}

Routing by Tool Calls

Another way is to wrap the downstream agents into a tool function, so that the routing agent decides which tool to call based on the user query.

We first define several tool functions:

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

After that, we define a routing agent and equip it with the above tool functions.

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_a1fdbed651a3473ba6f2d2"
}
PythonAgent: Certainly! Here's a Python implementation of the Quick Sort algorithm. This is a popular and efficient sorting algorithm that uses a divide-and-conquer approach to sort elements.

```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)

# Example usage:
arr = [3, 6, 8, 10, 1, 2, 1]
sorted_arr = quick_sort(arr)
print("Sorted array:", sorted_arr)
```

### Explanation:
- **Base Case**: If the array has 0 or 1 elements, it is already sorted, so we return it as is.
- **Pivot Selection**: We choose the middle element of the array as the pivot.
- **Partitioning**: We create three sub-arrays:
  - `left`: All elements less than the pivot.
  - `middle`: All elements equal to the pivot.
  - `right`: All elements greater than the pivot.
- **Recursive Sorting**: We recursively apply the `quick_sort` function to the `left` and `right` sub-arrays and concatenate the results with the `middle` array.

This implementation is simple and easy to understand, but it may not be the most efficient in terms of space complexity due to the use of additional lists. For large datasets, an in-place version of Quick Sort would be more efficient.
system: {
    "type": "tool_result",
    "id": "call_a1fdbed651a3473ba6f2d2",
    "name": "generate_python",
    "output": [
        {
            "type": "text",
            "text": "Certainly! Here's a Python implementation of the Quick Sort algorithm. This is a popular and efficient sorting algorithm that uses a divide-and-conquer approach to sort elements.\n\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\n# Example usage:\narr = [3, 6, 8, 10, 1, 2, 1]\nsorted_arr = quick_sort(arr)\nprint(\"Sorted array:\", sorted_arr)\n```\n\n### Explanation:\n- **Base Case**: If the array has 0 or 1 elements, it is already sorted, so we return it as is.\n- **Pivot Selection**: We choose the middle element of the array as the pivot.\n- **Partitioning**: We create three sub-arrays:\n  - `left`: All elements less than the pivot.\n  - `middle`: All elements equal to the pivot.\n  - `right`: All elements greater than the pivot.\n- **Recursive Sorting**: We recursively apply the `quick_sort` function to the `left` and `right` sub-arrays and concatenate the results with the `middle` array.\n\nThis implementation is simple and easy to understand, but it may not be the most efficient in terms of space complexity due to the use of additional lists. For large datasets, an in-place version of Quick Sort would be more efficient."
        }
    ]
}
Router: The Python function for the Quick Sort algorithm has been generated. Here's the code:

```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)

# Example usage:
arr = [3, 6, 8, 10, 1, 2, 1]
sorted_arr = quick_sort(arr)
print("Sorted array:", sorted_arr)
```

This implementation of Quick Sort uses a divide-and-conquer approach to sort elements. It selects a pivot (in this case, the middle element), and then partitions the other elements into those less than, equal to, and greater than the pivot, before recursively sorting the sub-arrays. This version is simple and clear but may not be the most space-efficient for large datasets, as it creates new lists during each recursive call.

You can use the `quick_sort` function by passing an array or list of numbers to it, and it will return a new list with the elements sorted.