跳转至

Function base tools

labridge.tools.base.function_base_tools

labridge.tools.base.function_base_tools.CallBackBaseTool

Bases: FunctionBaseTool

This is base of tools that will execute operations that need authorization. Refer to the callback module for details.

PARAMETER DESCRIPTION
fn

The function or method that will be called by the agent.

TYPE: Callable

async_fn

The asynchronous version of fn.

TYPE: Callable

callback_operation

The operation that needs the user's authorization.

TYPE: CallBackOperationBase

tool_name

The tool name.

TYPE: str DEFAULT: None

return_direct

Whether to return the tool output directly in Reasoning & Acting process.

TYPE: bool DEFAULT: False

Source code in labridge\tools\base\function_base_tools.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class CallBackBaseTool(FunctionBaseTool):
	r"""
	This is base of tools that will execute operations that need authorization.
	Refer to the `callback` module for details.

	Args:
		fn (Callable): The function or method that will be called by the agent.
		async_fn (Callable): The asynchronous version of `fn`.
		callback_operation (CallBackOperationBase): The operation that needs the user's authorization.
		tool_name (str): The tool name.
		return_direct (bool): Whether to return the tool output directly in Reasoning & Acting process.
	"""
	def __init__(
		self,
		fn: Callable[..., Any],
		async_fn: Callable[..., Any],
		callback_operation: CallBackOperationBase,
		tool_name: str = None,
		return_direct: bool = False,
	):
		self._callback_operation = callback_operation
		super().__init__(
			fn=fn,
			async_fn=async_fn,
			tool_name=tool_name,
			return_direct=return_direct,
		)

	@abstractmethod
	def log(self, **kwargs: Any) -> ToolLog:
		r""" Return the log json string, describing the tool's operation. """

labridge.tools.base.function_base_tools.CallBackBaseTool.log(**kwargs) abstractmethod

Return the log json string, describing the tool's operation.

Source code in labridge\tools\base\function_base_tools.py 
161
162
163
@abstractmethod
def log(self, **kwargs: Any) -> ToolLog:
	r""" Return the log json string, describing the tool's operation. """

labridge.tools.base.function_base_tools.FuncOutputWithLog

This class is the output format of the function in a FunctionBaseTool.

PARAMETER DESCRIPTION
fn_output

The output of the function.

TYPE: str

fn_log

The log of the function.

TYPE: Union[str, Dict[str, Any]]

Source code in labridge\tools\base\function_base_tools.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class FuncOutputWithLog:
	r"""
	This class is the output format of the function in a FunctionBaseTool.

	Args:
		fn_output (str): The output of the function.
		fn_log (Union[str, Dict[str, Any]]): The log of the function.
	"""
	def __init__(
		self,
		fn_output: Optional[str],
		fn_log: Union[str, Dict[str, Any]]
	):
		self.fn_output = fn_output
		self.fn_log = fn_log

labridge.tools.base.function_base_tools.FunctionBaseTool

Bases: CheckBaseTool

This tool is the base of function-type or method-type tools.

PARAMETER DESCRIPTION
fn

The function or method that will be called by the agent.

TYPE: Callable

async_fn

The asynchronous version of fn.

TYPE: Callable DEFAULT: None

tool_name

The tool name. If not specified, the fn.__name__ will be used as the tool name.

TYPE: str DEFAULT: None

return_direct

Whether to return the tool output directly in the Reasoning & Acting process. Refer to ReactAgent for details.

TYPE: str DEFAULT: False

Source code in labridge\tools\base\function_base_tools.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
class FunctionBaseTool(CheckBaseTool):
	r"""
	This tool is the base of function-type or method-type tools.

	Args:
		fn (Callable): The function or method that will be called by the agent.
		async_fn (Callable): The asynchronous version of `fn`.
		tool_name (str): The tool name. If not specified, the `fn.__name__` will be used as the tool name.
		return_direct (str): Whether to return the tool output directly in the Reasoning & Acting process.
			Refer to `ReactAgent` for details.
	"""
	def __init__(
		self,
		fn: Callable[..., Any],
		async_fn: Callable[..., Any] = None,
		tool_name: str = None,
		return_direct: bool = False,
	):
		name = tool_name or fn.__name__
		docstring = fn.__doc__
		description = f"{name}{signature(fn)}\n{docstring}"
		fn_schema = create_schema_from_fn_or_method(f"{name}", fn, additional_fields=None)
		metadata = ToolMetadata(
			name=name,
			description=description,
			fn_schema=fn_schema,
			return_direct=return_direct,
		)

		self._fn = fn
		if async_fn is not None:
			self._async_fn = async_fn
		else:
			self._async_fn = sync_to_async(self._fn)
		super().__init__(metadata=metadata)

	@abstractmethod
	def log(self, **kwargs: Any) -> ToolLog:
		r""" Return the log json string, describing the tool's operation. """

	def call(self, **kwargs: Any) -> ToolOutput:
		""" Call, return output and log. """
		checked_kwargs = self._get_input(**kwargs)
		output_with_log = self._fn(**checked_kwargs)

		if not isinstance(output_with_log, FuncOutputWithLog):
			raise ValueError("The function of a function tool must output 'FuncOutputWithLog'.")

		fn_output = output_with_log.fn_output
		fn_log = output_with_log.fn_log
		if not isinstance(fn_log, dict):
			fn_log = {"fn_log": fn_log}
		else:
			fn_log = fn_log

		checked_kwargs.update(fn_log)
		tool_log = self.log(**checked_kwargs)
		tool_output = pack_tool_output(tool_output=fn_output, tool_log=tool_log.dumps())

		return ToolOutput(
			content=str(tool_output),
			tool_name=self.metadata.name,
			raw_input={"kwargs": kwargs},
			raw_output=tool_output,
		)

	async def acall(self, **kwargs: Any) -> ToolOutput:
		""" Asynchronous Call. """
		checked_kwargs = self._get_input(**kwargs)
		output_with_log = await self._async_fn(**checked_kwargs)
		if not isinstance(output_with_log, FuncOutputWithLog):
			raise ValueError("The function of a function tool must output 'FuncOutputWithLog'.")

		fn_output = output_with_log.fn_output
		fn_log = output_with_log.fn_log
		if not isinstance(fn_log, dict):
			fn_log = {"fn_log": fn_log}
		else:
			fn_log = fn_log

		checked_kwargs.update(fn_log)
		tool_log = self.log(**checked_kwargs)
		tool_output = pack_tool_output(tool_output=fn_output, tool_log=tool_log.dumps())
		return ToolOutput(
			content=str(tool_output),
			tool_name=self.metadata.name,
			raw_input={"kwargs": kwargs},
			raw_output=tool_output,
		)

labridge.tools.base.function_base_tools.FunctionBaseTool.acall(**kwargs) async

Asynchronous Call.

Source code in labridge\tools\base\function_base_tools.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
async def acall(self, **kwargs: Any) -> ToolOutput:
	""" Asynchronous Call. """
	checked_kwargs = self._get_input(**kwargs)
	output_with_log = await self._async_fn(**checked_kwargs)
	if not isinstance(output_with_log, FuncOutputWithLog):
		raise ValueError("The function of a function tool must output 'FuncOutputWithLog'.")

	fn_output = output_with_log.fn_output
	fn_log = output_with_log.fn_log
	if not isinstance(fn_log, dict):
		fn_log = {"fn_log": fn_log}
	else:
		fn_log = fn_log

	checked_kwargs.update(fn_log)
	tool_log = self.log(**checked_kwargs)
	tool_output = pack_tool_output(tool_output=fn_output, tool_log=tool_log.dumps())
	return ToolOutput(
		content=str(tool_output),
		tool_name=self.metadata.name,
		raw_input={"kwargs": kwargs},
		raw_output=tool_output,
	)

labridge.tools.base.function_base_tools.FunctionBaseTool.call(**kwargs)

Call, return output and log.

Source code in labridge\tools\base\function_base_tools.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def call(self, **kwargs: Any) -> ToolOutput:
	""" Call, return output and log. """
	checked_kwargs = self._get_input(**kwargs)
	output_with_log = self._fn(**checked_kwargs)

	if not isinstance(output_with_log, FuncOutputWithLog):
		raise ValueError("The function of a function tool must output 'FuncOutputWithLog'.")

	fn_output = output_with_log.fn_output
	fn_log = output_with_log.fn_log
	if not isinstance(fn_log, dict):
		fn_log = {"fn_log": fn_log}
	else:
		fn_log = fn_log

	checked_kwargs.update(fn_log)
	tool_log = self.log(**checked_kwargs)
	tool_output = pack_tool_output(tool_output=fn_output, tool_log=tool_log.dumps())

	return ToolOutput(
		content=str(tool_output),
		tool_name=self.metadata.name,
		raw_input={"kwargs": kwargs},
		raw_output=tool_output,
	)

labridge.tools.base.function_base_tools.FunctionBaseTool.log(**kwargs) abstractmethod

Return the log json string, describing the tool's operation.

Source code in labridge\tools\base\function_base_tools.py 
78
79
80
@abstractmethod
def log(self, **kwargs: Any) -> ToolLog:
	r""" Return the log json string, describing the tool's operation. """