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": "Help me to write a poem"
    },
    "id": "call_878aaced6a844877a53dd2"
}
system: {
    "type": "tool_result",
    "id": "call_878aaced6a844877a53dd2",
    "name": "generate_response",
    "output": [
        {
            "type": "text",
            "text": "Successfully generated response."
        }
    ]
}
Router: I understand you're looking for some help with writing a poem. That sounds like a wonderful creative endeavor! To get started, could you share a bit more about what you have in mind? For example, do you have a specific theme or style in mind? Any particular emotions or messages you'd like to convey? The more details you can provide, the better I can assist you in crafting something beautiful and meaningful.
The structured output:
{
    "your_choice": "Content Generation",
    "task_description": "Help me to write a poem"
}

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": "Generate a quick sort function."
    },
    "id": "call_0b3271082df04ae7b17bee"
}
PythonAgent: Certainly! Below is a Python implementation of the Quick Sort algorithm. This version uses the Lomuto partition scheme, which is simpler to understand and implement.

```python
def quick_sort(arr):
    # Helper function to perform the partitioning
    def partition(low, high):
        pivot = arr[high]
        i = low - 1
        for j in range(low, high):
            if arr[j] < pivot:
                i += 1
                arr[i], arr[j] = arr[j], arr[i]
        arr[i + 1], arr[high] = arr[high], arr[i + 1]
        return i + 1

    # Recursive function to perform the quick sort
    def quick_sort_recursive(low, high):
        if low < high:
            pi = partition(low, high)
            quick_sort_recursive(low, pi - 1)
            quick_sort_recursive(pi + 1, high)

    # Initial call to the recursive function
    quick_sort_recursive(0, len(arr) - 1)
    return arr

# Example usage:
arr = [10, 7, 8, 9, 1, 5]
sorted_arr = quick_sort(arr)
print("Sorted array:", sorted_arr)
```

### Explanation:
- **partition(low, high)**: This function takes the last element as the pivot, places the pivot element at its correct position in the sorted array, and places all smaller elements to the left of the pivot and all greater elements to the right.
- **quick_sort_recursive(low, high)**: This function recursively sorts the sub-arrays before and after the pivot.
- **quick_sort(arr)**: This is the main function that initializes the recursive sorting process.

### Example Output:
If you run the example with the array `[10, 7, 8, 9, 1, 5]`, the output will be:
```
Sorted array: [1, 5, 7, 8, 9, 10]
```

This implementation is in-place, meaning it doesn't require additional memory proportional to the input size.
system: {
    "type": "tool_result",
    "id": "call_0b3271082df04ae7b17bee",
    "name": "generate_python",
    "output": [
        {
            "type": "text",
            "text": "Certainly! Below is a Python implementation of the Quick Sort algorithm. This version uses the Lomuto partition scheme, which is simpler to understand and implement.\n\n```python\ndef quick_sort(arr):\n    # Helper function to perform the partitioning\n    def partition(low, high):\n        pivot = arr[high]\n        i = low - 1\n        for j in range(low, high):\n            if arr[j] < pivot:\n                i += 1\n                arr[i], arr[j] = arr[j], arr[i]\n        arr[i + 1], arr[high] = arr[high], arr[i + 1]\n        return i + 1\n\n    # Recursive function to perform the quick sort\n    def quick_sort_recursive(low, high):\n        if low < high:\n            pi = partition(low, high)\n            quick_sort_recursive(low, pi - 1)\n            quick_sort_recursive(pi + 1, high)\n\n    # Initial call to the recursive function\n    quick_sort_recursive(0, len(arr) - 1)\n    return arr\n\n# Example usage:\narr = [10, 7, 8, 9, 1, 5]\nsorted_arr = quick_sort(arr)\nprint(\"Sorted array:\", sorted_arr)\n```\n\n### Explanation:\n- **partition(low, high)**: This function takes the last element as the pivot, places the pivot element at its correct position in the sorted array, and places all smaller elements to the left of the pivot and all greater elements to the right.\n- **quick_sort_recursive(low, high)**: This function recursively sorts the sub-arrays before and after the pivot.\n- **quick_sort(arr)**: This is the main function that initializes the recursive sorting process.\n\n### Example Output:\nIf you run the example with the array `[10, 7, 8, 9, 1, 5]`, the output will be:\n```\nSorted array: [1, 5, 7, 8, 9, 10]\n```\n\nThis implementation is in-place, meaning it doesn't require additional memory proportional to the input size."
        }
    ]
}
Router: The Python code for a quick sort function has been generated. Here's the implementation using the Lomuto partition scheme, which is simple to understand and implement:

```python
def quick_sort(arr):
    # Helper function to perform the partitioning
    def partition(low, high):
        pivot = arr[high]
        i = low - 1
        for j in range(low, high):
            if arr[j] < pivot:
                i += 1
                arr[i], arr[j] = arr[j], arr[i]
        arr[i + 1], arr[high] = arr[high], arr[i + 1]
        return i + 1

    # Recursive function to perform the quick sort
    def quick_sort_recursive(low, high):
        if low < high:
            pi = partition(low, high)
            quick_sort_recursive(low, pi - 1)
            quick_sort_recursive(pi + 1, high)

    # Initial call to the recursive function
    quick_sort_recursive(0, len(arr) - 1)
    return arr

# Example usage:
arr = [10, 7, 8, 9, 1, 5]
sorted_arr = quick_sort(arr)
print("Sorted array:", sorted_arr)
```

### Explanation:
- **partition(low, high):** This function takes the last element as the pivot, places the pivot element at its correct position in the sorted array, and places all smaller elements to the left of the pivot and all greater elements to the right.
- **quick_sort_recursive(low, high):** This function recursively sorts the sub-arrays before and after the pivot.
- **quick_sort(arr):** This is the main function that initializes the recursive sorting process.

### Example Output:
If you run the example with the array `[10, 7, 8, 9, 1, 5]`, the output will be:
```
Sorted array: [1, 5, 7, 8, 9, 10]
```

This implementation is in-place, meaning it doesn't require additional memory proportional to the input size. You can use this code to sort arrays in your Python program.