Source code for agentscope.formatters._anthropic_formatter

# -*- coding: utf-8 -*-
"""The formatter for Anthropic models."""
from typing import Union

from loguru import logger

from ._formatter_base import FormatterBase
from ..message import Msg
from ..utils.common import _to_anthropic_image_url


[docs] class AnthropicFormatter(FormatterBase): """The formatter for Anthropic models.""" supported_model_regexes: list[str] = [ "claude-3-5.*", "claude-3-7.*", ]
[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. Args: msgs (`Union[Msg, list[Msg], None]`): The message(s) to be formatted. The `None` input will be ignored. """ msgs = cls.check_and_flat_messages(*msgs) formatted_msgs = [] for index, msg in enumerate(msgs): content = [] for block in msg.get_content_blocks(): if block.get("type") == "text": content.append( { "type": "text", "text": block.get("text"), }, ) elif block.get("type") == "image": content.append( { "type": "image", "source": _to_anthropic_image_url( str(block.get("url")), ), }, ) elif block.get("type") == "tool_use": content.append(dict(block)) elif block.get("type") == "tool_result": content.append( { "type": "tool_result", "tool_use_id": block.get("id"), "content": block.get("output"), }, ) else: logger.warning( f"Unsupported block type: {block.get('type')}", "skipped", ) # Claude only allow the first message to be system message if msg.role == "system" and index != 0: role = "user" else: role = msg.role formatted_msgs.append( { "role": role, "content": content, }, ) 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.""" # Parse all information into a list of messages input_msgs = cls.check_and_flat_messages(*msgs) if len(input_msgs) == 0: raise ValueError( "At least one message should be provided. An empty message " "list is not allowed.", ) # record dialog history as a list of strings dialogue = [] image_blocks = [] sys_prompt = None for i, msg in enumerate(input_msgs): if i == 0 and msg.role == "system": # if system prompt is available, place it at the beginning sys_prompt = msg.get_text_content() else: # Merge all messages into a conversation history prompt for block in msg.get_content_blocks(): typ = block.get("type") if typ == "text": dialogue.append( f"{msg.name}: {block.get('text')}", ) elif typ == "tool_use": dialogue.append( f"<tool_use>{block}</tool_use>", ) elif typ == "tool_result": dialogue.append( f"<tool_result>{block}</tool_result>", ) elif typ == "image": image_blocks.append( { "type": "image", "source": _to_anthropic_image_url( str(block.get("url")), ), }, ) content_components = [] # The conversation history is added to the user message if not empty if len(dialogue) > 0: content_components.extend(["## Conversation History"] + dialogue) messages = [ { "role": "user", "content": [ { "type": "text", "text": "\n".join(content_components), }, ] + image_blocks, }, ] # Add system prompt at the beginning if provided if sys_prompt is not None: messages = [{"role": "system", "content": sys_prompt}] + messages 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 Anthropic 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 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"], } } } } The formatted JSON schema will be like: ..code-block:: json [ { "name": "bing_search", "description": "Search the web using Bing.", "input_schema": { "type": "object", "properties": { "query": { "type": "string", "description": "The search query.", } }, "required": ["query"], } } ] Args: schemas (`dict[str, dict]`): The JSON schema of the tool functions, where the key is the function name, and the value is the schema. Returns: `list[dict]`: The formatted JSON schema. """ assert isinstance( schemas, dict, ), f"Expect a dict of schemas, got {type(schemas)}." formatted_schemas = [] for schema in schemas.values(): assert ( "function" in schema ), f"Invalid schema: {schema}, expect key 'function'." assert "name" in schema["function"], ( f"Invalid schema: {schema}, " "expect key 'name' in 'function' field." ) formatted_schemas.append( { "name": schema["function"]["name"], "description": schema["function"].get("description", ""), "input_schema": schema["function"].get("parameters", {}), }, ) return formatted_schemas