跳转至

Retrieve

labridge.tools.memory.experiment.retrieve

labridge.tools.memory.experiment.retrieve.ExperimentLogRetrieveTool

Bases: RetrieverBaseTool

This tool is used to retrieve the relevant experiment log in the user's experiment log storage. The tool description is set as the docstring of the method retrieve of the retriever.

PARAMETER DESCRIPTION
embed_model

The used embedding model. If not specified, the Setting.embed_model will be used.

TYPE: BaseEmbedding DEFAULT: None

final_use_context

Whether to add the context nodes to the final retrieving results.

TYPE: bool DEFAULT: True

relevant_top_k

The top-k relevant nodes will be used as the retrieved results.

TYPE: int DEFAULT: None

Source code in labridge\tools\memory\experiment\retrieve.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
class ExperimentLogRetrieveTool(RetrieverBaseTool):
	r"""
	This tool is used to retrieve the relevant experiment log in the user's experiment log storage.
	The tool description is set as the docstring of the method `retrieve` of the `retriever`.

	Args:
		embed_model (BaseEmbedding): The used embedding model. If not specified, the `Setting.embed_model` will be used.
		final_use_context (bool): Whether to add the context nodes to the final retrieving results.
		relevant_top_k (int): The top-k relevant nodes will be used as the retrieved results.
	"""
	def __init__(
		self,
		embed_model: BaseEmbedding = None,
		final_use_context: bool = True,
		relevant_top_k: int = None,
	):
		retriever = ExperimentLogRetriever(
			embed_model=embed_model,
			final_use_context=final_use_context,
			relevant_top_k=relevant_top_k,
		)
		super().__init__(
			name=ExperimentLogRetrieveTool.__name__,
			retriever=retriever,
			retrieve_fn=retriever.retrieve,
		)

	def log(self, log_dict: dict) -> ToolLog:
		r"""
		Record the tool log.

		Args:
			log_dict (dict): Including the input keyword arguments and the (output, log) of retrieving.

		Returns:
			The tool log.
		"""
		user_id = log_dict["memory_id"]
		item_to_be_retrieved = log_dict["item_to_be_retrieved"]
		start_date = log_dict.get("start_date", None)
		end_date = log_dict.get("end_date", None)

		ref_infos: List[ExperimentLogRefInfo] = log_dict.get(TOOL_LOG_REF_INFO_KEY)

		log_string = (
			f"Retrieve in the experiment log memory of the user: {user_id}.\n"
			f"retrieve string: {item_to_be_retrieved}\n"
		)
		if None not in [start_date, end_date]:
			log_string += (
				f"start_date: {start_date}\n"
				f"end_date: {end_date}"
			)

		log_to_user = None
		log_to_system = {
			TOOL_OP_DESCRIPTION: log_string,
			TOOL_REFERENCES: [ref_info.dumps() for ref_info in ref_infos],
		}
		return ToolLog(
			log_to_user=log_to_user,
			log_to_system=log_to_system,
			tool_name=self.metadata.name,
		)

	def get_ref_info(self, nodes: List[NodeWithScore]) -> List[ExperimentLogRefInfo]:
		r"""
		Get the reference paper infos

		Returns:
			List[PaperInfo]: The reference paper infos in answering.
		"""
		ref_infos = []
		for node_score in nodes:
			metadata = node_score.node.metadata
			log_str = node_score.node.text
			experiment_name = node_score.node.parent_node.node_id
			attachment_path = metadata.get(EXPERIMENT_LOG_ATTACHMENT_KEY, None)
			date = metadata.get(LOG_DATE_NAME)
			h_m_s = metadata.get(LOG_TIME_NAME)
			log_ref_info = ExperimentLogRefInfo(
				date_time=f"{date} {h_m_s}",
				log_str=log_str,
				attachment_path=attachment_path,
				experiment_name=experiment_name,
			)
			ref_infos.append(log_ref_info)
		return ref_infos

	def _retrieve(self, retrieve_kwargs: dict) -> List[NodeWithScore]:
		r""" Use the retriever to retrieve relevant nodes. """
		nodes = self._retriever.retrieve(**retrieve_kwargs)
		return nodes

	async def _aretrieve(self, retrieve_kwargs: dict) -> List[NodeWithScore]:
		r""" Asynchronously use the retriever to retrieve relevant nodes. """
		nodes = await self._retriever.aretrieve(**retrieve_kwargs)
		return nodes

	def _nodes_to_tool_output(self, nodes: List[NodeWithScore]) -> Tuple[str, dict]:
		r""" output the retrieved contents in a specific format, and the output log. """
		expr_nodes_dict = {}

		ref_infos = self.get_ref_info(nodes=nodes)
		log_dict = {
			TOOL_LOG_REF_INFO_KEY: ref_infos,
		}

		for node in nodes:
			expr_node = node.node.parent_node
			if expr_node is not None:
				if expr_node.node_id not in expr_nodes_dict.keys():
					expr_nodes_dict[expr_node.node_id] = [node]
				else:
					expr_nodes_dict[expr_node.node_id].append(node)

		if expr_nodes_dict:
			msg = "Have retrieved the following experiment logs:"
			contents = [msg]
			for expr_name in expr_nodes_dict.keys():
				msg = f"The following logs are from the experiment: {expr_name}"
				contents.append(msg)
				for idx, node in enumerate(expr_nodes_dict[expr_name]):
					content_str = (
						f"Log {idx + 1}:\n"
						f"{node.node.get_content(metadata_mode=MetadataMode.LLM)}"
					)
					contents.append(content_str.strip())
			output_str = "\n\n".join(contents)
		else:
			output_str = "Have retrieved nothing relevant."
		return output_str, log_dict

labridge.tools.memory.experiment.retrieve.ExperimentLogRetrieveTool.get_ref_info(nodes)

Get the reference paper infos

RETURNS DESCRIPTION
List[ExperimentLogRefInfo]

List[PaperInfo]: The reference paper infos in answering.
Source code in labridge\tools\memory\experiment\retrieve.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def get_ref_info(self, nodes: List[NodeWithScore]) -> List[ExperimentLogRefInfo]:
	r"""
	Get the reference paper infos

	Returns:
		List[PaperInfo]: The reference paper infos in answering.
	"""
	ref_infos = []
	for node_score in nodes:
		metadata = node_score.node.metadata
		log_str = node_score.node.text
		experiment_name = node_score.node.parent_node.node_id
		attachment_path = metadata.get(EXPERIMENT_LOG_ATTACHMENT_KEY, None)
		date = metadata.get(LOG_DATE_NAME)
		h_m_s = metadata.get(LOG_TIME_NAME)
		log_ref_info = ExperimentLogRefInfo(
			date_time=f"{date} {h_m_s}",
			log_str=log_str,
			attachment_path=attachment_path,
			experiment_name=experiment_name,
		)
		ref_infos.append(log_ref_info)
	return ref_infos

labridge.tools.memory.experiment.retrieve.ExperimentLogRetrieveTool.log(log_dict)

Record the tool log.

PARAMETER DESCRIPTION
log_dict

Including the input keyword arguments and the (output, log) of retrieving.

TYPE: dict

RETURNS DESCRIPTION
ToolLog

The tool log.
Source code in labridge\tools\memory\experiment\retrieve.py 
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
def log(self, log_dict: dict) -> ToolLog:
	r"""
	Record the tool log.

	Args:
		log_dict (dict): Including the input keyword arguments and the (output, log) of retrieving.

	Returns:
		The tool log.
	"""
	user_id = log_dict["memory_id"]
	item_to_be_retrieved = log_dict["item_to_be_retrieved"]
	start_date = log_dict.get("start_date", None)
	end_date = log_dict.get("end_date", None)

	ref_infos: List[ExperimentLogRefInfo] = log_dict.get(TOOL_LOG_REF_INFO_KEY)

	log_string = (
		f"Retrieve in the experiment log memory of the user: {user_id}.\n"
		f"retrieve string: {item_to_be_retrieved}\n"
	)
	if None not in [start_date, end_date]:
		log_string += (
			f"start_date: {start_date}\n"
			f"end_date: {end_date}"
		)

	log_to_user = None
	log_to_system = {
		TOOL_OP_DESCRIPTION: log_string,
		TOOL_REFERENCES: [ref_info.dumps() for ref_info in ref_infos],
	}
	return ToolLog(
		log_to_user=log_to_user,
		log_to_system=log_to_system,
		tool_name=self.metadata.name,
	)