Note
Go to the end to download the full example code.
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:231: DeprecationWarning: 'required' is not supported by DashScope API. It will be converted to 'auto'.
warnings.warn(
Router: {
"type": "tool_use",
"name": "generate_response",
"input": {
"task_description": "Help me to write a poem",
"your_choice": "Content Generation"
},
"id": "call_87dad27ae4844fc5bb51f9"
}
system: {
"type": "tool_result",
"id": "call_87dad27ae4844fc5bb51f9",
"name": "generate_response",
"output": [
{
"type": "text",
"text": "Successfully generated response."
}
]
}
Router: I understand you'd like help with writing a poem. I can certainly assist with that! To get started, could you please share more details? For example, do you have a specific theme or style in mind? Any particular emotions or messages you wish to convey through your poem? The more information you provide, the better I can tailor my assistance to your needs.
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": "a quick sort function"
},
"id": "call_72a737deed8740459aa529"
}
PythonAgent: Certainly! Here's a Python implementation of the Quick Sort algorithm:
```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_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 lists:
- `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 sort the `left` and `right` lists and concatenate them with the `middle` list to get the final sorted 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 a more in-place version, you can use the following:
```python
def quick_sort_in_place(arr, low, high):
if low < high:
pi = partition(arr, low, high)
quick_sort_in_place(arr, low, pi - 1)
quick_sort_in_place(arr, pi + 1, high)
def partition(arr, 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
# Example usage:
arr = [3, 6, 8, 10, 1, 2, 1]
quick_sort_in_place(arr, 0, len(arr) - 1)
print(arr)
```
### Explanation:
- **In-Place Partitioning**: The `partition` function rearranges the elements in the array such that all elements less than the pivot are on the left, and all elements greater than the pivot are on the right.
- **Recursive Sorting**: The `quick_sort_in_place` function recursively sorts the sub-arrays on both sides of the pivot.
This in-place version is more space-efficient and is generally preferred for larger arrays.
system: {
"type": "tool_result",
"id": "call_72a737deed8740459aa529",
"name": "generate_python",
"output": [
{
"type": "text",
"text": "Certainly! Here's a Python implementation of the Quick Sort algorithm:\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_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 lists:\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 sort the `left` and `right` lists and concatenate them with the `middle` list to get the final sorted 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 a more in-place version, you can use the following:\n\n```python\ndef quick_sort_in_place(arr, low, high):\n if low < high:\n pi = partition(arr, low, high)\n quick_sort_in_place(arr, low, pi - 1)\n quick_sort_in_place(arr, pi + 1, high)\n\ndef partition(arr, 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# Example usage:\narr = [3, 6, 8, 10, 1, 2, 1]\nquick_sort_in_place(arr, 0, len(arr) - 1)\nprint(arr)\n```\n\n### Explanation:\n- **In-Place Partitioning**: The `partition` function rearranges the elements in the array such that all elements less than the pivot are on the left, and all elements greater than the pivot are on the right.\n- **Recursive Sorting**: The `quick_sort_in_place` function recursively sorts the sub-arrays on both sides of the pivot.\n\nThis in-place version is more space-efficient and is generally preferred for larger arrays."
}
]
}
Router: The Python code for a quick sort function has been generated. Here's the provided implementation:
```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_arr)
```
Additionally, an in-place version of the quick sort algorithm was also provided, which is more space-efficient and suitable for larger arrays:
```python
def quick_sort_in_place(arr, low, high):
if low < high:
pi = partition(arr, low, high)
quick_sort_in_place(arr, low, pi - 1)
quick_sort_in_place(arr, pi + 1, high)
def partition(arr, 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
# Example usage:
arr = [3, 6, 8, 10, 1, 2, 1]
quick_sort_in_place(arr, 0, len(arr) - 1)
print(arr)
```
You can use either of these implementations based on your preference and the requirements of your project. If you need further assistance or have any other questions, feel free to ask!
Total running time of the script: (0 minutes 56.920 seconds)