【通义千问—Qwen-Agent系列2】案例分析（图像理解图文生成Agent||多模态助手|| 基于ReAct范式的数据分析Agent）

---

前言

Qwen-Agent是一个开发框架。开发者可基于本框架开发Agent应用，充分利用基于通义千问模型（Qwen）的指令遵循、工具使用、规划、记忆能力。本框架也提供了浏览器助手、代码解释器、自定义助手等示例应用，该篇为系列2。

【通义千问——Qwen-Agent系列文章】：
【通义千问—Qwen-Agent系列1】Qwen-Agent 快速开始&使用和开发过程.
【通义千问—Qwen-Agent系列2】Qwen-Agent 的案例分析（图像理解&图文生成Agent||多模态助手|| 基于ReAct范式的数据分析Agent）
【通义千问—Qwen-Agent系列3】Qwen-Agent 的案例分析（五子棋游戏&多Agent冒险游戏&多智能体群组交流）

一、快速开始

1-1、介绍

Qwen-Agent: 是一个开发框架。开发者可基于本框架开发Agent应用，充分利用基于通义千问模型（Qwen）的指令遵循、工具使用、规划、记忆能力。本项目也提供了浏览器助手、代码解释器、自定义助手等示例应用。

1-2、安装

1、使用pip安装：

pip install -U qwen-agent

2、从Github安装最新版本

git clone https://github.com/QwenLM/Qwen-Agent.git
cd Qwen-Agent
pip install -e ./

1-3、开发你自己的Agent

概述：下面的示例说明了创建一个能够读取PDF文件和利用工具的代理的过程，以及构建自定义工具，以下为详细介绍：

• 添加一个自定义工具：图片生成工具
• 使用到的LLM模型配置。
• 创建Agent，这里我们以“Assistant”代理为例，它能够使用工具和读取文件。
• 以聊天机器人的形式运行助理。

import pprint
import urllib.parse
import json5
from qwen_agent.agents import Assistant
from qwen_agent.tools.base import BaseTool, register_tool# Step 1 (Optional): Add a custom tool named `my_image_gen`.
@register_tool('my_image_gen')
class MyImageGen(BaseTool):# The `description` tells the agent the functionality of this tool.description = 'AI painting (image generation) service, input text description, and return the image URL drawn based on text information.'# The `parameters` tell the agent what input parameters the tool has.parameters = [{'name': 'prompt','type': 'string','description': 'Detailed description of the desired image content, in English','required': True}]def call(self, params: str, **kwargs) -> str:# `params` are the arguments generated by the LLM agent.prompt = json5.loads(params)['prompt']# 对提示词进行URL编码prompt = urllib.parse.quote(prompt)# return json5.dumps({'image_url': f'https://image.pollinations.ai/prompt/{prompt}'},ensure_ascii=False)# Step 2: Configure the LLM you are using.
# 这里是需要配置模型的地方。需要填写模型名字，以及model_server，即模型所在服务器名字，如果没有，也可以考虑使用api_key。
llm_cfg = {# Use the model service provided by DashScope:# model：模型名称# model_server：模型所在的服务器# api_key： 所使用到的api-key，可以显示的设置，也可以从环境变量中获取'model': 'qwen-max','model_server': 'dashscope',# 'api_key': 'YOUR_DASHSCOPE_API_KEY',# It will use the `DASHSCOPE_API_KEY' environment variable if 'api_key' is not set here.# Use a model service compatible with the OpenAI API, such as vLLM or Ollama:# 'model': 'Qwen1.5-7B-Chat',# 'model_server': 'http://localhost:8000/v1',  # base_url, also known as api_base# 'api_key': 'EMPTY',# (Optional) LLM hyperparameters for generation:# 用于调整生成参数的可选配置'generate_cfg': {'top_p': 0.8}
}# Step 3: Create an agent. Here we use the `Assistant` agent as an example, which is capable of using tools and reading files.# agent的提示词指令
system_instruction = '''You are a helpful assistant.
After receiving the user's request, you should:
- first draw an image and obtain the image url,
- then run code `request.get(image_url)` to download the image,
- and finally select an image operation from the given document to process the image.
Please show the image using `plt.show()`.'''# 工具列表，指定Assistant可以访问的工具，一个是自定义的工具，一个是代码执行器
tools = ['my_image_gen', 'code_interpreter']  # `code_interpreter` is a built-in tool for executing code.
# 助理可以读取的文件路径
files = ['./examples/resource/doc.pdf']  # Give the bot a PDF file to read.# 初始化Assistant
bot = Assistant(llm=llm_cfg,system_message=system_instruction,function_list=tools,files=files)# Step 4: Run the agent as a chatbot.
messages = []  # This stores the chat history.
while True:# For example, enter the query "draw a dog and rotate it 90 degrees".query = input('user query: ')# Append the user query to the chat history.messages.append({'role': 'user', 'content': query})response = []for response in bot.run(messages=messages):# Streaming output.print('bot response:')pprint.pprint(response, indent=2)# Append the bot responses to the chat history.messages.extend(response)

• 首先输入任务目标：draw a dog and rotate it 90 degrees
  

• 绘制的狗子图片：
  

• 结果输出：
  

• Agent处理后的狗子图片展示：

二、基于Qwen-Agent的案例分析

2-0、环境安装

# 更新qwen_agent  以及 modelscope-studio
pip install --upgrade qwen_agent  
pip install --upgrade modelscope-studio

2-1、图像理解&文本生成Agent

概述：该Agent首先可以将图片转换为文字描述，之后根据文字描述转化为一个小故事。

创建了两个Assistant实例image_agent 和 writing_agent：

• image_agent 用于图像理解，使用了一个特定的视觉语言模型 qwen-vl-max（通义千问的多模态大模型）
• writing_agent 则负责根据图像生成的描述写文章。

"""Customize an agent to implement visual storytelling"""
import copy
from typing import Dict, Iterator, List, Optional, Unionfrom qwen_agent import Agent
from qwen_agent.agents import Assistant
from qwen_agent.gui import WebUI
from qwen_agent.llm import BaseChatModel
from qwen_agent.llm.schema import ContentItem, Message
from qwen_agent.tools import BaseToolclass VisualStorytelling(Agent):"""Customize an agent for writing story from pictures"""# 接收function_list和LLM参数，这里指的是文字生成所使用的模型# def __init__(self,function_list: Optional[List[Union[str, Dict, BaseTool]]] = None,llm: Optional[Union[Dict, BaseChatModel]] = None):# 调用父类的构造函数，传递语言模型super().__init__(llm=llm)# Nest one vl assistant for image understandingself.image_agent = Assistant(llm={'model': 'qwen-vl-max'})# Nest one assistant for article writingself.writing_agent = Assistant(llm=self.llm,function_list=function_list,system_message='你扮演一个想象力丰富的学生，你需要先理解图片内容，根据描述图片信息以后，' +'参考知识库中教你的写作技巧，发挥你的想象力，写一篇800字的记叙文',files=['https://www.jianshu.com/p/cdf82ff33ef8'])# Agent执行的核心方法，定义了处理消息的工作流程def _run(self, messages: List[Message], lang: str = 'zh', **kwargs) -> Iterator[List[Message]]:"""Define the workflow"""# assert isinstance(messages[-1]['content'], list)# 检查输入消息是否包含图像assert any([item.image for item in messages[-1]['content']]), 'This agent requires input of images'# image_agent 首先处理图像，生成对图像内容的详细描述。# 然后，writing_agent 使用这些描述来编写一个根据图像内容的记叙文。# Image understandingnew_messages = copy.deepcopy(messages)new_messages[-1]['content'].append(ContentItem(text='请详细描述这张图片的所有细节内容'))response = []for rsp in self.image_agent.run(new_messages):yield response + rspresponse.extend(rsp)new_messages.extend(rsp)# Writing articlenew_messages.append(Message('user', '开始根据以上图片内容编写你的记叙文吧！'))for rsp in self.writing_agent.run(new_messages, lang=lang, **kwargs):yield response + rspdef test(query: Optional[str] = '看图说话',image: str = 'https://img01.sc115.com/uploads3/sc/vector/201809/51413-20180914205509.jpg'):# define a writer agentbot = VisualStorytelling(llm={'model': 'qwen-max'})# Chatmessages = [Message('user', [ContentItem(image=image)])]if query:messages[-1]['content'].append(ContentItem(text=query))for response in bot.run(messages):print('bot response:', response)def app_tui():# Define a writer agentbot = VisualStorytelling(llm={'model': 'qwen-max'})# Chatmessages = []while True:query = input('user question: ')# image example: https://img01.sc115.com/uploads3/sc/vector/201809/51413-20180914205509.jpgimage = input('image url: ').strip()if not image:print('image cannot be empty！')continuemessages.append(Message('user', [ContentItem(image=image)]))if query:messages[-1]['content'].append(ContentItem(text=query))response = []for response in bot.run(messages):print('bot response:', response)messages.extend(response)def app_gui():bot = VisualStorytelling(llm={'model': 'qwen-max'})WebUI(bot).run()if __name__ == '__main__':# test()# app_tui()app_gui()

输出：（随便找的一张新闻截图）前半段为图片描述，后半段为故事生成，Perfect！

只要换一换提示词，就可以成为穿搭描述+穿搭建议的Agent啦！

2-2、 基于ReAct范式的数据分析Agent

ReAct： 上传文件，指定模型，基于ReAct范式，与大模型交互进行文件的分析。

模型服务初始化：

• 指定模型名称
• 指定api_key
• 指定tools工具列表
• 使用ReAct范式来定义bot，指定llm配置、名称、描述以及调用工具列表。

"""A data analysis example implemented by assistant"""
import os
from pprint import pprint
from typing import Optionalfrom qwen_agent.agents import ReActChat
from qwen_agent.gui import WebUIROOT_RESOURCE = os.path.join(os.path.dirname(__file__), 'resource')def init_agent_service():llm_cfg = {# 'model': 'Qwen/Qwen1.5-72B-Chat',# 'model_server': 'https://api.together.xyz',# 'api_key': os.getenv('TOGETHER_API_KEY'),'model': 'qwen-max','model_server': 'dashscope','api_key': os.getenv('DASHSCOPE_API_KEY'),}tools = ['code_interpreter']bot = ReActChat(llm=llm_cfg,name='code interpreter',description='This agent can run code to solve the problem',function_list=tools)return botdef test(query: str = 'pd.head the file first and then help me draw a line chart to show the changes in stock prices',file: Optional[str] = os.path.join(ROOT_RESOURCE, 'stock_prices.csv')):# Define the agentbot = init_agent_service()# Chatmessages = []if not file:messages.append({'role': 'user', 'content': query})else:messages.append({'role': 'user', 'content': [{'text': query}, {'file': file}]})for response in bot.run(messages):pprint(response, indent=2)def app_tui():# Define the agentbot = init_agent_service()# Chatmessages = []while True:# Query example: pd.head the file first and then help me draw a line chart to show the changes in stock pricesquery = input('user question: ')# File example: resource/stock_prices.csvfile = input('file url (press enter if no file): ').strip()if not query:print('user question cannot be empty！')continueif not file:messages.append({'role': 'user', 'content': query})else:messages.append({'role': 'user', 'content': [{'text': query}, {'file': file}]})response = []for response in bot.run(messages):print('bot response:', response)messages.extend(response)def app_gui():bot = init_agent_service()chatbot_config = {'prompt.suggestions': [{'text': 'pd.head the file first and then help me draw a line chart to show the changes in stock prices','files': [os.path.join(ROOT_RESOURCE, 'stock_prices.csv')]}, 'Draw a line graph y=x^2']}WebUI(bot, chatbot_config=chatbot_config).run()if __name__ == '__main__':# test()# app_tui()app_gui()

2-3、 多模态助手

概述： 多智能体合作实例——多模态助手。

初始化智能助手服务：init_agent_service() 函数：

• 配置语言模型和智能助手的类型。
• 定义了两种类型的助手：bot_vl（多模态助手，能够理解图像内容）和 bot_tool（工具助手，提供画图和代码执行工具）。
• 定义了一个 Router 类的实例，它充当路由器和文本智能助手，管理并分发任务到 bot_vl 和 bot_tool。

功能分析

• 根据不同的输入类型（文本、图像和文件）调用相应的智能助手处理。这样的设计使得系统在处理各种查询时更加灵活和强大。
• 例如，对于包含图像的查询，bot_vl 可能更适合处理；而纯文本编程问题则可能由 bot_tool 处理。

Router 源码的详细介绍见附录2：Router 源码

"""A multi-agent cooperation example implemented by router and assistant"""import os
from typing import Optionalfrom qwen_agent.agents import Assistant, ReActChat, Router
from qwen_agent.gui import WebUIROOT_RESOURCE = os.path.join(os.path.dirname(__file__), 'resource')def init_agent_service():# settingsllm_cfg = {'model': 'qwen-max'}llm_cfg_vl = {'model': 'qwen-vl-max'}tools = ['image_gen', 'code_interpreter']# Define a vl agentbot_vl = Assistant(llm=llm_cfg_vl, name='多模态助手', description='可以理解图像内容。')# Define a tool agentbot_tool = ReActChat(llm=llm_cfg,name='工具助手',description='可以使用画图工具和运行代码来解决问题',function_list=tools,)# Define a router (simultaneously serving as a text agent)bot = Router(llm=llm_cfg,agents=[bot_vl, bot_tool],)return botdef test(query: str = 'hello',image: str = 'https://img01.sc115.com/uploads/sc/jpgs/1505/apic11540_sc115.com.jpg',file: Optional[str] = os.path.join(ROOT_RESOURCE, 'poem.pdf'),
):# Define the agentbot = init_agent_service()# Chatmessages = []if not image and not file:messages.append({'role': 'user', 'content': query})else:messages.append({'role': 'user', 'content': [{'text': query}]})if image:messages[-1]['content'].append({'image': image})if file:messages[-1]['content'].append({'file': file})for response in bot.run(messages):print('bot response:', response)def app_tui():# Define the agentbot = init_agent_service()# Chatmessages = []while True:query = input('user question: ')# Image example: https://img01.sc115.com/uploads/sc/jpgs/1505/apic11540_sc115.com.jpgimage = input('image url (press enter if no image): ')# File example: resource/poem.pdffile = input('file url (press enter if no file): ').strip()if not query:print('user question cannot be empty！')continueif not image and not file:messages.append({'role': 'user', 'content': query})else:messages.append({'role': 'user', 'content': [{'text': query}]})if image:messages[-1]['content'].append({'image': image})if file:messages[-1]['content'].append({'file': file})response = []for response in bot.run(messages):print('bot response:', response)messages.extend(response)def app_gui():bot = init_agent_service()chatbot_config = {'verbose': True,}WebUI(bot, chatbot_config=chatbot_config).run()if __name__ == '__main__':# test()# app_tui()app_gui()

输出 （以图片+文档+文字作为输入）：

附录

1、agent源码

概述：定义Agent基类以及其实现和使用方法。

（1）init：初始化实例

• function_list: 可选参数，接收一个包含工具名称、配置字典或工具对象的列表。这些工具用于在Agent内执行各种任务。
• llm: 可选参数，可以是字典（指定LLM的配置）或已实例化的LLM模型对象。如果是字典，则使用 get_chat_model 方法将其转换成模型实例。
• system_message: 定义在LLM对话中使用的系统默认消息。
• name 和 description: 分别代表代理的名称和描述，有助于在多Agent环境中识别和描述Agent的用途。

（2）方法 run：run: 这个方法接收一系列消息，并调用 _run 方法（抽象方法，需要在子类中实现）来生成响应。

• 首先对输入消息进行深拷贝，并确定返回消息的类型（字典还是消息对象）。
• 检查输入消息的语言并调整语言参数，以确保正确的语言环境。
• 在生成响应时，将每个消息的 name 属性设置为代理的名称（如果存在）。

（3）抽象方法 _run：_run: 是一个抽象方法，要求所有继承自 Agent 的子类必须实现此方法来定义如何处理消息和生成响应。

（4）方法 _call_llm：_call_llm: 这个方法用于调用语言学习模型来处理消息。

• 在调用LLM之前，会将系统消息作为首条消息插入，或者将其添加到第一条消息的内容中。
• 使用 merge_generate_cfgs 方法来合并生成配置，以调整LLM的响应。

（5）方法 _call_tool：_call_tool: 用于调用特定的工具来处理特定的任务。

• 检查工具名称是否已注册，若未注册，则返回错误。
• 尝试调用工具并捕获任何异常，以便记录和处理错误。

（6）方法 _init_tool：_init_tool: 初始化和注册传入的工具。

• 检查工具是否已在工具注册表（TOOL_REGISTRY）中，如果没有，则抛出异常。
• 如果工具已经存在于 function_map 中，则发出警告并使用最新的工具实例替换旧的。

（7）方法 _detect_tool：_detect_tool: 用于检测消息是否包含工具调用的请求。

• 解析消息中的函数调用信息，确定是否需要执行工具调用，并提取工具名称和参数。

完整代码如下：

import copy
import json
import traceback
from abc import ABC, abstractmethod
from typing import Dict, Iterator, List, Optional, Tuple, Unionfrom qwen_agent.llm import get_chat_model
from qwen_agent.llm.base import BaseChatModel
from qwen_agent.llm.schema import CONTENT, DEFAULT_SYSTEM_MESSAGE, ROLE, SYSTEM, ContentItem, Message
from qwen_agent.log import logger
from qwen_agent.tools import TOOL_REGISTRY, BaseTool
from qwen_agent.utils.utils import has_chinese_messages, merge_generate_cfgsclass Agent(ABC):"""A base class for Agent.An agent can receive messages and provide response by LLM or Tools.Different agents have distinct workflows for processing messages and generating responses in the `_run` method."""def __init__(self,function_list: Optional[List[Union[str, Dict, BaseTool]]] = None,llm: Optional[Union[Dict, BaseChatModel]] = None,system_message: Optional[str] = DEFAULT_SYSTEM_MESSAGE,name: Optional[str] = None,description: Optional[str] = None,**kwargs):"""Initialization the agent.Args:function_list: One list of tool name, tool configuration or Tool object,such as 'code_interpreter', {'name': 'code_interpreter', 'timeout': 10}, or CodeInterpreter().llm: The LLM model configuration or LLM model object.Set the configuration as {'model': '', 'api_key': '', 'model_server': ''}.system_message: The specified system message for LLM chat.name: The name of this agent.description: The description of this agent, which will be used for multi_agent."""if isinstance(llm, dict):self.llm = get_chat_model(llm)else:self.llm = llmself.extra_generate_cfg: dict = {}self.function_map = {}if function_list:for tool in function_list:self._init_tool(tool)self.system_message = system_messageself.name = nameself.description = descriptiondef run(self, messages: List[Union[Dict, Message]],**kwargs) -> Union[Iterator[List[Message]], Iterator[List[Dict]]]:"""Return one response generator based on the received messages.This method performs a uniform type conversion for the inputted messages,and calls the _run method to generate a reply.Args:messages: A list of messages.Yields:The response generator."""messages = copy.deepcopy(messages)_return_message_type = 'dict'new_messages = []# Only return dict when all input messages are dictif not messages:_return_message_type = 'message'for msg in messages:if isinstance(msg, dict):new_messages.append(Message(**msg))else:new_messages.append(msg)_return_message_type = 'message'if 'lang' not in kwargs:if has_chinese_messages(new_messages):kwargs['lang'] = 'zh'else:kwargs['lang'] = 'en'for rsp in self._run(messages=new_messages, **kwargs):for i in range(len(rsp)):if not rsp[i].name and self.name:rsp[i].name = self.nameif _return_message_type == 'message':yield [Message(**x) if isinstance(x, dict) else x for x in rsp]else:yield [x.model_dump() if not isinstance(x, dict) else x for x in rsp]@abstractmethoddef _run(self, messages: List[Message], lang: str = 'en', **kwargs) -> Iterator[List[Message]]:"""Return one response generator based on the received messages.The workflow for an agent to generate a reply.Each agent subclass needs to implement this method.Args:messages: A list of messages.lang: Language, which will be used to select the language of the promptduring the agent's execution process.Yields:The response generator."""raise NotImplementedErrordef _call_llm(self,messages: List[Message],functions: Optional[List[Dict]] = None,stream: bool = True,extra_generate_cfg: Optional[dict] = None,) -> Iterator[List[Message]]:"""The interface of calling LLM for the agent.We prepend the system_message of this agent to the messages, and call LLM.Args:messages: A list of messages.functions: The list of functions provided to LLM.stream: LLM streaming output or non-streaming output.For consistency, we default to using streaming output across all agents.Yields:The response generator of LLM."""messages = copy.deepcopy(messages)if messages[0][ROLE] != SYSTEM:messages.insert(0, Message(role=SYSTEM, content=self.system_message))elif isinstance(messages[0][CONTENT], str):messages[0][CONTENT] = self.system_message + messages[0][CONTENT]else:assert isinstance(messages[0][CONTENT], list)messages[0][CONTENT] = [ContentItem(text=self.system_message)] + messages[0][CONTENT]return self.llm.chat(messages=messages,functions=functions,stream=stream,extra_generate_cfg=merge_generate_cfgs(base_generate_cfg=self.extra_generate_cfg,new_generate_cfg=extra_generate_cfg,))def _call_tool(self, tool_name: str, tool_args: Union[str, dict] = '{}', **kwargs) -> str:"""The interface of calling tools for the agent.Args:tool_name: The name of one tool.tool_args: Model generated or user given tool parameters.Returns:The output of tools."""if tool_name not in self.function_map:return f'Tool {tool_name} does not exists.'tool = self.function_map[tool_name]try:tool_result = tool.call(tool_args, **kwargs)except Exception as ex:exception_type = type(ex).__name__exception_message = str(ex)traceback_info = ''.join(traceback.format_tb(ex.__traceback__))error_message = f'An error occurred when calling tool `{tool_name}`:\n' \f'{exception_type}: {exception_message}\n' \f'Traceback:\n{traceback_info}'logger.warning(error_message)return error_messageif isinstance(tool_result, str):return tool_resultelse:return json.dumps(tool_result, ensure_ascii=False, indent=4)def _init_tool(self, tool: Union[str, Dict, BaseTool]):if isinstance(tool, BaseTool):tool_name = tool.nameif tool_name in self.function_map:logger.warning(f'Repeatedly adding tool {tool_name}, will use the newest tool in function list')self.function_map[tool_name] = toolelse:if isinstance(tool, dict):tool_name = tool['name']tool_cfg = toolelse:tool_name = tooltool_cfg = Noneif tool_name not in TOOL_REGISTRY:raise ValueError(f'Tool {tool_name} is not registered.')if tool_name in self.function_map:logger.warning(f'Repeatedly adding tool {tool_name}, will use the newest tool in function list')self.function_map[tool_name] = TOOL_REGISTRY[tool_name](tool_cfg)def _detect_tool(self, message: Message) -> Tuple[bool, str, str, str]:"""A built-in tool call detection for func_call format message.Args:message: one message generated by LLM.Returns:Need to call tool or not, tool name, tool args, text replies."""func_name = Nonefunc_args = Noneif message.function_call:func_call = message.function_callfunc_name = func_call.namefunc_args = func_call.argumentstext = message.contentif not text:text = ''return (func_name is not None), func_name, func_args, text

2、router源码

实现了一个高级的路由器功能，用于管理和协调多个智能助手代理（agents），以处理复杂的用户请求。这是通过继承和扩展了一个假想的 qwen_agent 库来完成的，其中包括多个模块和类，专门为建立智能对话系统而设计。下面我将详细解释这段代码的关键部分及其功能。

类定义：Router

Router 类继承自 Assistant 和 MultiAgentHub，旨在作为多个代理的中心节点，处理消息并根据需要将任务委托给其他代理。

构造函数 (init) 参数：

• function_list：可选，定义路由器可以执行的功能列表。
• llm：可选，定义了语言模型的配置或实例。
• files：可选，定义了与路由器相关的文件列表。
• name：可选，路由器的名称。
• description：可选，路由器的描述。
• agents：可选，定义了一组作为路由器部分的智能助手。
• rag_cfg：可选，定义了其他生成配置。

功能：

• 初始化路由器实例，同时设置系统消息，该消息是一个字符串模板，向用户解释可用的助手及其功能，但要求用户交互时不要向用户展示这些指令。
• 根据提供的助手列表，生成帮助描述和助手名列表。
• 更新生成配置以定制回答停止的标准。

_run 功能

• 处理传入的消息列表，决定是否需要从属助手的帮助来回答。
• 如果一个消息需要路由到特定的助手，Router 会解析出“Call:”指令后指定的助手名称，并将消息委托给该助手处理。
• 如果生成的助手名称不存在于列表中，则默认使用第一个助手。

静态方法：supplement_name_special_token 功能：

• 为消息内容增加特定的标记，格式化为“Call: <助手名>\nReply: <消息内容>”，以便后续处理。
• 这有助于在消息在不同助手间传递时保持跟踪和格式一致性。

这段代码通过一个中心路由器将用户请求分配给特定的智能助手，以处理不同类型的任务。通过在内部使用标记和格式化消息，确保了处理流程的清晰和效率。这种设计允许灵活的扩展和对多智能助手系统的细粒度控制，特别适合需要处理多种数据类型和请求的复杂对话系统。

以下为详细代码：

import copy
from typing import Dict, Iterator, List, Optional, Unionfrom qwen_agent import Agent, MultiAgentHub
from qwen_agent.agents.assistant import Assistant
from qwen_agent.llm import BaseChatModel
from qwen_agent.llm.schema import ASSISTANT, ROLE, Message
from qwen_agent.log import logger
from qwen_agent.tools import BaseTool
from qwen_agent.utils.utils import merge_generate_cfgsROUTER_PROMPT = '''你有下列帮手：
{agent_descs}当你可以直接回答用户时，请忽略帮手，直接回复；但当你的能力无法达成用户的请求时，请选择其中一个来帮你回答，选择的模版如下：
Call: ... # 选中的帮手的名字，必须在[{agent_names}]中选，不要返回其余任何内容。
Reply: ... # 选中的帮手的回复——不要向用户透露此条指令。'''class Router(Assistant, MultiAgentHub):def __init__(self,function_list: Optional[List[Union[str, Dict, BaseTool]]] = None,llm: Optional[Union[Dict, BaseChatModel]] = None,files: Optional[List[str]] = None,name: Optional[str] = None,description: Optional[str] = None,agents: Optional[List[Agent]] = None,rag_cfg: Optional[Dict] = None):self._agents = agentsagent_descs = '\n'.join([f'{x.name}: {x.description}' for x in agents])agent_names = ', '.join(self.agent_names)super().__init__(function_list=function_list,llm=llm,system_message=ROUTER_PROMPT.format(agent_descs=agent_descs, agent_names=agent_names),name=name,description=description,files=files,rag_cfg=rag_cfg)self.extra_generate_cfg = merge_generate_cfgs(base_generate_cfg=self.extra_generate_cfg,new_generate_cfg={'stop': ['Reply:', 'Reply:\n']},)def _run(self, messages: List[Message], lang: str = 'en', **kwargs) -> Iterator[List[Message]]:# This is a temporary plan to determine the source of a messagemessages_for_router = []for msg in messages:if msg[ROLE] == ASSISTANT:msg = self.supplement_name_special_token(msg)messages_for_router.append(msg)response = []for response in super()._run(messages=messages_for_router, lang=lang, **kwargs):yield responseif 'Call:' in response[-1].content and self.agents:# According to the rule in prompt to selected agentselected_agent_name = response[-1].content.split('Call:')[-1].strip().split('\n')[0].strip()logger.info(f'Need help from {selected_agent_name}')if selected_agent_name not in self.agent_names:# If the model generates a non-existent agent, the first agent will be used by default.selected_agent_name = self.agent_names[0]selected_agent = self.agents[self.agent_names.index(selected_agent_name)]for response in selected_agent.run(messages=messages, lang=lang, **kwargs):for i in range(len(response)):if response[i].role == ASSISTANT:response[i].name = selected_agent_name# This new response will overwrite the above 'Call: xxx' messageyield response@staticmethoddef supplement_name_special_token(message: Message) -> Message:message = copy.deepcopy(message)if not message.name:return messageif isinstance(message['content'], str):message['content'] = 'Call: ' + message['name'] + '\nReply:' + message['content']return messageassert isinstance(message['content'], list)for i, item in enumerate(message['content']):for k, v in item.model_dump().items():if k == 'text':message['content'][i][k] = 'Call: ' + message['name'] + '\nReply:' + message['content'][i][k]breakreturn message

参考文章：
Qwen-Agent ： GitHub官网.
Qwen-Agent 文档

---

总结

会调用工具的Agent太炫酷啦。🐏