跳转至

Retrieve log

labridge.func_modules.memory.experiment.retrieve_log

labridge.func_modules.memory.experiment.retrieve_log.ExperimentLogRetriever

Bases: LogBaseRetriever

This class retrieve in a specific user's experiment logs.

The docstring of the method retrieve or aretrieve are used as tool description of the corresponding retriever tool.

PARAMETER DESCRIPTION
embed_model

The embed model. Defaults to None. If set to None, the Settings.embed_model will be used.

TYPE: BaseEmbedding DEFAULT: None

final_use_context

Whether to add the context nodes of the retrieved nodes to the final results. Defaults to True.

TYPE: bool DEFAULT: True

relevant_top_k

The relevant_top_k log nodes will be selected as the retrieved nodes. Defaults to EXPERIMENT_LOG_RELEVANT_TOP_K.

TYPE: int DEFAULT: None

Source code in labridge\func_modules\memory\experiment\retrieve_log.py 
 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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
class ExperimentLogRetriever(LogBaseRetriever):
	r"""
	This class retrieve in a specific user's experiment logs.

	The docstring of the method `retrieve` or `aretrieve` are used as tool description of the
	corresponding retriever tool.

	Args:
		embed_model (BaseEmbedding): The embed model. Defaults to None.
			If set to None, the Settings.embed_model will be used.
		final_use_context (bool): Whether to add the context nodes of the retrieved nodes to the final results.
			Defaults to True.
		relevant_top_k (int): The `relevant_top_k` log nodes will be selected as the retrieved nodes.
			Defaults to `EXPERIMENT_LOG_RELEVANT_TOP_K`.
	"""
	def __init__(
		self,
		embed_model: BaseEmbedding = None,
		final_use_context: bool = True,
		relevant_top_k: int = None,
	):
		relevant_top_k = relevant_top_k or EXPERIMENT_LOG_RELEVANT_TOP_K
		super().__init__(
			embed_model=embed_model,
			final_use_context=final_use_context,
			relevant_top_k=relevant_top_k,
		)

	def get_memory_vector_index(self) -> VectorStoreIndex:
		r""" Get the vector index. """
		return self.memory.vector_index

	def get_memory_vector_retriever(self) -> VectorIndexRetriever:
		r""" Get the default vector index retriever, with default similarity_top_k and no date filters. """
		memory_retriever = self.memory.vector_index.as_retriever(
			similarity_top_k=self.relevant_top_k,
			filters=None,
		)
		return memory_retriever

	def reset_vector_retriever(self):
		r"""
		Reset the vector index retriever to defaults.
		Specifically, with no date filters and confined node ids.
		"""
		self.memory_vector_retriever._filters = [self._log_node_filter(),]
		self.memory_vector_retriever._node_ids = None

	@dispatcher.span
	def retrieve(
		self,
		item_to_be_retrieved: str,
		memory_id: str,
		start_date: str = None,
		end_date: str = None,
		experiment_name: str = None,
		**kwargs: Any,
	) -> List[NodeWithScore]:
		r"""
		This tool is used to retrieve experiment logs of a user.
		Use this tool to help you to answer questions about experimental records.

		Args:
			item_to_be_retrieved (str): This argument is necessary.
				It denotes things that you want to retrieve in the chat history memory.
			memory_id (str): This argument is necessary.
				It is the user_id of a lab member.
			start_date (str): This argument is optional.
				It denotes the start date in the format 'Year-Month-Day'.
				If both start_date and end_date are specified, only logs which are recorded between the
				start_date and end_date will be retrieved.
			end_date (str): This argument is optional.
				It denotes the end date in the format 'Year-Month-Day'.
			experiment_name (str): This argument is optional.
				It is the name of a specific experiment.
				If it is specified and is valid, only logs of this experiment will be retrieved.
			kwargs: Other arguments will be ignored.

		Returns:
			Retrieved experiment logs.
		"""
		if self.memory is None or self.memory.user_id != memory_id:
			self.memory = ExperimentLog.from_user_id(
				user_id=memory_id,
				embed_model=self.embed_model,
			)
			self.memory_vector_retriever = self.get_memory_vector_retriever()

		self.reset_vector_retriever()

		if experiment_name is None or not self.memory.is_expr_exist(experiment_name):
			retrieve_node_ids = None
		else:
			retrieve_node_ids = self.memory.get_expr_log_node_ids(experiment_name)

		filters = [self._log_node_filter(), ]
		if None not in [start_date, end_date]:
			# get the candidate date list.
			date_list = self._parse_date(start_date_str=start_date, end_date_str=end_date)
			filters.append(
				self.get_date_filter(date_list=date_list),
			)

		metadata_filters = MetadataFilters(filters=filters)

		self.memory_vector_retriever._filters = metadata_filters
		self.memory_vector_retriever._node_ids = retrieve_node_ids

		# TODO: hybrid retrieve: add retrieve experiment.
		log_nodes = self.memory_vector_retriever.retrieve(item_to_be_retrieved)

		if self.final_use_context:
			log_nodes = self._add_context(content_nodes=log_nodes)
		return log_nodes

	@dispatcher.span
	async def aretrieve(
		self,
		item_to_be_retrieved: str,
		memory_id: str,
		start_date: str = None,
		end_date: str = None,
		experiment_name: str = None,
		**kwargs: Any,
	) -> List[NodeWithScore]:
		r"""
		This tool is used to retrieve relevant experiment logs in a certain experiment log memory.

		Args:
			item_to_be_retrieved (str): This argument is necessary.
				It denotes things that you want to retrieve in the chat history memory.
			memory_id (str): This argument is necessary.
				It is the user_id of a lab member.
			start_date (str): This argument is optional.
				It denotes the start date in the format 'Year-Month-Day'.
				If both start_date and end_date are specified, only logs which are recorded between the
				start_date and end_date will be retrieved.
			end_date (str): This argument is optional.
				It denotes the end date in the format 'Year-Month-Day'.
			experiment_name (str): This argument is optional.
				It is the name of a specific experiment.
				If it is specified and is valid, only logs of this experiment will be retrieved.
			kwargs: Other arguments will be ignored.

		Returns:
			Retrieved experiment logs.
		"""
		if self.memory is None or self.memory.user_id != memory_id:
			self.memory = ExperimentLog.from_user_id(
				user_id=memory_id,
				embed_model=self.embed_model,
			)
			self.memory_vector_retriever = self.get_memory_vector_retriever()
		self.reset_vector_retriever()

		if experiment_name is None or not self.memory.is_expr_exist(experiment_name):
			retrieve_node_ids = None
		else:
			retrieve_node_ids = self.memory.get_expr_log_node_ids(experiment_name)

		filters = [self._log_node_filter(), ]
		if None not in [start_date, end_date]:
			# get the candidate date list.
			date_list = self._parse_date(start_date_str=start_date, end_date_str=end_date)
			filters.append(
				self.get_date_filter(date_list=date_list)
			)

		metadata_filters = MetadataFilters(filters=filters)

		self.memory_vector_retriever._filters = metadata_filters
		self.memory_vector_retriever._node_ids = retrieve_node_ids
		log_nodes = await self.memory_vector_retriever.aretrieve(item_to_be_retrieved)

		if self.final_use_context:
			log_nodes = self._add_context(content_nodes=log_nodes)
		return log_nodes

labridge.func_modules.memory.experiment.retrieve_log.ExperimentLogRetriever.aretrieve(item_to_be_retrieved, memory_id, start_date=None, end_date=None, experiment_name=None, **kwargs) async

This tool is used to retrieve relevant experiment logs in a certain experiment log memory.

PARAMETER DESCRIPTION
item_to_be_retrieved

This argument is necessary. It denotes things that you want to retrieve in the chat history memory.

TYPE: str

memory_id

This argument is necessary. It is the user_id of a lab member.

TYPE: str

start_date

This argument is optional. It denotes the start date in the format 'Year-Month-Day'. If both start_date and end_date are specified, only logs which are recorded between the start_date and end_date will be retrieved.

TYPE: str DEFAULT: None

end_date

This argument is optional. It denotes the end date in the format 'Year-Month-Day'.

TYPE: str DEFAULT: None

experiment_name

This argument is optional. It is the name of a specific experiment. If it is specified and is valid, only logs of this experiment will be retrieved.

TYPE: str DEFAULT: None

kwargs

Other arguments will be ignored.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
List[NodeWithScore]

Retrieved experiment logs.
Source code in labridge\func_modules\memory\experiment\retrieve_log.py 
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
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
@dispatcher.span
async def aretrieve(
	self,
	item_to_be_retrieved: str,
	memory_id: str,
	start_date: str = None,
	end_date: str = None,
	experiment_name: str = None,
	**kwargs: Any,
) -> List[NodeWithScore]:
	r"""
	This tool is used to retrieve relevant experiment logs in a certain experiment log memory.

	Args:
		item_to_be_retrieved (str): This argument is necessary.
			It denotes things that you want to retrieve in the chat history memory.
		memory_id (str): This argument is necessary.
			It is the user_id of a lab member.
		start_date (str): This argument is optional.
			It denotes the start date in the format 'Year-Month-Day'.
			If both start_date and end_date are specified, only logs which are recorded between the
			start_date and end_date will be retrieved.
		end_date (str): This argument is optional.
			It denotes the end date in the format 'Year-Month-Day'.
		experiment_name (str): This argument is optional.
			It is the name of a specific experiment.
			If it is specified and is valid, only logs of this experiment will be retrieved.
		kwargs: Other arguments will be ignored.

	Returns:
		Retrieved experiment logs.
	"""
	if self.memory is None or self.memory.user_id != memory_id:
		self.memory = ExperimentLog.from_user_id(
			user_id=memory_id,
			embed_model=self.embed_model,
		)
		self.memory_vector_retriever = self.get_memory_vector_retriever()
	self.reset_vector_retriever()

	if experiment_name is None or not self.memory.is_expr_exist(experiment_name):
		retrieve_node_ids = None
	else:
		retrieve_node_ids = self.memory.get_expr_log_node_ids(experiment_name)

	filters = [self._log_node_filter(), ]
	if None not in [start_date, end_date]:
		# get the candidate date list.
		date_list = self._parse_date(start_date_str=start_date, end_date_str=end_date)
		filters.append(
			self.get_date_filter(date_list=date_list)
		)

	metadata_filters = MetadataFilters(filters=filters)

	self.memory_vector_retriever._filters = metadata_filters
	self.memory_vector_retriever._node_ids = retrieve_node_ids
	log_nodes = await self.memory_vector_retriever.aretrieve(item_to_be_retrieved)

	if self.final_use_context:
		log_nodes = self._add_context(content_nodes=log_nodes)
	return log_nodes

labridge.func_modules.memory.experiment.retrieve_log.ExperimentLogRetriever.get_memory_vector_index()

Get the vector index.

Source code in labridge\func_modules\memory\experiment\retrieve_log.py 
50
51
52
def get_memory_vector_index(self) -> VectorStoreIndex:
	r""" Get the vector index. """
	return self.memory.vector_index

labridge.func_modules.memory.experiment.retrieve_log.ExperimentLogRetriever.get_memory_vector_retriever()

Get the default vector index retriever, with default similarity_top_k and no date filters.

Source code in labridge\func_modules\memory\experiment\retrieve_log.py 
54
55
56
57
58
59
60
def get_memory_vector_retriever(self) -> VectorIndexRetriever:
	r""" Get the default vector index retriever, with default similarity_top_k and no date filters. """
	memory_retriever = self.memory.vector_index.as_retriever(
		similarity_top_k=self.relevant_top_k,
		filters=None,
	)
	return memory_retriever

labridge.func_modules.memory.experiment.retrieve_log.ExperimentLogRetriever.reset_vector_retriever()

Reset the vector index retriever to defaults. Specifically, with no date filters and confined node ids.

Source code in labridge\func_modules\memory\experiment\retrieve_log.py 
62
63
64
65
66
67
68
def reset_vector_retriever(self):
	r"""
	Reset the vector index retriever to defaults.
	Specifically, with no date filters and confined node ids.
	"""
	self.memory_vector_retriever._filters = [self._log_node_filter(),]
	self.memory_vector_retriever._node_ids = None

labridge.func_modules.memory.experiment.retrieve_log.ExperimentLogRetriever.retrieve(item_to_be_retrieved, memory_id, start_date=None, end_date=None, experiment_name=None, **kwargs)

This tool is used to retrieve experiment logs of a user. Use this tool to help you to answer questions about experimental records.

PARAMETER DESCRIPTION
item_to_be_retrieved

This argument is necessary. It denotes things that you want to retrieve in the chat history memory.

TYPE: str

memory_id

This argument is necessary. It is the user_id of a lab member.

TYPE: str

start_date

This argument is optional. It denotes the start date in the format 'Year-Month-Day'. If both start_date and end_date are specified, only logs which are recorded between the start_date and end_date will be retrieved.

TYPE: str DEFAULT: None

end_date

This argument is optional. It denotes the end date in the format 'Year-Month-Day'.

TYPE: str DEFAULT: None

experiment_name

This argument is optional. It is the name of a specific experiment. If it is specified and is valid, only logs of this experiment will be retrieved.

TYPE: str DEFAULT: None

kwargs

Other arguments will be ignored.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
List[NodeWithScore]

Retrieved experiment logs.
Source code in labridge\func_modules\memory\experiment\retrieve_log.py 
 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
@dispatcher.span
def retrieve(
	self,
	item_to_be_retrieved: str,
	memory_id: str,
	start_date: str = None,
	end_date: str = None,
	experiment_name: str = None,
	**kwargs: Any,
) -> List[NodeWithScore]:
	r"""
	This tool is used to retrieve experiment logs of a user.
	Use this tool to help you to answer questions about experimental records.

	Args:
		item_to_be_retrieved (str): This argument is necessary.
			It denotes things that you want to retrieve in the chat history memory.
		memory_id (str): This argument is necessary.
			It is the user_id of a lab member.
		start_date (str): This argument is optional.
			It denotes the start date in the format 'Year-Month-Day'.
			If both start_date and end_date are specified, only logs which are recorded between the
			start_date and end_date will be retrieved.
		end_date (str): This argument is optional.
			It denotes the end date in the format 'Year-Month-Day'.
		experiment_name (str): This argument is optional.
			It is the name of a specific experiment.
			If it is specified and is valid, only logs of this experiment will be retrieved.
		kwargs: Other arguments will be ignored.

	Returns:
		Retrieved experiment logs.
	"""
	if self.memory is None or self.memory.user_id != memory_id:
		self.memory = ExperimentLog.from_user_id(
			user_id=memory_id,
			embed_model=self.embed_model,
		)
		self.memory_vector_retriever = self.get_memory_vector_retriever()

	self.reset_vector_retriever()

	if experiment_name is None or not self.memory.is_expr_exist(experiment_name):
		retrieve_node_ids = None
	else:
		retrieve_node_ids = self.memory.get_expr_log_node_ids(experiment_name)

	filters = [self._log_node_filter(), ]
	if None not in [start_date, end_date]:
		# get the candidate date list.
		date_list = self._parse_date(start_date_str=start_date, end_date_str=end_date)
		filters.append(
			self.get_date_filter(date_list=date_list),
		)

	metadata_filters = MetadataFilters(filters=filters)

	self.memory_vector_retriever._filters = metadata_filters
	self.memory_vector_retriever._node_ids = retrieve_node_ids

	# TODO: hybrid retrieve: add retrieve experiment.
	log_nodes = self.memory_vector_retriever.retrieve(item_to_be_retrieved)

	if self.final_use_context:
		log_nodes = self._add_context(content_nodes=log_nodes)
	return log_nodes