123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185 |
- import json
- import os
- from abc import ABC, abstractmethod
- from typing import Dict, List, Optional, Union
- from qwen_agent.llm.schema import ContentItem
- from qwen_agent.settings import DEFAULT_WORKSPACE
- from qwen_agent.utils.utils import has_chinese_chars, json_loads, logger, print_traceback, save_url_to_local_work_dir
- TOOL_REGISTRY = {}
- def register_tool(name, allow_overwrite=False):
- def decorator(cls):
- if name in TOOL_REGISTRY:
- if allow_overwrite:
- logger.warning(f'Tool `{name}` already exists! Overwriting with class {cls}.')
- else:
- raise ValueError(f'Tool `{name}` already exists! Please ensure that the tool name is unique.')
- if cls.name and (cls.name != name):
- raise ValueError(f'{cls.__name__}.name="{cls.name}" conflicts with @register_tool(name="{name}").')
- cls.name = name
- TOOL_REGISTRY[name] = cls
- return cls
- return decorator
- def is_tool_schema(obj: dict) -> bool:
- """
- Check if obj is a valid JSON schema describing a tool compatible with OpenAI's tool calling.
- Example valid schema:
- {
- "name": "get_current_weather",
- "description": "Get the current weather in a given location",
- "parameters": {
- "type": "object",
- "properties": {
- "location": {
- "type": "string",
- "description": "The city and state, e.g. San Francisco, CA"
- },
- "unit": {
- "type": "string",
- "enum": ["celsius", "fahrenheit"]
- }
- },
- "required": ["location"]
- }
- }
- """
- import jsonschema
- try:
- assert set(obj.keys()) == {'name', 'description', 'parameters'}
- assert isinstance(obj['name'], str)
- assert obj['name'].strip()
- assert isinstance(obj['description'], str)
- assert isinstance(obj['parameters'], dict)
- assert set(obj['parameters'].keys()) == {'type', 'properties', 'required'}
- assert obj['parameters']['type'] == 'object'
- assert isinstance(obj['parameters']['properties'], dict)
- assert isinstance(obj['parameters']['required'], list)
- assert set(obj['parameters']['required']).issubset(set(obj['parameters']['properties'].keys()))
- except AssertionError:
- return False
- try:
- jsonschema.validate(instance={}, schema=obj['parameters'])
- except jsonschema.exceptions.SchemaError:
- return False
- except jsonschema.exceptions.ValidationError:
- pass
- return True
- class BaseTool(ABC):
- name: str = ''
- description: str = ''
- parameters: Union[List[dict], dict] = []
- def __init__(self, cfg: Optional[dict] = None):
- self.cfg = cfg or {}
- if not self.name:
- raise ValueError(
- f'You must set {self.__class__.__name__}.name, either by @register_tool(name=...) or explicitly setting {self.__class__.__name__}.name'
- )
- if isinstance(self.parameters, dict):
- if not is_tool_schema({'name': self.name, 'description': self.description, 'parameters': self.parameters}):
- raise ValueError(
- 'The parameters, when provided as a dict, must confirm to a valid openai-compatible JSON schema.')
- @abstractmethod
- def call(self, params: Union[str, dict], **kwargs) -> Union[str, list, dict, List[ContentItem]]:
- """The interface for calling tools.
- Each tool needs to implement this function, which is the workflow of the tool.
- Args:
- params: The parameters of func_call.
- kwargs: Additional parameters for calling tools.
- Returns:
- The result returned by the tool, implemented in the subclass.
- """
- raise NotImplementedError
- def _verify_json_format_args(self, params: Union[str, dict], strict_json: bool = False) -> dict:
- """Verify the parameters of the function call"""
- if isinstance(params, str):
- try:
- if strict_json:
- params_json: dict = json.loads(params)
- else:
- params_json: dict = json_loads(params)
- except json.decoder.JSONDecodeError:
- raise ValueError('Parameters must be formatted as a valid JSON!')
- else:
- params_json: dict = params
- if isinstance(self.parameters, list):
- for param in self.parameters:
- if 'required' in param and param['required']:
- if param['name'] not in params_json:
- raise ValueError('Parameters %s is required!' % param['name'])
- elif isinstance(self.parameters, dict):
- import jsonschema
- jsonschema.validate(instance=params_json, schema=self.parameters)
- else:
- raise ValueError
- return params_json
- @property
- def function(self) -> dict: # Bad naming. It should be `function_info`.
- return {
- 'name_for_human': self.name_for_human,
- 'name': self.name,
- 'description': self.description,
- 'parameters': self.parameters,
- 'args_format': self.args_format
- }
- @property
- def name_for_human(self) -> str:
- return self.cfg.get('name_for_human', self.name)
- @property
- def args_format(self) -> str:
- fmt = self.cfg.get('args_format')
- if fmt is None:
- if has_chinese_chars([self.name_for_human, self.name, self.description, self.parameters]):
- fmt = '此工具的输入应为JSON对象。'
- else:
- fmt = 'Format the arguments as a JSON object.'
- return fmt
- @property
- def file_access(self) -> bool:
- return False
- class BaseToolWithFileAccess(BaseTool, ABC):
- def __init__(self, cfg: Optional[Dict] = None):
- super().__init__(cfg)
- assert self.name
- default_work_dir = os.path.join(DEFAULT_WORKSPACE, 'tools', self.name)
- self.work_dir: str = self.cfg.get('work_dir', default_work_dir)
- @property
- def file_access(self) -> bool:
- return True
- def call(self, params: Union[str, dict], files: List[str] = None, **kwargs) -> str:
- # Copy remote files to the working directory:
- if files:
- os.makedirs(self.work_dir, exist_ok=True)
- for file in files:
- try:
- save_url_to_local_work_dir(file, self.work_dir)
- except Exception:
- print_traceback()
- # Then do something with the files:
- # ...
|