# ========= Copyright 2023-2026 @ 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-2026 @ CAMEL-AI.org. All Rights Reserved. ========= import ast import asyncio import functools import inspect import logging import textwrap import threading import warnings from concurrent.futures import ThreadPoolExecutor 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__) # Shared thread pool for running sync tools without blocking the event loop _SYNC_TOOL_EXECUTOR = ThreadPoolExecutor(max_workers=64) # Persistent event loop to avoid httpx connection pool issues _PERSISTENT_LOOP: Optional[asyncio.AbstractEventLoop] = None _PERSISTENT_LOOP_LOCK = threading.Lock() 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) 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 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 `_ """ 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 ): # OpenAI does not allow descriptions on properties that use $ref. # To avoid schema errors, we only add the description if "$ref" is # not present. prop = parameters_dict["properties"][name] if "$ref" not in prop: prop["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 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 are marked as required or have null type for optional fields. - Recursively adds additionalProperties: false to all nested objects. Args: parameters_dict (dict): The dictionary representing the function schema. Returns: dict: The updated dictionary with invalid defaults removed and all fields properly configured for strict mode. """ def _add_additional_properties_false(obj): r"""Recursively add additionalProperties: false to all objects.""" if isinstance(obj, dict): if ( obj.get("type") == "object" and "additionalProperties" not in obj ): obj["additionalProperties"] = False # Process nested structures for key, value in obj.items(): if key == "properties" and isinstance(value, dict): for prop_value in value.values(): _add_additional_properties_false(prop_value) elif key in [ "items", "allOf", "oneOf", "anyOf", ] and isinstance(value, (dict, list)): if isinstance(value, dict): _add_additional_properties_false(value) elif isinstance(value, list): for item in value: _add_additional_properties_false(item) elif key == "$defs" and isinstance(value, dict): for def_value in value.values(): _add_additional_properties_false(def_value) # 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', {}) # Track which fields should be required vs optional required_fields = [] # Process each property for field_name, field_schema in properties.items(): # Check if this field had a default value (making it optional) had_default = 'default' in field_schema # Remove 'default' key from field schema as required by OpenAI field_schema.pop('default', None) if had_default: # This field is optional - add null to its type current_type = field_schema.get('type') has_ref = '$ref' in field_schema has_any_of = 'anyOf' in field_schema if has_ref: # Fields with $ref shouldn't have additional type field # The $ref itself defines the type structure pass elif has_any_of: # Field already has anyOf any_of_types = field_schema['anyOf'] has_null_type = any( item.get('type') == 'null' for item in any_of_types ) if not has_null_type: # Add null type to anyOf field_schema['anyOf'].append({'type': 'null'}) # Remove conflicting type field if it exists if 'type' in field_schema: del field_schema['type'] elif current_type: if isinstance(current_type, str): # Single type - convert to array with null field_schema['type'] = [current_type, 'null'] elif ( isinstance(current_type, list) and 'null' not in current_type ): # Array of types - add null if not present field_schema['type'] = [*current_type, 'null'] else: # No type specified, add null type field_schema['type'] = ['null'] # Optional fields are still marked as required in strict mode # but with null type to indicate they can be omitted required_fields.append(field_name) else: # This field is required required_fields.append(field_name) # Set all fields as required (strict mode requirement) parameters['required'] = required_fields # Recursively add additionalProperties: false to all objects _add_additional_properties_false(parameters) return parameters_dict 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 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 # Call the function first try: result = self.func(*args, **kwargs) except Exception as e: parts = [] if args: parts.append(f"args={args}") if kwargs: parts.append(f"kwargs={kwargs}") args_str = ", ".join(parts) if parts else "no arguments" raise ValueError( f"Execution of function {self.func.__name__} failed with " f"{args_str}. Error: {e}" ) # Handle coroutine result (from async function or sync wrapper # returning coroutine) if inspect.iscoroutine(result): # Check if there's already a running event loop try: asyncio.get_running_loop() has_running_loop = True except RuntimeError: has_running_loop = False if has_running_loop: # Already in an async context warnings.warn( f"Async tool '{self.func.__name__}' is being called " f"synchronously within an async context. Consider using " f"'await tool.async_call()' or 'await agent.astep()' for " f"better performance.", RuntimeWarning, stacklevel=2, ) # Must run in separate thread to avoid blocking current loop future = _SYNC_TOOL_EXECUTOR.submit( self._run_async_in_persistent_loop, result ) return future.result() else: warnings.warn( f"Async tool '{self.func.__name__}' is being called " f"synchronously. Consider using 'await tool.async_call()' " f"or 'await agent.astep()' for better performance.", RuntimeWarning, stacklevel=2, ) return self._run_async_in_persistent_loop(result) return result @staticmethod def _run_async_in_persistent_loop(coro): r"""Run coroutine in persistent loop to preserve httpx connections.""" global _PERSISTENT_LOOP with _PERSISTENT_LOOP_LOCK: need_new_loop = ( _PERSISTENT_LOOP is None or _PERSISTENT_LOOP.is_closed() or not _PERSISTENT_LOOP.is_running() ) if need_new_loop: _PERSISTENT_LOOP = asyncio.new_event_loop() t = threading.Thread( target=_PERSISTENT_LOOP.run_forever, daemon=True ) t.start() while not _PERSISTENT_LOOP.is_running(): pass # Wait for loop to start future = asyncio.run_coroutine_threadsafe(coro, _PERSISTENT_LOOP) return future.result() async def async_call(self, *args: Any, **kwargs: Any) -> Any: if self.synthesize_output: result = self.synthesize_execution_output(args, kwargs) return result # Check if the function itself (not unwrapped) is a coroutine function if inspect.iscoroutinefunction(self.func): return await self.func(*args, **kwargs) # For sync functions (including sync wrappers around async functions), # run in executor to avoid blocking loop = asyncio.get_running_loop() result = await loop.run_in_executor( _SYNC_TOOL_EXECUTOR, functools.partial(self.func, *args, **kwargs), ) # If the sync wrapper returned a coroutine, await it if inspect.iscoroutine(result): return await result return result @property def is_async(self) -> bool: return inspect.iscoroutinefunction(inspect.unwrap(self.func)) @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 the " f"function '{openai_tool_schema['function']['name']}'. " f"The parameter definition is {param_dict}. " f"This may affect the quality of tool calling." ) 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 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 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"] 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 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"] 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 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"] 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 def get_parameter_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"] def set_parameter_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 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 ] 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 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 {} 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 def tool( func: Optional[Callable] = None, *, openai_tool_schema: Optional[Dict[str, Any]] = None, synthesize_schema: bool = False, synthesize_schema_model: Optional[BaseModelBackend] = None, synthesize_schema_max_retries: int = 2, synthesize_output: bool = False, synthesize_output_model: Optional[BaseModelBackend] = None, synthesize_output_format: Optional[Type[BaseModel]] = None, ): r"""A decorator that converts a Python function into a FunctionTool instance. This decorator can be used with or without parentheses: - @tool - without parentheses, uses default settings - @tool() - with parentheses, uses default settings - @tool(synthesize_output=True) - with custom settings Args: func (Optional[Callable], optional): The function to be decorated. This is automatically passed when using @tool without parentheses. (default: :obj:`None`) openai_tool_schema (Optional[Dict[str, Any]], optional): A user-defined OpenAI tool schema to override the default result. (default: :obj:`None`) synthesize_schema (bool, optional): Whether to enable schema synthesis. (default: :obj:`False`) synthesize_schema_model (Optional[BaseModelBackend], optional): Model to use for schema synthesis. (default: :obj:`None`) synthesize_schema_max_retries (int, optional): Maximum number of retries for schema synthesis. (default: :obj:`2`) synthesize_output (bool, optional): Whether to enable output synthesis. (default: :obj:`False`) synthesize_output_model (Optional[BaseModelBackend], optional): Model to use for output synthesis. (default: :obj:`None`) synthesize_output_format (Optional[Type[BaseModel]], optional): Format for synthesized output. (default: :obj:`None`) Returns: Callable[[Callable], FunctionTool]: A decorator function that converts the decorated function into a FunctionTool instance. Example: Using @tool without parentheses:: @tool def add(a: int, b: int = 0) -> int: '''Add two numbers.''' return a + b Using @tool() with parentheses:: @tool() def multiply(a: int, b: int) -> int: '''Multiply two numbers.''' return a * b """ def decorator(f: Callable) -> FunctionTool: r"""The actual decorator function. Args: f (Callable): The function to be converted into a FunctionTool. Returns: FunctionTool: The function wrapped as a FunctionTool instance. """ return FunctionTool( func=f, openai_tool_schema=openai_tool_schema, synthesize_schema=synthesize_schema, synthesize_schema_model=synthesize_schema_model, synthesize_schema_max_retries=synthesize_schema_max_retries, synthesize_output=synthesize_output, synthesize_output_model=synthesize_output_model, synthesize_output_format=synthesize_output_format, ) # Support both @tool and @tool() usage patterns if func is not None: return decorator(func) else: return decorator