Insert

labridge.tools.memory.experiment.insert

labridge.tools.memory.experiment.insert.CreateNewExperimentLogTool

Bases: CollectAndAuthorizeTool

This tool is used to create a new experiment record in the user's experiment log storage. It is a CollectAndAuthorizeTool that needs information collection and authorization.

PARAMETER	DESCRIPTION
llm	The used LLM. TYPE: LLM DEFAULT: None
embed_model	The used embedding model. TYPE: BaseEmbedding DEFAULT: None
verbose	Whether to show the inner progress. TYPE: bool DEFAULT: False

Source code in labridge\tools\memory\experiment\insert.py

class CreateNewExperimentLogTool(CollectAndAuthorizeTool):
	r"""
	This tool is used to create a new experiment record in the user's experiment log storage.
	It is a `CollectAndAuthorizeTool` that needs information collection and authorization.

	Args:
		llm (LLM): The used LLM.
		embed_model (BaseEmbedding): The used embedding model.
		verbose (bool): Whether to show the inner progress.
	"""
	def __init__(
		self,
		llm: LLM = None,
		embed_model: BaseEmbedding = None,
		verbose: bool = False
	):
		super().__init__(
			tool_fn=self.create_new_experiment_log,
			tool_async_fn=self.acreate_new_experiment_log,
			tool_name=CreateNewExperimentLogTool.__name__,
			callback_operation=CreateNewExperimentLogOperation,
			llm=llm,
			embed_model=embed_model,
			verbose=verbose,
		)

	def required_info_dict(self) -> Dict[str, str]:
		r"""
		The required information.

		Returns:
			Dict[str, str]:

				- key: the information name.
				- value: the information description.
		"""
		return NEW_EXPERIMENT_REQUIRED_INFOS

	def required_infos(self) -> List[CollectingInfoBase]:
		r"""
		The required infos.

		Returns:
			List[CollectingInfoBase]:
				Return the packed info in CollectingInfoBase, such as `CollectingCommonInfo`.
		"""
		infos = []
		info_dict = self.required_info_dict()
		for key in info_dict.keys():
			common_info = CollectingCommonInfo(
				info_name=key,
				info_description=info_dict[key],
			)
			infos.append(common_info)
		return infos

	def create_new_experiment_log(
		self,
		user_id: str,
	) -> FuncOutputWithLog:
		r"""
		This tool is used to create a new experiment log record for the user.
		This tool is only used when the user asks for creating a new experiment log record, or other tools call this tool.

		Args:
			user_id (str): The user_id of a lab member.

		Returns:
			The tool's output and log.
		"""
		# This docstring is used as the tool description.
		return self.collect_and_authorize(
			user_id=user_id,
			query_str=COLLECT_NEW_EXPERIMENT_INFO_QUERY,
		)

	async def acreate_new_experiment_log(
		self,
		user_id: str,
	) -> FuncOutputWithLog:
		r"""
		This tool is used to create a new experiment log record for the user.
		This tool is only used when the user asks for creating a new experiment log record,
		or when other tools call this tool.

		Args:
			user_id (str): The user_id of a lab member.

		Returns:
			The tool's output and log.
		"""
		output = await self.acollect_and_authorize(
			user_id=user_id,
			query_str=COLLECT_NEW_EXPERIMENT_INFO_QUERY,
		)
		print("Here: ", output.fn_output)
		print("Here: ", output.fn_log)
		return output

labridge.tools.memory.experiment.insert.CreateNewExperimentLogTool.acreate_new_experiment_log(user_id) async

This tool is used to create a new experiment log record for the user. This tool is only used when the user asks for creating a new experiment log record, or when other tools call this tool.

PARAMETER	DESCRIPTION
user_id	The user_id of a lab member. TYPE: str

RETURNS	DESCRIPTION
FuncOutputWithLog	The tool's output and log.

Source code in labridge\tools\memory\experiment\insert.py

async def acreate_new_experiment_log(
	self,
	user_id: str,
) -> FuncOutputWithLog:
	r"""
	This tool is used to create a new experiment log record for the user.
	This tool is only used when the user asks for creating a new experiment log record,
	or when other tools call this tool.

	Args:
		user_id (str): The user_id of a lab member.

	Returns:
		The tool's output and log.
	"""
	output = await self.acollect_and_authorize(
		user_id=user_id,
		query_str=COLLECT_NEW_EXPERIMENT_INFO_QUERY,
	)
	print("Here: ", output.fn_output)
	print("Here: ", output.fn_log)
	return output

labridge.tools.memory.experiment.insert.CreateNewExperimentLogTool.create_new_experiment_log(user_id)

This tool is used to create a new experiment log record for the user. This tool is only used when the user asks for creating a new experiment log record, or other tools call this tool.

PARAMETER	DESCRIPTION
user_id	The user_id of a lab member. TYPE: str

RETURNS	DESCRIPTION
FuncOutputWithLog	The tool's output and log.

Source code in labridge\tools\memory\experiment\insert.py

def create_new_experiment_log(
	self,
	user_id: str,
) -> FuncOutputWithLog:
	r"""
	This tool is used to create a new experiment log record for the user.
	This tool is only used when the user asks for creating a new experiment log record, or other tools call this tool.

	Args:
		user_id (str): The user_id of a lab member.

	Returns:
		The tool's output and log.
	"""
	# This docstring is used as the tool description.
	return self.collect_and_authorize(
		user_id=user_id,
		query_str=COLLECT_NEW_EXPERIMENT_INFO_QUERY,
	)

labridge.tools.memory.experiment.insert.CreateNewExperimentLogTool.required_info_dict()

The required information.

RETURNS	DESCRIPTION
Dict[str, str]	Dict[str, str]: key: the information name. value: the information description.

Source code in labridge\tools\memory\experiment\insert.py

def required_info_dict(self) -> Dict[str, str]:
	r"""
	The required information.

	Returns:
		Dict[str, str]:

			- key: the information name.
			- value: the information description.
	"""
	return NEW_EXPERIMENT_REQUIRED_INFOS

labridge.tools.memory.experiment.insert.CreateNewExperimentLogTool.required_infos()

The required infos.

RETURNS	DESCRIPTION
List[CollectingInfoBase]	List[CollectingInfoBase]: Return the packed info in CollectingInfoBase, such as CollectingCommonInfo.

Source code in labridge\tools\memory\experiment\insert.py

def required_infos(self) -> List[CollectingInfoBase]:
	r"""
	The required infos.

	Returns:
		List[CollectingInfoBase]:
			Return the packed info in CollectingInfoBase, such as `CollectingCommonInfo`.
	"""
	infos = []
	info_dict = self.required_info_dict()
	for key in info_dict.keys():
		common_info = CollectingCommonInfo(
			info_name=key,
			info_description=info_dict[key],
		)
		infos.append(common_info)
	return infos

labridge.tools.memory.experiment.insert.RecordExperimentLogTool

Bases: FunctionBaseTool

This tool is used to record the experiment log for users. Use this tool When the user asks you to record anything about his/her experiment.

• If the recorded experiment in progress is valid (That is, current time is not beyond this experiment's duration), the experiment log will be directly record to the record of the experiment in progress.
• If the recorded experiment in progress is not valid, this tool will implicitly call the SetCurrentExperimentTool to set current experiment (Output the tool call requirement in the tool output.).

PARAMETER	DESCRIPTION
llm	The used LLM. If not specified, the Settings.llm will be used. TYPE: LLM DEFAULT: None
embed_model	The used embedding model. If not specified, the Settings.embed_model will be used. TYPE: BaseEmbedding DEFAULT: None
verbose	Whether to show the inner progress. TYPE: bool DEFAULT: False

Source code in labridge\tools\memory\experiment\insert.py

class RecordExperimentLogTool(FunctionBaseTool):
	r"""
	This tool is used to record the experiment log for users.
	Use this tool When the user asks you to record anything about his/her experiment.

	- If the recorded experiment in progress is valid (That is, current time is not beyond this experiment's duration),
	the experiment log will be directly record to the record of the experiment in progress.
	- If the recorded experiment in progress is not valid, this tool will implicitly call the `SetCurrentExperimentTool`
	to set current experiment (Output the tool call requirement in the tool output.).

	Args:
		llm (LLM): The used LLM. If not specified, the `Settings.llm` will be used.
		embed_model (BaseEmbedding): The used embedding model. If not specified, the `Settings.embed_model` will be used.
		verbose (bool): Whether to show the inner progress.
	"""
	def __init__(
		self,
		llm: LLM = None,
		embed_model: BaseEmbedding = None,
		verbose: bool = False,
	):
		self._llm = llm or Settings.llm
		self._embed_model = embed_model or Settings.embed_model
		self._verbose = verbose
		super().__init__(
			fn=self.record_log,
			async_fn=self.arecord_log,
			tool_name=RecordExperimentLogTool.__name__,
		)

	def log(self, **kwargs: Any) -> ToolLog:
		r"""
		Record the tool log.

		Args:
			**kwargs (Any): The input keyword arguments and the (output, log) of the executed operation.

		Returns:

		"""
		op_log: OperationOutputLog = kwargs["operation_log"]

		return ToolLog.construct(
			tool_name=self.metadata.name,
			tool_op_description=op_log.log_to_system[OP_DESCRIPTION],
		)

	def record_log(
		self,
		user_id: str,
		log_str: str,
		attached_file_path: str = None,
		**kwargs,
	) -> FuncOutputWithLog:
		r"""
		This tool is used to record the experiment log of the experiment in progress for a user.

		If the no experiment record exists or experiment in progress is not valid, this tool will call
		the corresponding tools to help the user.

		Args:
			user_id (str): The user_id of a lab member.
			log_str (str): The experiment log to be recorded.
			attached_file_path (str): The path of the attached file, If the user attaches a file along with the experiment log.
				Defaults to None.

		Returns:
			The tool output and log.
		"""
		# This docstring is used as the tool description.
		expr_log_store = ExperimentLog.from_user_id(
			user_id=user_id,
			embed_model=self._embed_model,
		)

		# If no experiment log record exists.
		if expr_log_store.get_all_experiments() is None:
			create_tool = CreateNewExperimentLogTool(
				llm=self._llm,
				embed_model=self._embed_model,
			)
			create_output = create_tool.call(user_id=user_id)
			abort = whether_abort_tool(tool_output=create_output)
			if abort is None or abort:
				status = (
					f"The user does not want to continue, and abort the create experiment operation."
					f"This conversation should be ended."
				)
				if abort is None:
					status = f"The tool output of the CreateNewExperimentLogTool is invalid."

				op_log = OperationOutputLog.construct(
					operation_name=create_tool.metadata.name,
					op_description=status,
					operation_output=status,
				)
				return FuncOutputWithLog(
					fn_output=status,
					fn_log={"operation_log": op_log}
				)

		# If current experiment in progress is not valid.
		recent_expr = expr_log_store.get_recent_experiment()
		if recent_expr is None:
			set_tool = SetCurrentExperimentTool(
				llm=self._llm,
				embed_model=self._embed_model,
			)

			set_output = set_tool.call(user_id=user_id)
			abort = whether_abort_tool(tool_output=set_output)
			if abort is None or abort:
				status = (
					f"The user does not want to continue, and abort the set experiment operation."
					f"This conversation should be ended."
				)
				if abort is None:
					status = f"The tool output of the SetCurrentExperimentTool is invalid."

				op_log = OperationOutputLog.construct(
					operation_name=set_tool.metadata.name,
					op_description=status,
					operation_output=status,
				)
				return FuncOutputWithLog(
					fn_output=status,
					fn_log={"operation_log": op_log},
				)

			# reload
			expr_log_store = ExperimentLog.from_user_id(
				user_id=user_id,
				embed_model=self._embed_model,
			)
			recent_expr = expr_log_store.get_recent_experiment()

		expr_log_store.put(
			experiment_name=recent_expr,
			log_str=log_str,
			attached_file_path=attached_file_path,
		)
		expr_log_store.persist()

		op_log_str = (
			f"Have put a new experiment log into the experiment log store of the user: {user_id}.\n" 
			f"Experiment name: {recent_expr}\n"
		)
		op_log = OperationOutputLog.construct(
			operation_name=self.metadata.name,
			op_description=op_log_str,
			operation_output=op_log_str,
		)
		return FuncOutputWithLog(
			fn_output=f"Have record the log {log_str}",
			fn_log={"operation_log": op_log}
		)

	async def arecord_log(
		self,
		user_id: str,
		log_str: str,
		attached_file_path: str = None,
		**kwargs,
	) -> FuncOutputWithLog:
		r"""
		This tool is used to record the experiment log of the experiment in progress for a user.

		If the no experiment record exists or experiment in progress is not valid, this tool will call
		the corresponding tools to help the user.

		Args:
			user_id (str): The user_id of a lab member.
			log_str (str): The experiment log to be recorded.
			attached_file_path (str): The path of the attached file, If the user attaches a file along with the experiment log.
				Defaults to None.

		Returns:
			The tool output and log.
		"""
		# This docstring is used as the tool description.
		expr_log_store = ExperimentLog.from_user_id(
			user_id=user_id,
			embed_model=self._embed_model,
		)

		# If no experiment log record exists.
		if expr_log_store.get_all_experiments() is None:
			create_tool = CreateNewExperimentLogTool(
				llm=self._llm,
				embed_model=self._embed_model,
			)
			create_output = await create_tool.acall(user_id=user_id)

			abort = whether_abort_tool(tool_output=create_output)
			if abort is None or abort:
				status = f"The user does not want to continue, and abort the create experiment operation."
				if abort is None:
					status = f"The tool output of the CreateNewExperimentLogTool is invalid."

				op_log = OperationOutputLog.construct(
					operation_name=create_tool.metadata.name,
					op_description=status,
					operation_output=status,
				)
				return FuncOutputWithLog(
					fn_output=status,
					fn_log={"operation_log": op_log}
				)

		# If current experiment in progress is not valid.
		recent_expr = expr_log_store.get_recent_experiment()
		if recent_expr is None:
			set_tool = SetCurrentExperimentTool(
				llm=self._llm,
				embed_model=self._embed_model,
			)
			set_output = await set_tool.acall(user_id=user_id)

			abort = whether_abort_tool(tool_output=set_output)
			if abort is None or abort:
				status = f"The user does not want to continue, and abort the set experiment operation."
				if abort is None:
					status = f"The tool output of the SetCurrentExperimentTool is invalid."

				op_log = OperationOutputLog.construct(
					operation_name=set_tool.metadata.name,
					op_description=status,
					operation_output=status,
				)
				return FuncOutputWithLog(
					fn_output=status,
					fn_log={"operation_log": op_log},
				)

			# reload
			expr_log_store = ExperimentLog.from_user_id(
				user_id=user_id,
				embed_model=self._embed_model,
			)
			recent_expr = expr_log_store.get_recent_experiment()

		expr_log_store.put(
			experiment_name=recent_expr,
			log_str=log_str,
			attached_file_path=attached_file_path,
		)
		expr_log_store.persist()

		op_log_str = (
			f"Have put a new experiment log into the experiment log store of the user: {user_id}.\n"
			f"Experiment name: {recent_expr}\n"
		)
		op_log = OperationOutputLog.construct(
			operation_name=self.metadata.name,
			op_description=op_log_str,
			operation_output=op_log_str,
		)
		return FuncOutputWithLog(
			fn_output=f"Have record the log {log_str}",
			fn_log={"operation_log": op_log}
		)

labridge.tools.memory.experiment.insert.RecordExperimentLogTool.arecord_log(user_id, log_str, attached_file_path=None, **kwargs) async

This tool is used to record the experiment log of the experiment in progress for a user.

If the no experiment record exists or experiment in progress is not valid, this tool will call the corresponding tools to help the user.

PARAMETER	DESCRIPTION
user_id	The user_id of a lab member. TYPE: str
log_str	The experiment log to be recorded. TYPE: str
attached_file_path	The path of the attached file, If the user attaches a file along with the experiment log. Defaults to None. TYPE: str DEFAULT: None

RETURNS	DESCRIPTION
FuncOutputWithLog	The tool output and log.

Source code in labridge\tools\memory\experiment\insert.py

async def arecord_log(
	self,
	user_id: str,
	log_str: str,
	attached_file_path: str = None,
	**kwargs,
) -> FuncOutputWithLog:
	r"""
	This tool is used to record the experiment log of the experiment in progress for a user.

	If the no experiment record exists or experiment in progress is not valid, this tool will call
	the corresponding tools to help the user.

	Args:
		user_id (str): The user_id of a lab member.
		log_str (str): The experiment log to be recorded.
		attached_file_path (str): The path of the attached file, If the user attaches a file along with the experiment log.
			Defaults to None.

	Returns:
		The tool output and log.
	"""
	# This docstring is used as the tool description.
	expr_log_store = ExperimentLog.from_user_id(
		user_id=user_id,
		embed_model=self._embed_model,
	)

	# If no experiment log record exists.
	if expr_log_store.get_all_experiments() is None:
		create_tool = CreateNewExperimentLogTool(
			llm=self._llm,
			embed_model=self._embed_model,
		)
		create_output = await create_tool.acall(user_id=user_id)

		abort = whether_abort_tool(tool_output=create_output)
		if abort is None or abort:
			status = f"The user does not want to continue, and abort the create experiment operation."
			if abort is None:
				status = f"The tool output of the CreateNewExperimentLogTool is invalid."

			op_log = OperationOutputLog.construct(
				operation_name=create_tool.metadata.name,
				op_description=status,
				operation_output=status,
			)
			return FuncOutputWithLog(
				fn_output=status,
				fn_log={"operation_log": op_log}
			)

	# If current experiment in progress is not valid.
	recent_expr = expr_log_store.get_recent_experiment()
	if recent_expr is None:
		set_tool = SetCurrentExperimentTool(
			llm=self._llm,
			embed_model=self._embed_model,
		)
		set_output = await set_tool.acall(user_id=user_id)

		abort = whether_abort_tool(tool_output=set_output)
		if abort is None or abort:
			status = f"The user does not want to continue, and abort the set experiment operation."
			if abort is None:
				status = f"The tool output of the SetCurrentExperimentTool is invalid."

			op_log = OperationOutputLog.construct(
				operation_name=set_tool.metadata.name,
				op_description=status,
				operation_output=status,
			)
			return FuncOutputWithLog(
				fn_output=status,
				fn_log={"operation_log": op_log},
			)

		# reload
		expr_log_store = ExperimentLog.from_user_id(
			user_id=user_id,
			embed_model=self._embed_model,
		)
		recent_expr = expr_log_store.get_recent_experiment()

	expr_log_store.put(
		experiment_name=recent_expr,
		log_str=log_str,
		attached_file_path=attached_file_path,
	)
	expr_log_store.persist()

	op_log_str = (
		f"Have put a new experiment log into the experiment log store of the user: {user_id}.\n"
		f"Experiment name: {recent_expr}\n"
	)
	op_log = OperationOutputLog.construct(
		operation_name=self.metadata.name,
		op_description=op_log_str,
		operation_output=op_log_str,
	)
	return FuncOutputWithLog(
		fn_output=f"Have record the log {log_str}",
		fn_log={"operation_log": op_log}
	)

labridge.tools.memory.experiment.insert.RecordExperimentLogTool.log(**kwargs)

Record the tool log.

PARAMETER	DESCRIPTION
**kwargs	The input keyword arguments and the (output, log) of the executed operation. TYPE: Any DEFAULT: {}

Source code in labridge\tools\memory\experiment\insert.py

def log(self, **kwargs: Any) -> ToolLog:
	r"""
	Record the tool log.

	Args:
		**kwargs (Any): The input keyword arguments and the (output, log) of the executed operation.

	Returns:

	"""
	op_log: OperationOutputLog = kwargs["operation_log"]

	return ToolLog.construct(
		tool_name=self.metadata.name,
		tool_op_description=op_log.log_to_system[OP_DESCRIPTION],
	)

labridge.tools.memory.experiment.insert.RecordExperimentLogTool.record_log(user_id, log_str, attached_file_path=None, **kwargs)

This tool is used to record the experiment log of the experiment in progress for a user.

If the no experiment record exists or experiment in progress is not valid, this tool will call the corresponding tools to help the user.

PARAMETER	DESCRIPTION
user_id	The user_id of a lab member. TYPE: str
log_str	The experiment log to be recorded. TYPE: str
attached_file_path	The path of the attached file, If the user attaches a file along with the experiment log. Defaults to None. TYPE: str DEFAULT: None

RETURNS	DESCRIPTION
FuncOutputWithLog	The tool output and log.

Source code in labridge\tools\memory\experiment\insert.py

def record_log(
	self,
	user_id: str,
	log_str: str,
	attached_file_path: str = None,
	**kwargs,
) -> FuncOutputWithLog:
	r"""
	This tool is used to record the experiment log of the experiment in progress for a user.

	If the no experiment record exists or experiment in progress is not valid, this tool will call
	the corresponding tools to help the user.

	Args:
		user_id (str): The user_id of a lab member.
		log_str (str): The experiment log to be recorded.
		attached_file_path (str): The path of the attached file, If the user attaches a file along with the experiment log.
			Defaults to None.

	Returns:
		The tool output and log.
	"""
	# This docstring is used as the tool description.
	expr_log_store = ExperimentLog.from_user_id(
		user_id=user_id,
		embed_model=self._embed_model,
	)

	# If no experiment log record exists.
	if expr_log_store.get_all_experiments() is None:
		create_tool = CreateNewExperimentLogTool(
			llm=self._llm,
			embed_model=self._embed_model,
		)
		create_output = create_tool.call(user_id=user_id)
		abort = whether_abort_tool(tool_output=create_output)
		if abort is None or abort:
			status = (
				f"The user does not want to continue, and abort the create experiment operation."
				f"This conversation should be ended."
			)
			if abort is None:
				status = f"The tool output of the CreateNewExperimentLogTool is invalid."

			op_log = OperationOutputLog.construct(
				operation_name=create_tool.metadata.name,
				op_description=status,
				operation_output=status,
			)
			return FuncOutputWithLog(
				fn_output=status,
				fn_log={"operation_log": op_log}
			)

	# If current experiment in progress is not valid.
	recent_expr = expr_log_store.get_recent_experiment()
	if recent_expr is None:
		set_tool = SetCurrentExperimentTool(
			llm=self._llm,
			embed_model=self._embed_model,
		)

		set_output = set_tool.call(user_id=user_id)
		abort = whether_abort_tool(tool_output=set_output)
		if abort is None or abort:
			status = (
				f"The user does not want to continue, and abort the set experiment operation."
				f"This conversation should be ended."
			)
			if abort is None:
				status = f"The tool output of the SetCurrentExperimentTool is invalid."

			op_log = OperationOutputLog.construct(
				operation_name=set_tool.metadata.name,
				op_description=status,
				operation_output=status,
			)
			return FuncOutputWithLog(
				fn_output=status,
				fn_log={"operation_log": op_log},
			)

		# reload
		expr_log_store = ExperimentLog.from_user_id(
			user_id=user_id,
			embed_model=self._embed_model,
		)
		recent_expr = expr_log_store.get_recent_experiment()

	expr_log_store.put(
		experiment_name=recent_expr,
		log_str=log_str,
		attached_file_path=attached_file_path,
	)
	expr_log_store.persist()

	op_log_str = (
		f"Have put a new experiment log into the experiment log store of the user: {user_id}.\n" 
		f"Experiment name: {recent_expr}\n"
	)
	op_log = OperationOutputLog.construct(
		operation_name=self.metadata.name,
		op_description=op_log_str,
		operation_output=op_log_str,
	)
	return FuncOutputWithLog(
		fn_output=f"Have record the log {log_str}",
		fn_log={"operation_log": op_log}
	)

labridge.tools.memory.experiment.insert.SetCurrentExperimentTool

Bases: CollectAndAuthorizeTool

This tool is used to set the experiment in progress for a user.

PARAMETER	DESCRIPTION
llm	The used LLM. TYPE: LLM DEFAULT: None
embed_model	The used embedding model. TYPE: BaseEmbedding DEFAULT: None
verbose	Whether to show the inner progress. TYPE: bool DEFAULT: False

Source code in labridge\tools\memory\experiment\insert.py

class SetCurrentExperimentTool(CollectAndAuthorizeTool):
	r"""
	This tool is used to set the experiment in progress for a user.

	Args:
		llm (LLM): The used LLM.
		embed_model (BaseEmbedding): The used embedding model.
		verbose (bool): Whether to show the inner progress.
	"""
	def __init__(
		self,
		llm: LLM = None,
		embed_model: BaseEmbedding = None,
		verbose: bool = False,
	):
		self.expr_log_store = None
		super().__init__(
			tool_fn=self.set_current_experiment,
			tool_async_fn=self.aset_current_experiment,
			tool_name=SetCurrentExperimentTool.__name__,
			callback_operation=SetCurrentExperimentOperation,
			llm=llm,
			embed_model=embed_model,
			verbose=verbose,
		)

	def required_info_dict(self) -> Dict[str, str]:
		r"""
		The required information.

		Returns:
			Dict[str, str]:

				- key: the information name.
				- value: the information description.
		"""
		return SET_CURRENT_EXPERIMENT_REQUIRED_INFOS

	def required_infos(self) -> List[CollectingInfoBase]:
		r"""
		The required infos.

		Returns:
			List[CollectingInfoBase]:
				Return the packed info in CollectingInfoBase, such as `CollectingCommonInfo` and `CollectingSelectInfo`.
		"""
		experiments = self.expr_log_store.get_all_experiments_with_description()
		select_name_info = CollectingSelectInfo(
			info_name=CURRENT_EXPERIMENT_NAME_KEY,
			info_description=self.required_info_dict()[CURRENT_EXPERIMENT_NAME_KEY],
			choices=experiments,
		)

		duration_info = CollectingCommonInfo(
			info_name=CURRENT_EXPERIMENT_DURATION_KEY,
			info_description=self.required_info_dict()[CURRENT_EXPERIMENT_DURATION_KEY]
		)
		return [select_name_info, duration_info]

	def set_experiment_log_store(self, user_id: str):
		r"""
		Load the user's experiment log storage.

		Args:
			user_id (str): The user_id of a lab member.
		"""
		if self.expr_log_store is None or self.expr_log_store.user_id != user_id:
			self.expr_log_store = ExperimentLog.from_user_id(
				user_id=user_id,
				embed_model=self._embed_model,
			)

	def set_current_experiment(
		self,
		user_id: str,
	) -> FuncOutputWithLog:
		r"""
		This tool is used to set the experiment in progress for the user, through interacting with the user.
		This tool is used ONLY when the user ask for setting his/her experiment in progress,
		or when other tools call this tool.

		Args:
			user_id (str): The user_id of a lab member.

		Returns:
			The tool output and log.
		"""
		# This docstring is used as tool description.
		self.set_experiment_log_store(user_id=user_id)

		expr_list = self.expr_log_store.get_all_experiments()

		if expr_list is None:
			create_tool = CreateNewExperimentLogTool(
				llm=self._llm,
				embed_model=self._embed_model,
			)
			create_output = create_tool.call(user_id=user_id)

			abort = whether_abort_tool(tool_output=create_output)
			if abort is None or abort:
				status = (
					f"The user does not want to continue, and abort the set experiment operation."
					f"This conversation should be ended."
				)
				if abort is None:
					status = f"The tool output of the CreateNewExperimentLogTool is invalid."
				fn_log = OperationOutputLog.construct(
					operation_name=create_tool.metadata.name,
					operation_output=status,
					op_description=status,
					operation_abort=True,
				)
				return FuncOutputWithLog(
					fn_output=status,
					fn_log={"operation_log": fn_log},
				)

		query_str = SET_CURRENT_EXPERIMENT_MSG
		output_log = self.collect_and_authorize(
			user_id=user_id,
			query_str=query_str,
		)
		return output_log

	async def aset_current_experiment(
		self,
		user_id: str,
	) -> FuncOutputWithLog:
		r"""
		This tool is used to set the experiment in progress for the user, through interacting with the user.
		This tool is used ONLY when the user ask for setting his/her experiment in progress,
		or when other tools call this tool.

		Args:
			user_id (str): The user_id of a lab member.

		Returns:
			The tool output and log.
		"""
		self.set_experiment_log_store(user_id=user_id)
		expr_list = self.expr_log_store.get_all_experiments()
		if expr_list is None:
			create_tool = CreateNewExperimentLogTool(
				llm=self._llm,
				embed_model=self._embed_model,
			)
			create_output = await create_tool.acall(user_id=user_id)
			abort = whether_abort_tool(tool_output=create_output)
			if abort is None or abort:
				status = (
					f"The user does not want to continue, and abort the set experiment operation."
					f"This conversation should be ended."
				)
				if abort is None:
					status = f"The tool output of the CreateNewExperimentLogTool is invalid."
				fn_log = OperationOutputLog.construct(
					operation_name=create_tool.metadata.name,
					operation_output=status,
					op_description=status,
					operation_abort=True,
				)
				return FuncOutputWithLog(
					fn_output=status,
					fn_log={"operation_log": fn_log},
				)

		query_str = SET_CURRENT_EXPERIMENT_MSG
		output_log = await self.acollect_and_authorize(
			user_id=user_id,
			query_str=query_str,
		)
		return output_log

labridge.tools.memory.experiment.insert.SetCurrentExperimentTool.aset_current_experiment(user_id) async

This tool is used to set the experiment in progress for the user, through interacting with the user. This tool is used ONLY when the user ask for setting his/her experiment in progress, or when other tools call this tool.

PARAMETER	DESCRIPTION
user_id	The user_id of a lab member. TYPE: str

RETURNS	DESCRIPTION
FuncOutputWithLog	The tool output and log.

Source code in labridge\tools\memory\experiment\insert.py

async def aset_current_experiment(
	self,
	user_id: str,
) -> FuncOutputWithLog:
	r"""
	This tool is used to set the experiment in progress for the user, through interacting with the user.
	This tool is used ONLY when the user ask for setting his/her experiment in progress,
	or when other tools call this tool.

	Args:
		user_id (str): The user_id of a lab member.

	Returns:
		The tool output and log.
	"""
	self.set_experiment_log_store(user_id=user_id)
	expr_list = self.expr_log_store.get_all_experiments()
	if expr_list is None:
		create_tool = CreateNewExperimentLogTool(
			llm=self._llm,
			embed_model=self._embed_model,
		)
		create_output = await create_tool.acall(user_id=user_id)
		abort = whether_abort_tool(tool_output=create_output)
		if abort is None or abort:
			status = (
				f"The user does not want to continue, and abort the set experiment operation."
				f"This conversation should be ended."
			)
			if abort is None:
				status = f"The tool output of the CreateNewExperimentLogTool is invalid."
			fn_log = OperationOutputLog.construct(
				operation_name=create_tool.metadata.name,
				operation_output=status,
				op_description=status,
				operation_abort=True,
			)
			return FuncOutputWithLog(
				fn_output=status,
				fn_log={"operation_log": fn_log},
			)

	query_str = SET_CURRENT_EXPERIMENT_MSG
	output_log = await self.acollect_and_authorize(
		user_id=user_id,
		query_str=query_str,
	)
	return output_log

labridge.tools.memory.experiment.insert.SetCurrentExperimentTool.required_info_dict()

The required information.

RETURNS	DESCRIPTION
Dict[str, str]	Dict[str, str]: key: the information name. value: the information description.

Source code in labridge\tools\memory\experiment\insert.py

def required_info_dict(self) -> Dict[str, str]:
	r"""
	The required information.

	Returns:
		Dict[str, str]:

			- key: the information name.
			- value: the information description.
	"""
	return SET_CURRENT_EXPERIMENT_REQUIRED_INFOS

labridge.tools.memory.experiment.insert.SetCurrentExperimentTool.required_infos()

The required infos.

RETURNS	DESCRIPTION
List[CollectingInfoBase]	List[CollectingInfoBase]: Return the packed info in CollectingInfoBase, such as CollectingCommonInfo and CollectingSelectInfo.

Source code in labridge\tools\memory\experiment\insert.py

def required_infos(self) -> List[CollectingInfoBase]:
	r"""
	The required infos.

	Returns:
		List[CollectingInfoBase]:
			Return the packed info in CollectingInfoBase, such as `CollectingCommonInfo` and `CollectingSelectInfo`.
	"""
	experiments = self.expr_log_store.get_all_experiments_with_description()
	select_name_info = CollectingSelectInfo(
		info_name=CURRENT_EXPERIMENT_NAME_KEY,
		info_description=self.required_info_dict()[CURRENT_EXPERIMENT_NAME_KEY],
		choices=experiments,
	)

	duration_info = CollectingCommonInfo(
		info_name=CURRENT_EXPERIMENT_DURATION_KEY,
		info_description=self.required_info_dict()[CURRENT_EXPERIMENT_DURATION_KEY]
	)
	return [select_name_info, duration_info]

labridge.tools.memory.experiment.insert.SetCurrentExperimentTool.set_current_experiment(user_id)

This tool is used to set the experiment in progress for the user, through interacting with the user. This tool is used ONLY when the user ask for setting his/her experiment in progress, or when other tools call this tool.

PARAMETER	DESCRIPTION
user_id	The user_id of a lab member. TYPE: str

RETURNS	DESCRIPTION
FuncOutputWithLog	The tool output and log.

Source code in labridge\tools\memory\experiment\insert.py

def set_current_experiment(
	self,
	user_id: str,
) -> FuncOutputWithLog:
	r"""
	This tool is used to set the experiment in progress for the user, through interacting with the user.
	This tool is used ONLY when the user ask for setting his/her experiment in progress,
	or when other tools call this tool.

	Args:
		user_id (str): The user_id of a lab member.

	Returns:
		The tool output and log.
	"""
	# This docstring is used as tool description.
	self.set_experiment_log_store(user_id=user_id)

	expr_list = self.expr_log_store.get_all_experiments()

	if expr_list is None:
		create_tool = CreateNewExperimentLogTool(
			llm=self._llm,
			embed_model=self._embed_model,
		)
		create_output = create_tool.call(user_id=user_id)

		abort = whether_abort_tool(tool_output=create_output)
		if abort is None or abort:
			status = (
				f"The user does not want to continue, and abort the set experiment operation."
				f"This conversation should be ended."
			)
			if abort is None:
				status = f"The tool output of the CreateNewExperimentLogTool is invalid."
			fn_log = OperationOutputLog.construct(
				operation_name=create_tool.metadata.name,
				operation_output=status,
				op_description=status,
				operation_abort=True,
			)
			return FuncOutputWithLog(
				fn_output=status,
				fn_log={"operation_log": fn_log},
			)

	query_str = SET_CURRENT_EXPERIMENT_MSG
	output_log = self.collect_and_authorize(
		user_id=user_id,
		query_str=query_str,
	)
	return output_log

labridge.tools.memory.experiment.insert.SetCurrentExperimentTool.set_experiment_log_store(user_id)

Load the user's experiment log storage.

PARAMETER	DESCRIPTION
user_id	The user_id of a lab member. TYPE: str

Source code in labridge\tools\memory\experiment\insert.py

def set_experiment_log_store(self, user_id: str):
	r"""
	Load the user's experiment log storage.

	Args:
		user_id (str): The user_id of a lab member.
	"""
	if self.expr_log_store is None or self.expr_log_store.user_id != user_id:
		self.expr_log_store = ExperimentLog.from_user_id(
			user_id=user_id,
			embed_model=self._embed_model,
		)