Source code for agentscope.formatters._dashscope_formatter

# -*- coding: utf-8 -*-
"""The dashscope formatter class."""
import json
import os
from typing import Union

from loguru import logger

from ..formatters import FormatterBase
from ..message import Msg
from ..utils.common import _guess_type_by_extension


[docs] class DashScopeFormatter(FormatterBase): """The DashScope formatter, which is responsible for formatting messages, JSON schemas description of the tool functions.""" supported_model_regexes: list[str] = [ "qwen-.*", ] @staticmethod def _convert_url(url: str) -> list[dict]: """Convert the url to the format of DashScope API. Note for local files, a prefix "file://" will be added. Args: url (`Union[str, Sequence[str], None]`): A string of url of a list of urls to be converted. Returns: `List[dict]`: A list of dictionaries with key as the type of the url and value as the url. Only "image" and "audio" are supported. """ if isinstance(url, str): url_type = _guess_type_by_extension(url) if url_type in ["audio", "image"]: # Add prefix for local files if os.path.exists(url): url = "file://" + url return [{url_type: url}] else: # skip unsupported url logger.warning( f"Skip unsupported url ({url_type}), " f"expect image or audio.", ) return [] else: raise TypeError( f"Unsupported url type {type(url)}, " f"str or list expected.", )
[docs] @classmethod def format_chat(cls, *msgs: Union[Msg, list[Msg], None]) -> list[dict]: """Format the messages in chat scenario, where only one user and one assistant are involved. The current format function supports: - [x] image - [x] audio - [x] text - [x] tool_use """ input_msgs = cls.check_and_flat_messages(*msgs) formatted_msgs: list[dict] = [] for msg in input_msgs: content_blocks = [] tool_calls = [] for block in msg.get_content_blocks(): typ = block.get("type") if typ in ["text", "image", "audio"]: content_blocks.append( { typ: block.get("text", block.get("url")), }, ) elif typ == "tool_use": tool_calls.append( { "id": block.get("id"), "type": "function", "function": { "name": block.get("name"), "arguments": json.dumps( block.get("input", {}), ensure_ascii=False, ), }, }, ) elif typ == "tool_result": formatted_msgs.append( { "role": "tool", "tool_call_id": block.get("id"), "content": block.get("output"), "name": block.get("name"), }, ) else: logger.warning( f"Unsupported block type {typ} in the message, " f"skipped.", ) msg_dashscope = { "role": msg.role, "content": content_blocks or None, } if tool_calls: msg_dashscope["tool_calls"] = tool_calls if msg_dashscope["content"] or msg_dashscope.get("tool_calls"): formatted_msgs.append(msg_dashscope) return formatted_msgs
[docs] @classmethod def format_multi_agent( cls, *msgs: Union[Msg, list[Msg], None], ) -> list[dict]: """Format the messages in multi-agent scenario, where multiple agents are involved.""" input_msgs = cls.check_and_flat_messages(*msgs) messages = [] # record dialog history as a list of strings dialogue = [] image_or_audio_dicts = [] for i, msg in enumerate(input_msgs): if i == 0 and msg.role == "system" and msg.get_text_content(): # system prompt content = [{"text": msg.get_text_content()}] messages.append( { "role": msg.role, "content": content, }, ) else: # text message if msg.get_text_content(): dialogue.append( f"{msg.name}: {msg.get_text_content()}", ) # image and audio for block in msg.get_content_blocks(): if block.get("type") in ["image", "audio"]: image_or_audio_dicts.extend( cls._convert_url(str(block.get("url"))), ) dialogue_history = "\n".join(dialogue) user_content_template = "## Conversation History\n{dialogue_history}" messages.append( { "role": "user", "content": [ # Place the image or audio before the conversation history *image_or_audio_dicts, { "text": user_content_template.format( dialogue_history=dialogue_history, ), }, ], }, ) return messages
[docs] @classmethod def format_tools_json_schemas(cls, schemas: dict[str, dict]) -> list[dict]: """Format the JSON schemas of the tool functions to the format that DashScope API expects. This function will take the parsed JSON schema from `agentscope.service.ServiceToolkit` as input and return the formatted JSON schema. Note: An example of the input/output tool JSON schema ..code-block:: json { "bing_search": { "type": "function", "function": { "name": "bing_search", "description": "Search the web using Bing.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "The search query.", } }, "required": ["query"], } } } } Args: schemas (`dict[str, dict]`): The JSON schema of the tool functions. Returns: `list[dict]`: The formatted JSON schema. """ # The schemas from ServiceToolkit is exactly the same with the format # that DashScope API expects, so we just return the input schemas. assert isinstance( schemas, dict, ), f"Expect dict of schemas, got {type(schemas)}." for schema in schemas.values(): assert isinstance( schema, dict, ), f"Expect dict schema, got {type(schema)}." assert ( "type" in schema and "function" in schema ), f"Invalid schema: {schema}, expect keys 'type' and 'function'." assert ( schema["type"] == "function" ), f"Invalid schema type: {schema['type']}, expect 'function'." assert "name" in schema["function"], ( f"Invalid schema: {schema}, " f"expect key 'name' in 'function' field." ) return list(schemas.values())