Source code for camel.toolkits.function_tool
# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
import ast
import inspect
import logging
import textwrap
import warnings
from inspect import Parameter, getsource, signature
from typing import Any, Callable, Dict, Mapping, Optional, Tuple, Type
from docstring_parser import parse
from jsonschema.exceptions import SchemaError
from jsonschema.validators import Draft202012Validator as JSONValidator
from pydantic import BaseModel, create_model
from pydantic.fields import FieldInfo
from camel.models import BaseModelBackend, ModelFactory
from camel.types import ModelPlatformType, ModelType
from camel.utils import get_pydantic_object_schema, to_pascal
logger = logging.getLogger(__name__)
def _remove_a_key(d: Dict, remove_key: Any) -> None:
r"""Remove a key from a dictionary recursively."""
if isinstance(d, dict):
for key in list(d.keys()):
if key == remove_key:
del d[key]
else:
_remove_a_key(d[key], remove_key)
def _remove_title_recursively(data, parent_key=None):
r"""Recursively removes the 'title' key from all levels of a nested
dictionary, except when 'title' is an argument name in the schema.
"""
if isinstance(data, dict):
# Only remove 'title' if it's not an argument name
if parent_key not in [
"properties",
"$defs",
"items",
"allOf",
"oneOf",
"anyOf",
]:
data.pop("title", None)
# Recursively process each key-value pair
for key, value in data.items():
_remove_title_recursively(value, parent_key=key)
elif isinstance(data, list):
# Recursively process each element in the list
for item in data:
_remove_title_recursively(item, parent_key=parent_key)
[docs]
def get_openai_function_schema(func: Callable) -> Dict[str, Any]:
r"""Generates a schema dict for an OpenAI function based on its signature.
This function is deprecated and will be replaced by
:obj:`get_openai_tool_schema()` in future versions. It parses the
function's parameters and docstring to construct a JSON schema-like
dictionary.
Args:
func (Callable): The OpenAI function to generate the schema for.
Returns:
Dict[str, Any]: A dictionary representing the JSON schema of the
function, including its name, description, and parameter
specifications.
"""
openai_function_schema = get_openai_tool_schema(func)["function"]
return openai_function_schema
[docs]
def get_openai_tool_schema(func: Callable) -> Dict[str, Any]:
r"""Generates an OpenAI JSON schema from a given Python function.
This function creates a schema compatible with OpenAI's API specifications,
based on the provided Python function. It processes the function's
parameters, types, and docstrings, and constructs a schema accordingly.
Note:
- Each parameter in `func` must have a type annotation; otherwise, it's
treated as 'Any'.
- Variable arguments (*args) and keyword arguments (**kwargs) are not
supported and will be ignored.
- A functional description including a brief and detailed explanation
should be provided in the docstring of `func`.
- All parameters of `func` must be described in its docstring.
- Supported docstring styles: ReST, Google, Numpydoc, and Epydoc.
Args:
func (Callable): The Python function to be converted into an OpenAI
JSON schema.
Returns:
Dict[str, Any]: A dictionary representing the OpenAI JSON schema of
the provided function.
See Also:
`OpenAI API Reference
<https://platform.openai.com/docs/api-reference/assistants/object>`_
"""
params: Mapping[str, Parameter] = signature(func).parameters
fields: Dict[str, Tuple[type, FieldInfo]] = {}
for param_name, p in params.items():
param_type = p.annotation
param_default = p.default
param_kind = p.kind
param_annotation = p.annotation
# Variable parameters are not supported
if (
param_kind == Parameter.VAR_POSITIONAL
or param_kind == Parameter.VAR_KEYWORD
):
continue
# If the parameter type is not specified, it defaults to typing.Any
if param_annotation is Parameter.empty:
param_type = Any
# Check if the parameter has a default value
if param_default is Parameter.empty:
fields[param_name] = (param_type, FieldInfo())
else:
fields[param_name] = (param_type, FieldInfo(default=param_default))
# Applying `create_model()` directly will result in a mypy error,
# create an alias to avoid this.
def _create_mol(name, field):
return create_model(name, **field)
model = _create_mol(to_pascal(func.__name__), fields)
parameters_dict = get_pydantic_object_schema(model)
# The `"title"` is generated by `model.model_json_schema()`
# but is useless for openai json schema, remove generated 'title' from
# parameters_dict
_remove_title_recursively(parameters_dict)
docstring = parse(func.__doc__ or "")
for param in docstring.params:
if (name := param.arg_name) in parameters_dict["properties"] and (
description := param.description
):
parameters_dict["properties"][name]["description"] = description
short_description = docstring.short_description or ""
long_description = docstring.long_description or ""
if long_description:
func_description = f"{short_description}\n{long_description}"
else:
func_description = short_description
# OpenAI client.beta.chat.completions.parse for structured output has
# additional requirements for the schema, refer:
# https://platform.openai.com/docs/guides/structured-outputs/some-type-specific-keywords-are-not-yet-supported#supported-schemas
parameters_dict["additionalProperties"] = False
openai_function_schema = {
"name": func.__name__,
"description": func_description,
"strict": True,
"parameters": parameters_dict,
}
openai_tool_schema = {
"type": "function",
"function": openai_function_schema,
}
openai_tool_schema = sanitize_and_enforce_required(openai_tool_schema)
return openai_tool_schema
[docs]
def sanitize_and_enforce_required(parameters_dict):
r"""Cleans and updates the function schema to conform with OpenAI's
requirements:
- Removes invalid 'default' fields from the parameters schema.
- Ensures all fields or function parameters are marked as required.
Args:
parameters_dict (dict): The dictionary representing the function
schema.
Returns:
dict: The updated dictionary with invalid defaults removed and all
fields set as required.
"""
# Check if 'function' and 'parameters' exist
if (
'function' in parameters_dict
and 'parameters' in parameters_dict['function']
):
# Access the 'parameters' section
parameters = parameters_dict['function']['parameters']
properties = parameters.get('properties', {})
# Remove 'default' key from each property
for field in properties.values():
field.pop('default', None)
# Mark all keys in 'properties' as required
parameters['required'] = list(properties.keys())
return parameters_dict
[docs]
def generate_docstring(
code: str,
model: Optional[BaseModelBackend] = None,
) -> str:
r"""Generates a docstring for a given function code using LLM.
This function leverages a language model to generate a
PEP 8/PEP 257-compliant docstring for a provided Python function.
If no model is supplied, a default gpt-4o-mini is used.
Args:
code (str): The source code of the function.
model (Optional[BaseModelBackend]): An optional language model backend
instance. If not provided, a default gpt-4o-mini is used.
Returns:
str: The generated docstring.
"""
from camel.agents import ChatAgent
# Create the docstring prompt
docstring_prompt = textwrap.dedent(
"""\
**Role**: Generate professional Python docstrings conforming to PEP 8/PEP 257.
**Requirements**:
- Use appropriate format: reST, Google, or NumPy, as needed.
- Include parameters, return values, and exceptions.
- Reference any existing docstring in the function and retain useful information.
**Input**: Python function.
**Output**: Docstring content (plain text, no code markers).
**Example:**
Input:
```python
def add(a: int, b: int) -> int:
return a + b
```
Output:
Adds two numbers.
Args:
a (int): The first number.
b (int): The second number.
Returns:
int: The sum of the two numbers.
**Task**: Generate a docstring for the function below.
""" # noqa: E501
)
# Initialize assistant with system message and model
assistant_sys_msg = "You are a helpful assistant."
docstring_assistant = ChatAgent(assistant_sys_msg, model=model)
# Create user message to prompt the assistant
user_msg = docstring_prompt + code
# Get the response containing the generated docstring
response = docstring_assistant.step(user_msg)
return response.msg.content
[docs]
class FunctionTool:
r"""An abstraction of a function that OpenAI chat models can call. See
https://platform.openai.com/docs/api-reference/chat/create.
By default, the tool schema will be parsed from the func, or you can
provide a user-defined tool schema to override.
Args:
func (Callable): The function to call. The tool schema is parsed from
the function signature and docstring by default.
openai_tool_schema (Optional[Dict[str, Any]], optional): A
user-defined OpenAI tool schema to override the default result.
(default: :obj:`None`)
synthesize_schema (Optional[bool], optional): Whether to enable the
use of a schema assistant model to automatically synthesize the
schema if validation fails or no valid schema is provided.
(default: :obj:`False`)
synthesize_schema_model (Optional[BaseModelBackend], optional): An
assistant model (e.g., an LLM model) used to synthesize the schema
if `synthesize_schema` is enabled and no valid schema is
provided. (default: :obj:`None`)
synthesize_schema_max_retries (int, optional): The maximum
number of attempts to retry schema synthesis using the schema
assistant model if the previous attempts fail. (default: 2)
synthesize_output (Optional[bool], optional): Flag for enabling
synthesis output mode, where output is synthesized based on the
function's execution. (default: :obj:`False`)
synthesize_output_model (Optional[BaseModelBackend], optional):
Model used for output synthesis in synthesis mode.
(default: :obj:`None`)
synthesize_output_format (Optional[Type[BaseModel]], optional): Format
for the response when synthesizing output. (default: :obj:`None`)
"""
def __init__(
self,
func: Callable,
openai_tool_schema: Optional[Dict[str, Any]] = None,
synthesize_schema: Optional[bool] = False,
synthesize_schema_model: Optional[BaseModelBackend] = None,
synthesize_schema_max_retries: int = 2,
synthesize_output: Optional[bool] = False,
synthesize_output_model: Optional[BaseModelBackend] = None,
synthesize_output_format: Optional[Type[BaseModel]] = None,
) -> None:
self.func = func
self.openai_tool_schema = openai_tool_schema or get_openai_tool_schema(
func
)
self.synthesize_output = synthesize_output
self.synthesize_output_model = synthesize_output_model
if synthesize_output and synthesize_output_model is None:
self.synthesize_output_model = ModelFactory.create(
model_platform=ModelPlatformType.DEFAULT,
model_type=ModelType.DEFAULT,
)
logger.warning(
"Warning: No synthesize_output_model provided. "
f"Use `{self.synthesize_output_model.model_type}` to "
"synthesize the output."
)
self.synthesize_output_format: Optional[type[BaseModel]] = None
return_annotation = inspect.signature(self.func).return_annotation
if synthesize_output_format is not None:
self.synthesize_output_format = synthesize_output_format
elif isinstance(return_annotation, type) and issubclass(
return_annotation, BaseModel
):
self.synthesize_output_format = return_annotation
self.synthesize_schema_model = synthesize_schema_model
if synthesize_schema:
if openai_tool_schema:
logger.warning("""The user-defined OpenAI tool schema will be
overridden by the schema assistant model.""")
if self.synthesize_schema_model is None:
self.synthesize_schema_model = ModelFactory.create(
model_platform=ModelPlatformType.DEFAULT,
model_type=ModelType.DEFAULT,
)
logger.warning(
"Warning: No synthesize_schema_model provided. "
f"Use `{self.synthesize_schema_model.model_type}` to "
"synthesize the schema."
)
schema = self.synthesize_openai_tool_schema(
synthesize_schema_max_retries
)
if schema:
self.openai_tool_schema = schema
else:
raise ValueError(
f"Failed to synthesize a valid schema for "
f"{self.func.__name__}."
)
def __call__(self, *args: Any, **kwargs: Any) -> Any:
if self.synthesize_output:
result = self.synthesize_execution_output(args, kwargs)
return result
else:
# Pass the extracted arguments to the indicated function
try:
result = self.func(*args, **kwargs)
return result
except Exception as e:
raise ValueError(
f"Execution of function {self.func.__name__} failed with "
f"arguments {args} and {kwargs}. "
f"Error: {e}"
)
[docs]
@staticmethod
def validate_openai_tool_schema(
openai_tool_schema: Dict[str, Any],
) -> None:
r"""Validates the OpenAI tool schema against
:obj:`ToolAssistantToolsFunction`.
This function checks if the provided :obj:`openai_tool_schema` adheres
to the specifications required by OpenAI's
:obj:`ToolAssistantToolsFunction`. It ensures that the function
description and parameters are correctly formatted according to JSON
Schema specifications.
Args:
openai_tool_schema (Dict[str, Any]): The OpenAI tool schema to
validate.
Raises:
ValidationError: If the schema does not comply with the
specifications.
SchemaError: If the parameters do not meet JSON Schema reference
specifications.
"""
# Check the type
if not openai_tool_schema["type"]:
raise ValueError("miss `type` in tool schema.")
# Check the function description, if no description then raise warming
if not openai_tool_schema["function"].get("description"):
warnings.warn(f"""Function description is missing for
{openai_tool_schema['function']['name']}. This may
affect the quality of tool calling.""")
# Validate whether parameters
# meet the JSON Schema reference specifications.
# See https://platform.openai.com/docs/guides/gpt/function-calling
# for examples, and the
# https://json-schema.org/understanding-json-schema/ for
# documentation about the format.
parameters = openai_tool_schema["function"]["parameters"]
try:
JSONValidator.check_schema(parameters)
except SchemaError as e:
raise e
# Check the parameter description, if no description then raise warming
properties: Dict[str, Any] = parameters["properties"]
for param_name in properties.keys():
param_dict = properties[param_name]
if "description" not in param_dict:
warnings.warn(f"""Parameter description is missing for
{param_dict}. This may affect the quality of tool
calling.""")
[docs]
def get_openai_tool_schema(self) -> Dict[str, Any]:
r"""Gets the OpenAI tool schema for this function.
This method returns the OpenAI tool schema associated with this
function, after validating it to ensure it meets OpenAI's
specifications.
Returns:
Dict[str, Any]: The OpenAI tool schema for this function.
"""
self.validate_openai_tool_schema(self.openai_tool_schema)
return self.openai_tool_schema
[docs]
def set_openai_tool_schema(self, schema: Dict[str, Any]) -> None:
r"""Sets the OpenAI tool schema for this function.
Allows setting a custom OpenAI tool schema for this function.
Args:
schema (Dict[str, Any]): The OpenAI tool schema to set.
"""
self.openai_tool_schema = schema
[docs]
def get_openai_function_schema(self) -> Dict[str, Any]:
r"""Gets the schema of the function from the OpenAI tool schema.
This method extracts and returns the function-specific part of the
OpenAI tool schema associated with this function.
Returns:
Dict[str, Any]: The schema of the function within the OpenAI tool
schema.
"""
self.validate_openai_tool_schema(self.openai_tool_schema)
return self.openai_tool_schema["function"]
[docs]
def set_openai_function_schema(
self,
openai_function_schema: Dict[str, Any],
) -> None:
r"""Sets the schema of the function within the OpenAI tool schema.
Args:
openai_function_schema (Dict[str, Any]): The function schema to
set within the OpenAI tool schema.
"""
self.openai_tool_schema["function"] = openai_function_schema
[docs]
def get_function_name(self) -> str:
r"""Gets the name of the function from the OpenAI tool schema.
Returns:
str: The name of the function.
"""
self.validate_openai_tool_schema(self.openai_tool_schema)
return self.openai_tool_schema["function"]["name"]
[docs]
def set_function_name(self, name: str) -> None:
r"""Sets the name of the function in the OpenAI tool schema.
Args:
name (str): The name of the function to set.
"""
self.openai_tool_schema["function"]["name"] = name
[docs]
def get_function_description(self) -> str:
r"""Gets the description of the function from the OpenAI tool
schema.
Returns:
str: The description of the function.
"""
self.validate_openai_tool_schema(self.openai_tool_schema)
return self.openai_tool_schema["function"]["description"]
[docs]
def set_function_description(self, description: str) -> None:
r"""Sets the description of the function in the OpenAI tool schema.
Args:
description (str): The description for the function.
"""
self.openai_tool_schema["function"]["description"] = description
[docs]
def get_paramter_description(self, param_name: str) -> str:
r"""Gets the description of a specific parameter from the function
schema.
Args:
param_name (str): The name of the parameter to get the
description.
Returns:
str: The description of the specified parameter.
"""
self.validate_openai_tool_schema(self.openai_tool_schema)
return self.openai_tool_schema["function"]["parameters"]["properties"][
param_name
]["description"]
[docs]
def set_paramter_description(
self,
param_name: str,
description: str,
) -> None:
r"""Sets the description for a specific parameter in the function
schema.
Args:
param_name (str): The name of the parameter to set the description
for.
description (str): The description for the parameter.
"""
self.openai_tool_schema["function"]["parameters"]["properties"][
param_name
]["description"] = description
[docs]
def get_parameter(self, param_name: str) -> Dict[str, Any]:
r"""Gets the schema for a specific parameter from the function schema.
Args:
param_name (str): The name of the parameter to get the schema.
Returns:
Dict[str, Any]: The schema of the specified parameter.
"""
self.validate_openai_tool_schema(self.openai_tool_schema)
return self.openai_tool_schema["function"]["parameters"]["properties"][
param_name
]
[docs]
def set_parameter(self, param_name: str, value: Dict[str, Any]):
r"""Sets the schema for a specific parameter in the function schema.
Args:
param_name (str): The name of the parameter to set the schema for.
value (Dict[str, Any]): The schema to set for the parameter.
"""
try:
JSONValidator.check_schema(value)
except SchemaError as e:
raise e
self.openai_tool_schema["function"]["parameters"]["properties"][
param_name
] = value
[docs]
def synthesize_openai_tool_schema(
self,
max_retries: Optional[int] = None,
) -> Dict[str, Any]:
r"""Synthesizes an OpenAI tool schema for the specified function.
This method uses a language model (LLM) to synthesize the OpenAI tool
schema for the specified function by first generating a docstring and
then creating a schema based on the function's source code. The
schema synthesis and validation process is retried up to
`max_retries` times in case of failure.
Args:
max_retries (Optional[int], optional): The maximum number of
retries for schema synthesis and validation if the process
fails. (default: :obj:`None`)
Returns:
Dict[str, Any]: The synthesis OpenAI tool schema for the function.
Raises:
ValueError: If schema synthesis or validation fails after the
maximum number of retries, a ValueError is raised, prompting
manual schema setting.
"""
code = getsource(self.func)
retries = 0
if max_retries is None:
max_retries = 0
# Retry loop to handle schema synthesis and validation
while retries <= max_retries:
try:
# Generate the docstring and the schema
docstring = generate_docstring(
code, self.synthesize_schema_model
)
self.func.__doc__ = docstring
schema = get_openai_tool_schema(self.func)
# Validate the schema
self.validate_openai_tool_schema(schema)
return schema
except Exception as e:
retries += 1
if retries == max_retries:
raise ValueError(
f"Failed to synthesize the OpenAI tool Schema after "
f"{max_retries} retries. "
f"Please set the OpenAI tool schema for "
f"function {self.func.__name__} manually."
) from e
logger.warning("Schema validation failed. Retrying...")
return {}
[docs]
def synthesize_execution_output(
self,
args: Optional[tuple[Any, ...]] = None,
kwargs: Optional[Dict[str, Any]] = None,
) -> Any:
r"""Synthesizes the output of the function based on the provided
positional arguments and keyword arguments.
Args:
args (Optional[tuple]): Positional arguments to pass to the
function during synthesis. (default: :obj:`None`)
kwargs (Optional[Dict[str, Any]]): Keyword arguments to pass to the
function during synthesis. (default: :obj:`None`)
Returns:
Any: Synthesized output from the function execution. If no
synthesis model is provided, a warning is logged.
"""
from camel.agents import ChatAgent
# Retrieve the function source code
function_string = inspect.getsource(self.func)
# Check and update docstring if necessary
if self.func.__doc__ is not None:
function_string = textwrap.dedent(function_string)
tree = ast.parse(function_string)
func_node = (
tree.body[0]
if isinstance(tree.body[0], ast.FunctionDef)
else None
)
if func_node:
existing_docstring = ast.get_docstring(func_node)
if existing_docstring != self.func.__doc__:
func_node.body[0] = ast.Expr(
value=ast.Constant(value=self.func.__doc__, kind=None)
)
function_string = ast.unparse(tree)
# Append the args and kwargs information to the function string
if args:
function_string += f"\nargs:\n{list(args)}"
if kwargs:
function_string += f"\nkwargs:\n{kwargs}"
# Define the assistant system message
assistant_sys_msg = textwrap.dedent(
'''\
**Role:** AI Assistant specialized in synthesizing tool execution outputs without actual execution.
**Capabilities:**
- Analyzes function to understand their purpose and expected outputs.
- Generates synthetic outputs based on the function logic.
- Ensures the synthesized output is contextually accurate and aligns with the function's intended behavior.
**Instructions:**
1. **Input:** Provide the function code, function docstring, args, and kwargs.
2. **Output:** Synthesize the expected output of the function based on the provided args and kwargs.
**Example:**
- **User Input:**
def sum(a, b, c=0):
"""Adds three numbers together."""
return a + b + c
- **Input Arguments:**
args: (1, 2)
kwargs: {"c": 3}
- **Output:**
6
**Note:**
- Just return the synthesized output of the function without any explanation.
- The output should be in plain text without any formatting.
''' # noqa: E501
)
# Initialize the synthesis agent
synthesis_agent = ChatAgent(
assistant_sys_msg,
model=self.synthesize_output_model,
)
# User message combining function string and additional context
user_msg = function_string
response = synthesis_agent.step(
user_msg,
response_format=self.synthesize_output_format,
)
return response.msg.content
@property
def parameters(self) -> Dict[str, Any]:
r"""Getter method for the property :obj:`parameters`.
Returns:
Dict[str, Any]: the dictionary containing information of
parameters of this function.
"""
self.validate_openai_tool_schema(self.openai_tool_schema)
return self.openai_tool_schema["function"]["parameters"]["properties"]
@parameters.setter
def parameters(self, value: Dict[str, Any]) -> None:
r"""Setter method for the property :obj:`parameters`. It will
firstly check if the input parameters schema is valid. If invalid,
the method will raise :obj:`jsonschema.exceptions.SchemaError`.
Args:
value (Dict[str, Any]): the new dictionary value for the
function's parameters.
"""
try:
JSONValidator.check_schema(value)
except SchemaError as e:
raise e
self.openai_tool_schema["function"]["parameters"]["properties"] = value