8000 chore: update Anthropic tools to use new beta tools by willbakst · Pull Request #173 · Mirascope/mirascope · GitHub
[go: up one dir, main page]
More Web Proxy on the site http://driver.im/
Skip to content

chore: update Anthropic tools to use new beta tools #173

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Apr 5, 2024
Merged

chore: update Anthropic tools to use new beta tools #173

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
28cafd3
chore: update Anthropic tools to use new beta tools instead of deprec…
Apr 5, 2024
090c0a2
fix: lint issues
Apr 5, 2024
1683c55
chore: bump weave version
Apr 5, 2024
75864ec
fix: additional lint issues
Apr 5, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
81 changes: 8 additions & 73 deletions mirascope/anthropic/calls.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
"""A module for calling Anthropic's Claude API."""
import datetime
from textwrap import dedent
from typing import Any, AsyncGenerator, ClassVar, Generator, Optional, Type

from anthropic import Anthropic, AsyncAnthropic
Expand Down Expand Up @@ -61,10 +60,11 @@ def call(self, **kwargs: Any) -> AnthropicCallResponse:
client = Anthropic(api_key=self.api_key, base_url=self.base_url)
if self.call_params.wrapper is not None:
client = self.call_params.wrapper(client)
create = client.messages.create
if tool_types:
create = client.beta.tools.messages.create # type: ignore
if self.call_params.weave is not None:
create = self.call_params.weave(client.messages.create) # pragma: no cover
else:
create = client.messages.create
create = self.call_params.weave(create) # pragma: no cover
start_time = datetime.datetime.now().timestamp() * 1000
message = create(
messages=messages,
Expand Down Expand Up @@ -92,10 +92,11 @@ async def call_async(self, **kwargs: Any) -> AnthropicCallResponse:
client = AsyncAnthropic(api_key=self.api_key, base_url=self.base_url)
if self.call_params.wrapper_async is not None:
client = self.call_params.wrapper_async(client)
create = client.messages.create
if tool_types:
create = client.beta.tools.messages.create # type: ignore
if self.call_params.weave is not None:
create = self.call_params.weave(client.messages.create) # pragma: no cover
else:
create = client.messages.create
create = self.call_params.weave(create) # pragma: no cover
start_time = datetime.datetime.now().timestamp() * 1000
message = await create(
messages=messages,
Expand Down Expand Up @@ -167,72 +168,6 @@ def _setup_anthropic_kwargs(
system_message += kwargs.pop("system")
if messages[0]["role"] == "system":
system_message += messages.pop(0)["content"]
if tool_types:
tool_schemas = kwargs.pop("tools")
system_message += self._write_tools_system_message(tool_schemas)
kwargs["stop_sequences"] = ["</function_calls>"]
if system_message:
kwargs["system"] = system_message
return messages, kwargs, tool_types

def _write_tools_system_message(self, tool_schemas: list[str]) -> str:
"""Returns the Anthropic Tools System Message from their guide."""
return dedent(
"""
In this environment you have access to a set of tools you can use to answer the user's question.

You may call them like this:
<function_calls>
<invoke>
<tool_name>$TOOL_NAME</tool_name>
<parameters>
<$PARAMETER_NAME>$PARAMETER_VALUE</$PARAMETER_NAME>
...
</parameters>
</invoke>
...
</function_calls>

Make sure to include all parameters in the tool schema when requested.
If you want to call multiple tools, you should put all of the tools inside of the <function_calls> tag as multiple <invoke> elements.

To output nested structured data, encode it as valid XML with tags and values. For example:

List[int]:
<parameterName>
<item>1</item>
<item>2</item>
<item>3</item>
</parameterName>

List[object]:
<parameterName>
<item>
<objectName>
<objectValue>value</objectValue>
</objectName>
</item>
</parameterName>

Dictionary:
<parameterName>
<entry>
<key>key1</key>
<value>value1</value>
</entry>
<entry>
<key>key2</key>
<value>value2</value>
</entry>
</parameterName>

Remember, the above are just examples.
Make sure to properly nest by wrapping elements in lists with the <item> tag and dictionary elements with <entry> as necessary.
DO NOT FORGET THESE TAGS. Without these tags, we cannot properly parse the information you send.

Here are the tools available:
<tools>
{tool_schemas}
</tools>
""".format(tool_schemas=tool_schemas)
)
4 changes: 2 additions & 2 deletions mirascope/anthropic/extractors.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,9 @@


EXTRACT_SYSTEM_MESSAGE = """
OUTPUT ONLY FUNCTION CALLS WITHOUT ANY ADDITIONAL TEXT.
You must wrap all invoke calls in the <function_calls> tag.
OUTPUT ONLY TOOL CALLS.
ONLY EXTRACT ANSWERS THAT YOU ARE ABSOLUTELY 100% CERTAIN ARE CORRECT.
If asked to generate information, you may generate the tool call input.
""".strip()


Expand Down
190 changes: 19 additions & 171 deletions mirascope/anthropic/tools.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,10 @@
"""Classes for using tools with Anthropic's Claude API."""

from __future__ import annotations

import xml.etree.ElementTree as ET
from json import JSONDecodeError, loads
from textwrap import dedent
from typing import Any, Callable, Type, TypeVar, Union, cast
from typing import Callable, Type, TypeVar

from anthropic.types.beta.tools import ToolParam, ToolUseBlock
from pydantic import BaseModel

from ..base import BaseTool, BaseType
Expand All @@ -18,7 +17,7 @@
BaseTypeT = TypeVar("BaseTypeT", bound=BaseType)


class AnthropicTool(BaseTool[ET.Element]):
class AnthropicTool(BaseTool[ToolUseBlock]):
'''A base class for easy use of tools with the Anthropic Claude client.

`AnthropicTool` internally handles the logic that allows you to use tools with
Expand Down Expand Up @@ -64,87 +63,35 @@ class AnimalMatcher(AnthropicCall):
'''

@classmethod
def tool_schema(cls) -> str:
"""Constructs XML tool schema string for use with Anthropic's Claude API."""
json_schema = super().tool_schema()
tool_schema = (
dedent(
"""
<tool_description>
<tool_name>{name}</tool_name>
<description>
{description}
</description>
"""
)
.strip()
.format(name=cls.__name__, description=json_schema["description"])
def tool_schema(cls) -> ToolParam:
"""Constructs JSON tool schema for use with Anthropic's Claude API."""
schema = super().tool_schema()
return ToolParam(
input_schema=schema["parameters"],
name=schema["name"],
description=schema["description"],
)

tool_schema += "\n"
if "parameters" in json_schema:
tool_schema += _process_schema(json_schema["parameters"])
tool_schema += "</tool_description>"
return tool_schema

@classmethod
def from_tool_call(cls, tool_call: ET.Element) -> AnthropicTool:
def from_tool_call(cls, tool_call: ToolUseBlock) -> AnthropicTool:
"""Extracts an instance of the tool constructed from a tool call response.

Given the `<invoke>...</invoke>` block in a `Message` from an Anthropic call
response, this method parses out the XML defining the tool call and creates an
`AnthropicTool` instance from it.
Given the tool call contents in a `Message` from an Anthropic call response,
this method parses out the arguments of the tool call and creates an
`AnthropicTool` instance from them.

Args:
tool_call: The XML `str` from which to extract the tool.
tool_call: The list of `TextBlock` contents.

Returns:
An instance of the tool constructed from the tool call.

Raises:
ValidationError: if the tool call doesn't match the tool schema.
"""

def _parse_xml_element(element: ET.Element) -> Union[str, dict, list]:
"""Recursively parse an XML element into a Python data structure."""
children = list(element)
if not children:
if element.text:
text = element.text.strip().replace("\\n", "").replace("\\", "")
# Attempt to load JSON-like strings as Python data structures
try:
return loads(text)
except JSONDecodeError:
return text
else:
return ""

if (
len(children) > 1
and all(child.tag == children[0].tag for child in children)
) or (
len(children) == 1
and (children[0].tag == "entry" or children[0].tag == "item")
):
if children[0].tag == "entry":
dict_result = {}
for child in children:
key, value = child.find("key"), child.find("value")
if key is not None and value is not None:
dict_result[key.text] = _parse_xml_element(value)
return dict_result
else:
return [_parse_xml_element(child) for child in children]

result: dict[str, Any] = {}
for child in children:
child_value = _parse_xml_element(child)
result[child.tag] = child_value
return result

parameters = cast(dict[str, Any], _parse_xml_element(tool_call))
parameters["tool_call"] = tool_call
return cls.model_validate(parameters)
model_json = tool_call.input
model_json["tool_call"] = tool_call.model_dump() # type: ignore
return cls.model_validate(model_json)

@classmethod
def from_model(cls, model: Type[BaseModel]) -> Type[AnthropicTool]:
Expand All @@ -160,102 +107,3 @@ def from_fn(cls, fn: Callable) -> Type[AnthropicTool]:
def from_base_type(cls, base_type: Type[BaseTypeT]) -> Type[AnthropicTool]:
"""Constructs a `AnthropicTool` type from a `BaseType` type."""
return convert_base_type_to_tool(base_type, AnthropicTool)


def _process_schema(schema: dict[str, Any]) -> str:
schema_xml = ""
if "$defs" in schema:
schema_xml += "<definitions>\n"
for def_name, definition in schema["$defs"].items():
schema_xml += f"<definition name='{def_name}'>\n"
schema_xml += _process_property(definition)
schema_xml += "</definition>\n"
schema_xml += "</definitions>"
if "properties" in schema:
schema_xml += "<parameters>\n"
for prop, definition in schema["properties"].items():
schema_xml += "<parameter>\n"
schema_xml += f"<name>{prop}</name>\n"
schema_xml += _process_property(definition)
schema_xml += "</parameter>\n"
schema_xml += "</parameters>\n"
return schema_xml


# TODO: consider using ET subelements instead of string concat
def _process_property(definition: dict[str, Any]) -> str:
prop_xml = ""
prop_type = definition.get("type", "object")

if "$ref" in definition:
ref_name = definition["$ref"].split("/")[-1]
prop_xml += f"<reference name='{ref_name}'/>\n"
elif prop_type == "array":
prop_xml += "<type>list</type>\n"
items_def = None
if "items" in definition:
items_def = definition["items"]
elif "prefixItems" in definition:
items_def = definition["prefixItems"]
if items_def:
if isinstance(items_def, dict):
prop_xml += "<element_type>\n"
prop_xml += _process_property(items_def)
prop_xml += "</element_type>\n"
else:
for item in items_def:
prop_xml += "<element_type>\n"
prop_xml += _process_property(item)
prop_xml += "</element_type>\n"
if "maxItems" in definition:
prop_xml += f"<maxItems>{definition['maxItems']}</maxItems>\n"
if "minItems" in definition:
prop_xml += f"<minItems>{definition['minItems']}</minItems>\n"
if "uniqueItems" in definition:
prop_xml += (
f"<uniqueItems>{str(definition['uniqueItems']).lower()}</uniqueItems>\n"
)
elif "enum" in definition:
prop_xml += "<type>enum</type>\n"
prop_xml += "<values>\n"
for value in definition["enum"]:
prop_xml += f"<value>{value}</value>\n"
prop_xml += "</values>\n"
prop_xml += f"<value_type>{prop_type}</value_type>\n"
elif "anyOf" in definition:
prop_xml += "<type>union</type>\n"
prop_xml += "<options>\n"
for option in definition["anyOf"]:
prop_xml += "<option>\n"
prop_xml += _process_property(option)
prop_xml += "</option>\n"
prop_xml += "</options>\n"
elif "const" in definition:
prop_xml += "<type>literal</type>\n"
prop_xml += f"<value>{definition['const']}</value>\n"
elif "additionalProperties" in definition:
prop_xml += "<type>dictionary</type>\n"
prop_xml += "<entry>\n"
prop_xml += "<key>\n<type>string</type>\n</key>\n"
prop_xml += "<value>\n"
prop_xml += _process_property(definition["additionalProperties"])
prop_xml += "</value>\n"
prop_xml += "</entry>\n"
elif prop_type in ["string", "number", "integer", "boolean"]:
prop_xml += f"<type>{prop_type}</type>\n"
elif prop_type == "null":
prop_xml += "<type>null</type>\n"
else:
prop_xml += "<type>object</type>\n"
prop_xml += _process_schema(definition)

if "description" in definition:
prop_xml += f"<description>{definition['description']}</description>\n"
if "default" in definition:
default_value = definition["default"]
if default_value is None:
prop_xml += "<default>null</default>\n"
else:
prop_xml += f"<default>{default_value}</default>\n"

return prop_xml
Loading
0