【通義千問—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