Utils

labridge.tools.utils

labridge.tools.utils.create_schema_from_fn_or_method(name, func, additional_fields=None)

Create schema from function.

Source code in labridge\tools\utils.py

def create_schema_from_fn_or_method(
    name: str,
    func: Callable[..., Any],
    additional_fields: Optional[
        List[Union[Tuple[str, Type, Any], Tuple[str, Type]]]
    ] = None,
) -> Type[BaseModel]:
	"""Create schema from function."""
	fields = {}
	params = signature(func).parameters
	for param_name in params:
		if param_name in ["self", "cls"]:
			continue
		param_type = params[param_name].annotation
		param_default = params[param_name].default

		if param_type is params[param_name].empty:
			param_type = Any

		if param_default is params[param_name].empty:
			# Required field
			fields[param_name] = (param_type, FieldInfo())
		elif isinstance(param_default, FieldInfo):
			# Field with pydantic.Field as default value
			fields[param_name] = (param_type, param_default)
		else:
			fields[param_name] = (param_type, FieldInfo(default=param_default))

	additional_fields = additional_fields or []
	for field_info in additional_fields:
		if len(field_info) == 3:
			field_info = cast(Tuple[str, Type, Any], field_info)
			field_name, field_type, field_default = field_info
			fields[field_name] = (field_type, FieldInfo(default=field_default))
		elif len(field_info) == 2:
			# Required field has no default value
			field_info = cast(Tuple[str, Type], field_info)
			field_name, field_type = field_info
			fields[field_name] = (field_type, FieldInfo())
		else:
			raise ValueError(
				f"Invalid additional field info: {field_info}. "
				"Must be a tuple of length 2 or 3."
			)

	return create_model(name, **fields)  # type: ignore

labridge.tools.utils.get_extra_str_to_user(tool_logs)

The log_to_user and the value of the key references in ToolLog will be presented to the user.

Source code in labridge\tools\utils.py

def get_extra_str_to_user(tool_logs: List[ToolLog]) -> str:
	r"""
	The `log_to_user` and the value of the key `references` in ToolLog will be presented to the user.
	"""
	str_list = []

	extra_refs_dict = get_all_ref_info(tool_logs=tool_logs)

	for log in tool_logs:
		if log.log_to_user is not None:
			str_list.append(log.log_to_user.strip())

	for ref_type in extra_refs_dict.keys():
		fn = REF_INFO_TO_STR_FUNC_DICT[ref_type]
		ref_str = fn(extra_refs_dict[ref_type])
		str_list.append(ref_str.strip())
	return "\n".join(str_list)

labridge.tools.utils.get_ref_file_paths(tool_logs)

Source code in labridge\tools\utils.py

def get_ref_file_paths(tool_logs: List[ToolLog]) -> List[str]:
	r"""

	"""
	extra_refs_dict = get_all_ref_info(tool_logs=tool_logs)

	file_paths = []
	for ref_type in extra_refs_dict.keys():
		if ref_type in REF_INFO_TO_FILE_PATH_FUNC_DICT.keys():
			fn = REF_INFO_TO_FILE_PATH_FUNC_DICT[ref_type]
			paths = fn(extra_refs_dict[ref_type])
			file_paths.extend(paths)
	return file_paths

labridge.tools.utils.pack_tool_output(tool_output, tool_log=None)

Pack the tool output and tool log in a dict and dump to string.

PARAMETER	DESCRIPTION
tool_output	The tool output string. TYPE: str
tool_log	The tool log string. TYPE: str DEFAULT: None

RETURNS	DESCRIPTION
str	The dumped tool output and log.

Source code in labridge\tools\utils.py

def pack_tool_output(tool_output: str, tool_log: str = None) -> str:
	r"""
	Pack the tool output and tool log in a dict and dump to string.

	Args:
		tool_output (str): The tool output string.
		tool_log (str): The tool log string.

	Returns:
		The dumped tool output and log.
	"""
	tool_out_dict = {
		"tool_output": tool_output,
		"tool_log": tool_log,
	}
	tool_out_str = json.dumps(tool_out_dict)
	return tool_out_str

labridge.tools.utils.unpack_tool_output(tool_out_json)

Unpack the tool output string and tool log string from

PARAMETER	DESCRIPTION
tool_out_json	TYPE: str

Source code in labridge\tools\utils.py

def unpack_tool_output(tool_out_json: str) -> Tuple[str, Optional[str]]:
	r"""
	Unpack the tool output string and tool log string from

	Args:
		tool_out_json:

	Returns:

	"""
	try:
		tool_out_dict = json.loads(tool_out_json)
		tool_output, tool_log = tool_out_dict["tool_output"], tool_out_dict["tool_log"]
		return tool_output, tool_log
	except Exception:
		return tool_out_json, None

labridge.tools.utils.whether_abort_tool(tool_output)

Whether a tool is aborted during execution.

PARAMETER	DESCRIPTION
tool_output	The tool output of a tool. TYPE: ToolOutput

RETURNS	DESCRIPTION
Optional[bool]	Optional[bool]: If the tool's execution is aborted, return True. If the tool's execution is performed normally, return False. If error raises in this function, return None.

Source code in labridge\tools\utils.py

def whether_abort_tool(tool_output: ToolOutput) -> Optional[bool]:
	r"""
	Whether a tool is aborted during execution.

	Args:
		tool_output (ToolOutput): The tool output of a tool.

	Returns:
		Optional[bool]:

			- If the tool's execution is aborted, return True.
			- If the tool's execution is performed normally, return False.
			- If error raises in this function, return None.

	"""
	try:
		_, create_log_str = unpack_tool_output(tool_output.content)
		create_log = ToolLog.loads(log_str=create_log_str)
		# if the user abort in the creation pipeline
		return create_log.tool_abort
	except ValueError:
		return None