Select collector

labridge.interact.collect.collector.select_collector

labridge.interact.collect.collector.select_collector.SelectInfoCollector

Collect SelectInfo from the user. refer to ..types.select_info.CollectingSelectInfo for the detail of SelectInfo.

PARAMETER	DESCRIPTION
llm	The used LLM. TYPE: LLM
required_infos	The required infos. TYPE: List[CollectingInfoBase]

Source code in labridge\interact\collect\collector\select_collector.py

class SelectInfoCollector:
	r"""
	Collect SelectInfo from the user.
	refer to `..types.select_info.CollectingSelectInfo` for the detail of SelectInfo.

	Args:
		llm (LLM): The used LLM.
		required_infos (List[CollectingInfoBase]): The required infos.
	"""
	def __init__(
		self,
		llm: LLM,
		required_infos: List[CollectingInfoBase],
	):
		self._llm = llm
		self._select_infos = self.get_select_infos(required_infos=required_infos)
		self._collect_manager = CollectManager(llm=llm)

	def get_select_infos(
		self,
		required_infos: List[CollectingInfoBase],
	) -> List[CollectingSelectInfo]:
		r"""
		Choose the CollectingSelectInfo from the required_infos.

		Args:
			required_infos (List[CollectingInfoBase]): The required infos.

		Returns:
			List[CollectingSelectInfo]: All required CollectingSelectInfo. They will be collected one by one.
		"""
		select_infos = []
		for info in required_infos:
			if isinstance(info, CollectingSelectInfo):
				select_infos.append(info)
		return select_infos

	def collect_single_info(
		self,
		info: CollectingSelectInfo,
		user_id: str,
		query_str: Optional[str] = None,
	) -> bool:
		r"""
		Collect a SelectInfo from the user.

		Args:
			info (CollectingSelectInfo): The info waiting for the user's selection.
			user_id (str): The user id of a Lab member.

		Returns:
			bool: Whether the user aborts the collecting process.
		"""
		# TODO: send to user:
		query_to_user = self.collecting_query(info=info)
		if query_str:
			query_to_user = "\n".join([query_str, query_to_user])

		print("Assistant: ", query_to_user)

		# TODO: receive from user.
		user_msg = ChatBuffer.test_get_user_text(user_id=user_id)
		user_response = user_msg.user_msg

		abort = self._collect_manager.analyze_whether_abort(user_response=user_response)
		if abort:
			return abort

		predict_kwargs = {
			"prompt": COLLECT_SELECT_INFO_PROMPT,
			CollectPromptKeys.user_response_key: user_response,
		}

		all_choices, all_relevances = [], []
		for batch_info_dict, batch_candidates in info.info_content():
			predict_kwargs.update(batch_info_dict)
			raw_response = self._llm.predict(**predict_kwargs)

			raw_choices, relevances = default_parse_choice_select_answer_fn(raw_response, len(batch_candidates))
			choice_idxs = [choice - 1 for choice in raw_choices]
			batch_choices = [batch_candidates[ci] for ci in choice_idxs]

			all_choices.extend(batch_choices)
			all_relevances.extend(relevances)

		if all_choices:
			zipped_list = list(zip(all_choices, all_relevances))
			sorted_list = sorted(zipped_list, key=lambda x: x[1], reverse=True)
			collected_info, score = sorted_list[0]
			info.update_collected_info(
				collected_info_dict={info.info_name: collected_info}
			)
		return abort

	async def acollect_single_info(
		self,
		info: CollectingSelectInfo,
		user_id: str,
		query_str: Optional[str] = None,
	) -> bool:
		r"""
		Asynchronously collect a SelectInfo from the user.

		Args:
			info (CollectingSelectInfo): The info waiting for the user's selection.
			user_id (str): The user id of a Lab member.

		Returns:
			bool: Whether the user aborts the collecting process.
		"""
		# TODO: send to user:
		query_to_user = self.collecting_query(info=info)
		if query_str:
			query_to_user = "\n".join([query_str, query_to_user])

		ChatBuffer.put_agent_reply(
			user_id=user_id,
			reply_str=query_to_user,
			inner_chat=True,
		)

		# TODO: receive from user.
		user_msg = await ChatBuffer.get_user_msg(user_id=user_id)
		user_response = user_msg.user_msg

		abort = await self._collect_manager.async_analyze_whether_abort(user_response=user_response)
		if abort:
			return abort

		predict_kwargs = {
			"prompt": COLLECT_SELECT_INFO_PROMPT,
			CollectPromptKeys.user_response_key: user_response,
		}

		all_choices, all_relevances = [], []
		for batch_info_dict, batch_candidates in info.info_content():
			predict_kwargs.update(batch_info_dict)
			raw_response = await self._llm.apredict(**predict_kwargs)

			raw_choices, relevances = default_parse_choice_select_answer_fn(raw_response, len(batch_candidates))
			choice_idxs = [choice - 1 for choice in raw_choices]
			batch_choices = [batch_candidates[ci] for ci in choice_idxs]

			all_choices.extend(batch_choices)
			all_relevances.extend(relevances)

		if all_choices:
			zipped_list = list(zip(all_choices, all_relevances))
			sorted_list = sorted(zipped_list, key=lambda x: x[1], reverse=True)
			collected_info, score = sorted_list[0]
			info.update_collected_info(
				collected_info_dict={info.info_name: collected_info}
			)
		return abort

	def collect(
		self,
		user_id: str,
		query_str: Optional[str] = None,
	) -> bool:
		r"""
		Collect all SelectInfo.

		Returns:
			bool: Whether the user aborts the collecting process.
		"""
		for info in self._select_infos:
			abort = self.collect_single_info(info=info, user_id=user_id, query_str=query_str)
			if abort:
				return True
		return False

	async def acollect(
		self,
		user_id: str,
		query_str: Optional[str] = None,
	) -> bool:
		r"""
		Asynchronously collect all SelectInfo.

		Returns:
			bool: Whether the user aborts the collecting process.
		"""
		for info in self._select_infos:
			abort = await self.acollect_single_info(info=info, user_id=user_id, query_str=query_str)
			if abort:
				return True
		return False

	def collecting_query(self, info: CollectingSelectInfo) -> str:
		r"""
		This query will be sent to user to collect rest Common information

		Args:
			info (CollectingSelectInfo): The SelectInfo to be collected.

		Returns:
			The query.
		"""
		if info is None:
			return ""
		query_to_user = f"{COLLECT_SELECT_INFO_QUERY}\n"
		for key in info.collecting_keys:
			query_to_user += f"\t{key}\n"
		query_to_user += "Candidates:\n"
		for choice in info.candidates:
			query_to_user += f"\t{choice}\n"
		return query_to_user

	def modify_single_info(self, user_response: str,  info: CollectingSelectInfo):
		r"""
		Modify a collected SelectInfo according to the user's comment.

		Args:
			user_response (str): The user's comment.
			info (CollectingSelectInfo): The collected SelectInfo.

		Returns:
			None
		"""
		predict_kwargs = {
			"prompt": MODIFY_SELECT_INFO_PROMPT,
			ModifyPromptKeys.user_comment_key: user_response,
		}

		all_choices, all_possibilities = [], []
		for batch_info_dict, batch_candidates in info.modify_info_content():
			predict_kwargs.update(batch_info_dict)
			raw_response = self._llm.predict(**predict_kwargs)

			raw_choices, possibilities = default_parse_choice_select_answer_fn(raw_response, len(batch_candidates))
			choice_idxs = [choice - 1 for choice in raw_choices]
			batch_choices = [batch_candidates[ci] for ci in choice_idxs]

			all_choices.extend(batch_choices)
			all_possibilities.extend(possibilities)

		zipped_list = list(zip(all_choices, all_possibilities))
		sorted_list = sorted(zipped_list, key=lambda x: x[1], reverse=True)
		if sorted_list:
			collected_info, score = sorted_list[0]
			if score >= SELECT_MIN_SCORE:
				info.update_collected_info(
					collected_info_dict={info.info_name: collected_info}
				)

	async def amodify_single_info(self, user_response: str,  info: CollectingSelectInfo):
		r"""
		Asynchronously modify a collected SelectInfo according to the user's comment.

		Args:
			user_response (str): The user's comment.
			info (CollectingSelectInfo): The collected SelectInfo.

		Returns:
			None
		"""
		predict_kwargs = {
			"prompt": MODIFY_SELECT_INFO_PROMPT,
			ModifyPromptKeys.user_comment_key: user_response,
		}

		all_choices, all_possibilities = [], []
		for batch_info_dict, batch_candidates in info.modify_info_content():
			predict_kwargs.update(batch_info_dict)
			raw_response = await self._llm.apredict(**predict_kwargs)

			raw_choices, possibilities = default_parse_choice_select_answer_fn(raw_response, len(batch_candidates))
			choice_idxs = [choice - 1 for choice in raw_choices]
			batch_choices = [batch_candidates[ci] for ci in choice_idxs]

			all_choices.extend(batch_choices)
			all_possibilities.extend(possibilities)

		zipped_list = list(zip(all_choices, all_possibilities))
		sorted_list = sorted(zipped_list, key=lambda x: x[1], reverse=True)
		if sorted_list:
			collected_info, score = sorted_list[0]
			if score >= SELECT_MIN_SCORE:
				info.update_collected_info(
					collected_info_dict={info.info_name: collected_info}
				)

	def modify(self, user_id: str) -> Tuple[bool, bool]:
		r"""
		Modify the collected SelectInfo according to the user's comment.

		Returns:
			Tuple[str, str]:
				- doing_modify: Whether the user thinks the collected information need modification.
				- abort: Whether the user aborts the collection process.
		"""
		if len(self._select_infos) < 1:
			return False, False

		doing_modify = True
		abort = False
		while doing_modify and not abort:
			query_to_user =self._collect_manager.verify_query(collected_info_dict=self.collected_infos)
			# TODO: send the message to the user.
			print(query_to_user)

			# TODO: receive the message from the user.
			user_msg = ChatBuffer.test_get_user_text(user_id=user_id)
			user_response = user_msg.user_msg
			abort = self._collect_manager.analyze_whether_abort(user_response=user_response)
			if abort:
				break
			doing_modify = self._collect_manager.analyze_whether_modify(
				user_response=user_response,
				collected_info_dict=self.collected_infos
			)
			if doing_modify:
				self.single_modify(user_response=user_response)
		return doing_modify, abort

	async def amodify(self, user_id: str) -> Tuple[bool, bool]:
		r"""
		Asynchronously modify the collected SelectInfo according to the user's comment.

		Returns:
			Tuple[str, str]:
				- doing_modify: Whether the user thinks the collected information need modification.
				- abort: Whether the user aborts the collection process.
		"""
		if len(self._select_infos) < 1:
			return False, False

		doing_modify = True
		abort = False
		while doing_modify and not abort:
			query_to_user =self._collect_manager.verify_query(collected_info_dict=self.collected_infos)
			# TODO: send the message to the user.
			ChatBuffer.put_agent_reply(
				user_id=user_id,
				reply_str=query_to_user,
				inner_chat=True,
			)

			# TODO: receive the message from the user.
			user_msg = await ChatBuffer.get_user_msg(user_id=user_id)
			user_response = user_msg.user_msg
			abort = await self._collect_manager.async_analyze_whether_abort(user_response=user_response)
			if abort:
				break
			doing_modify = await self._collect_manager.async_analyze_whether_modify(
				user_response=user_response,
				collected_info_dict=self.collected_infos
			)
			if doing_modify:
				await self.asingle_modify(user_response=user_response)
		return doing_modify, abort


	def single_modify(self, user_response: str):
		r""" Modify """
		for info in self._select_infos:
			self.modify_single_info(user_response=user_response, info=info)

	async def asingle_modify(self, user_response: str):
		r""" Asynchronously modify """
		for info in self._select_infos:
			await self.amodify_single_info(user_response=user_response, info=info)

	@property
	def collecting_keys(self) -> List[str]:
		r"""
		The SelectInfo to be collected currently.
		"""
		collecting = []
		for info in self._select_infos:
			collecting.extend(info.collecting_keys)
		return collecting

	@property
	def collected_infos(self) -> Dict[str, str]:
		r""" The Collected SelectInfo. """
		infos = {}
		for info in self._select_infos:
			infos.update(info.collected_infos)
		return infos

	@property
	def collected(self):
		r""" Whether all SelectInfo are collected or not. """
		for info in self._select_infos:
			if not info.collected:
				return False
		return True

labridge.interact.collect.collector.select_collector.SelectInfoCollector.collected property

Whether all SelectInfo are collected or not.

labridge.interact.collect.collector.select_collector.SelectInfoCollector.collected_infos: Dict[str, str] property

The Collected SelectInfo.

labridge.interact.collect.collector.select_collector.SelectInfoCollector.collecting_keys: List[str] property

The SelectInfo to be collected currently.

labridge.interact.collect.collector.select_collector.SelectInfoCollector.acollect(user_id, query_str=None) async

Asynchronously collect all SelectInfo.

RETURNS	DESCRIPTION
bool	Whether the user aborts the collecting process. TYPE: bool

Source code in labridge\interact\collect\collector\select_collector.py

async def acollect(
	self,
	user_id: str,
	query_str: Optional[str] = None,
) -> bool:
	r"""
	Asynchronously collect all SelectInfo.

	Returns:
		bool: Whether the user aborts the collecting process.
	"""
	for info in self._select_infos:
		abort = await self.acollect_single_info(info=info, user_id=user_id, query_str=query_str)
		if abort:
			return True
	return False

labridge.interact.collect.collector.select_collector.SelectInfoCollector.acollect_single_info(info, user_id, query_str=None) async

Asynchronously collect a SelectInfo from the user.

PARAMETER	DESCRIPTION
info	The info waiting for the user's selection. TYPE: CollectingSelectInfo
user_id	The user id of a Lab member. TYPE: str

RETURNS	DESCRIPTION
bool	Whether the user aborts the collecting process. TYPE: bool

Source code in labridge\interact\collect\collector\select_collector.py

async def acollect_single_info(
	self,
	info: CollectingSelectInfo,
	user_id: str,
	query_str: Optional[str] = None,
) -> bool:
	r"""
	Asynchronously collect a SelectInfo from the user.

	Args:
		info (CollectingSelectInfo): The info waiting for the user's selection.
		user_id (str): The user id of a Lab member.

	Returns:
		bool: Whether the user aborts the collecting process.
	"""
	# TODO: send to user:
	query_to_user = self.collecting_query(info=info)
	if query_str:
		query_to_user = "\n".join([query_str, query_to_user])

	ChatBuffer.put_agent_reply(
		user_id=user_id,
		reply_str=query_to_user,
		inner_chat=True,
	)

	# TODO: receive from user.
	user_msg = await ChatBuffer.get_user_msg(user_id=user_id)
	user_response = user_msg.user_msg

	abort = await self._collect_manager.async_analyze_whether_abort(user_response=user_response)
	if abort:
		return abort

	predict_kwargs = {
		"prompt": COLLECT_SELECT_INFO_PROMPT,
		CollectPromptKeys.user_response_key: user_response,
	}

	all_choices, all_relevances = [], []
	for batch_info_dict, batch_candidates in info.info_content():
		predict_kwargs.update(batch_info_dict)
		raw_response = await self._llm.apredict(**predict_kwargs)

		raw_choices, relevances = default_parse_choice_select_answer_fn(raw_response, len(batch_candidates))
		choice_idxs = [choice - 1 for choice in raw_choices]
		batch_choices = [batch_candidates[ci] for ci in choice_idxs]

		all_choices.extend(batch_choices)
		all_relevances.extend(relevances)

	if all_choices:
		zipped_list = list(zip(all_choices, all_relevances))
		sorted_list = sorted(zipped_list, key=lambda x: x[1], reverse=True)
		collected_info, score = sorted_list[0]
		info.update_collected_info(
			collected_info_dict={info.info_name: collected_info}
		)
	return abort

labridge.interact.collect.collector.select_collector.SelectInfoCollector.amodify(user_id) async

Asynchronously modify the collected SelectInfo according to the user's comment.

RETURNS	DESCRIPTION
Tuple[bool, bool]	Tuple[str, str]: - doing_modify: Whether the user thinks the collected information need modification. - abort: Whether the user aborts the collection process.

Source code in labridge\interact\collect\collector\select_collector.py

async def amodify(self, user_id: str) -> Tuple[bool, bool]:
	r"""
	Asynchronously modify the collected SelectInfo according to the user's comment.

	Returns:
		Tuple[str, str]:
			- doing_modify: Whether the user thinks the collected information need modification.
			- abort: Whether the user aborts the collection process.
	"""
	if len(self._select_infos) < 1:
		return False, False

	doing_modify = True
	abort = False
	while doing_modify and not abort:
		query_to_user =self._collect_manager.verify_query(collected_info_dict=self.collected_infos)
		# TODO: send the message to the user.
		ChatBuffer.put_agent_reply(
			user_id=user_id,
			reply_str=query_to_user,
			inner_chat=True,
		)

		# TODO: receive the message from the user.
		user_msg = await ChatBuffer.get_user_msg(user_id=user_id)
		user_response = user_msg.user_msg
		abort = await self._collect_manager.async_analyze_whether_abort(user_response=user_response)
		if abort:
			break
		doing_modify = await self._collect_manager.async_analyze_whether_modify(
			user_response=user_response,
			collected_info_dict=self.collected_infos
		)
		if doing_modify:
			await self.asingle_modify(user_response=user_response)
	return doing_modify, abort

labridge.interact.collect.collector.select_collector.SelectInfoCollector.amodify_single_info(user_response, info) async

Asynchronously modify a collected SelectInfo according to the user's comment.

PARAMETER	DESCRIPTION
user_response	The user's comment. TYPE: str
info	The collected SelectInfo. TYPE: CollectingSelectInfo

RETURNS	DESCRIPTION
	None

Source code in labridge\interact\collect\collector\select_collector.py

async def amodify_single_info(self, user_response: str,  info: CollectingSelectInfo):
	r"""
	Asynchronously modify a collected SelectInfo according to the user's comment.

	Args:
		user_response (str): The user's comment.
		info (CollectingSelectInfo): The collected SelectInfo.

	Returns:
		None
	"""
	predict_kwargs = {
		"prompt": MODIFY_SELECT_INFO_PROMPT,
		ModifyPromptKeys.user_comment_key: user_response,
	}

	all_choices, all_possibilities = [], []
	for batch_info_dict, batch_candidates in info.modify_info_content():
		predict_kwargs.update(batch_info_dict)
		raw_response = await self._llm.apredict(**predict_kwargs)

		raw_choices, possibilities = default_parse_choice_select_answer_fn(raw_response, len(batch_candidates))
		choice_idxs = [choice - 1 for choice in raw_choices]
		batch_choices = [batch_candidates[ci] for ci in choice_idxs]

		all_choices.extend(batch_choices)
		all_possibilities.extend(possibilities)

	zipped_list = list(zip(all_choices, all_possibilities))
	sorted_list = sorted(zipped_list, key=lambda x: x[1], reverse=True)
	if sorted_list:
		collected_info, score = sorted_list[0]
		if score >= SELECT_MIN_SCORE:
			info.update_collected_info(
				collected_info_dict={info.info_name: collected_info}
			)

labridge.interact.collect.collector.select_collector.SelectInfoCollector.asingle_modify(user_response) async

Asynchronously modify

Source code in labridge\interact\collect\collector\select_collector.py

async def asingle_modify(self, user_response: str):
	r""" Asynchronously modify """
	for info in self._select_infos:
		await self.amodify_single_info(user_response=user_response, info=info)

labridge.interact.collect.collector.select_collector.SelectInfoCollector.collect(user_id, query_str=None)

Collect all SelectInfo.

RETURNS	DESCRIPTION
bool	Whether the user aborts the collecting process. TYPE: bool

Source code in labridge\interact\collect\collector\select_collector.py

def collect(
	self,
	user_id: str,
	query_str: Optional[str] = None,
) -> bool:
	r"""
	Collect all SelectInfo.

	Returns:
		bool: Whether the user aborts the collecting process.
	"""
	for info in self._select_infos:
		abort = self.collect_single_info(info=info, user_id=user_id, query_str=query_str)
		if abort:
			return True
	return False

labridge.interact.collect.collector.select_collector.SelectInfoCollector.collect_single_info(info, user_id, query_str=None)

Collect a SelectInfo from the user.

PARAMETER	DESCRIPTION
info	The info waiting for the user's selection. TYPE: CollectingSelectInfo
user_id	The user id of a Lab member. TYPE: str

RETURNS	DESCRIPTION
bool	Whether the user aborts the collecting process. TYPE: bool

Source code in labridge\interact\collect\collector\select_collector.py

def collect_single_info(
	self,
	info: CollectingSelectInfo,
	user_id: str,
	query_str: Optional[str] = None,
) -> bool:
	r"""
	Collect a SelectInfo from the user.

	Args:
		info (CollectingSelectInfo): The info waiting for the user's selection.
		user_id (str): The user id of a Lab member.

	Returns:
		bool: Whether the user aborts the collecting process.
	"""
	# TODO: send to user:
	query_to_user = self.collecting_query(info=info)
	if query_str:
		query_to_user = "\n".join([query_str, query_to_user])

	print("Assistant: ", query_to_user)

	# TODO: receive from user.
	user_msg = ChatBuffer.test_get_user_text(user_id=user_id)
	user_response = user_msg.user_msg

	abort = self._collect_manager.analyze_whether_abort(user_response=user_response)
	if abort:
		return abort

	predict_kwargs = {
		"prompt": COLLECT_SELECT_INFO_PROMPT,
		CollectPromptKeys.user_response_key: user_response,
	}

	all_choices, all_relevances = [], []
	for batch_info_dict, batch_candidates in info.info_content():
		predict_kwargs.update(batch_info_dict)
		raw_response = self._llm.predict(**predict_kwargs)

		raw_choices, relevances = default_parse_choice_select_answer_fn(raw_response, len(batch_candidates))
		choice_idxs = [choice - 1 for choice in raw_choices]
		batch_choices = [batch_candidates[ci] for ci in choice_idxs]

		all_choices.extend(batch_choices)
		all_relevances.extend(relevances)

	if all_choices:
		zipped_list = list(zip(all_choices, all_relevances))
		sorted_list = sorted(zipped_list, key=lambda x: x[1], reverse=True)
		collected_info, score = sorted_list[0]
		info.update_collected_info(
			collected_info_dict={info.info_name: collected_info}
		)
	return abort

labridge.interact.collect.collector.select_collector.SelectInfoCollector.collecting_query(info)

This query will be sent to user to collect rest Common information

PARAMETER	DESCRIPTION
info	The SelectInfo to be collected. TYPE: CollectingSelectInfo

RETURNS	DESCRIPTION
str	The query.

Source code in labridge\interact\collect\collector\select_collector.py

def collecting_query(self, info: CollectingSelectInfo) -> str:
	r"""
	This query will be sent to user to collect rest Common information

	Args:
		info (CollectingSelectInfo): The SelectInfo to be collected.

	Returns:
		The query.
	"""
	if info is None:
		return ""
	query_to_user = f"{COLLECT_SELECT_INFO_QUERY}\n"
	for key in info.collecting_keys:
		query_to_user += f"\t{key}\n"
	query_to_user += "Candidates:\n"
	for choice in info.candidates:
		query_to_user += f"\t{choice}\n"
	return query_to_user

labridge.interact.collect.collector.select_collector.SelectInfoCollector.get_select_infos(required_infos)

Choose the CollectingSelectInfo from the required_infos.

PARAMETER	DESCRIPTION
required_infos	The required infos. TYPE: List[CollectingInfoBase]

RETURNS	DESCRIPTION
List[CollectingSelectInfo]	List[CollectingSelectInfo]: All required CollectingSelectInfo. They will be collected one by one.

Source code in labridge\interact\collect\collector\select_collector.py

def get_select_infos(
	self,
	required_infos: List[CollectingInfoBase],
) -> List[CollectingSelectInfo]:
	r"""
	Choose the CollectingSelectInfo from the required_infos.

	Args:
		required_infos (List[CollectingInfoBase]): The required infos.

	Returns:
		List[CollectingSelectInfo]: All required CollectingSelectInfo. They will be collected one by one.
	"""
	select_infos = []
	for info in required_infos:
		if isinstance(info, CollectingSelectInfo):
			select_infos.append(info)
	return select_infos

labridge.interact.collect.collector.select_collector.SelectInfoCollector.modify(user_id)

Modify the collected SelectInfo according to the user's comment.

RETURNS	DESCRIPTION
Tuple[bool, bool]	Tuple[str, str]: - doing_modify: Whether the user thinks the collected information need modification. - abort: Whether the user aborts the collection process.

Source code in labridge\interact\collect\collector\select_collector.py

def modify(self, user_id: str) -> Tuple[bool, bool]:
	r"""
	Modify the collected SelectInfo according to the user's comment.

	Returns:
		Tuple[str, str]:
			- doing_modify: Whether the user thinks the collected information need modification.
			- abort: Whether the user aborts the collection process.
	"""
	if len(self._select_infos) < 1:
		return False, False

	doing_modify = True
	abort = False
	while doing_modify and not abort:
		query_to_user =self._collect_manager.verify_query(collected_info_dict=self.collected_infos)
		# TODO: send the message to the user.
		print(query_to_user)

		# TODO: receive the message from the user.
		user_msg = ChatBuffer.test_get_user_text(user_id=user_id)
		user_response = user_msg.user_msg
		abort = self._collect_manager.analyze_whether_abort(user_response=user_response)
		if abort:
			break
		doing_modify = self._collect_manager.analyze_whether_modify(
			user_response=user_response,
			collected_info_dict=self.collected_infos
		)
		if doing_modify:
			self.single_modify(user_response=user_response)
	return doing_modify, abort

labridge.interact.collect.collector.select_collector.SelectInfoCollector.modify_single_info(user_response, info)

Modify a collected SelectInfo according to the user's comment.

PARAMETER	DESCRIPTION
user_response	The user's comment. TYPE: str
info	The collected SelectInfo. TYPE: CollectingSelectInfo

RETURNS	DESCRIPTION
	None

Source code in labridge\interact\collect\collector\select_collector.py

def modify_single_info(self, user_response: str,  info: CollectingSelectInfo):
	r"""
	Modify a collected SelectInfo according to the user's comment.

	Args:
		user_response (str): The user's comment.
		info (CollectingSelectInfo): The collected SelectInfo.

	Returns:
		None
	"""
	predict_kwargs = {
		"prompt": MODIFY_SELECT_INFO_PROMPT,
		ModifyPromptKeys.user_comment_key: user_response,
	}

	all_choices, all_possibilities = [], []
	for batch_info_dict, batch_candidates in info.modify_info_content():
		predict_kwargs.update(batch_info_dict)
		raw_response = self._llm.predict(**predict_kwargs)

		raw_choices, possibilities = default_parse_choice_select_answer_fn(raw_response, len(batch_candidates))
		choice_idxs = [choice - 1 for choice in raw_choices]
		batch_choices = [batch_candidates[ci] for ci in choice_idxs]

		all_choices.extend(batch_choices)
		all_possibilities.extend(possibilities)

	zipped_list = list(zip(all_choices, all_possibilities))
	sorted_list = sorted(zipped_list, key=lambda x: x[1], reverse=True)
	if sorted_list:
		collected_info, score = sorted_list[0]
		if score >= SELECT_MIN_SCORE:
			info.update_collected_info(
				collected_info_dict={info.info_name: collected_info}
			)

labridge.interact.collect.collector.select_collector.SelectInfoCollector.single_modify(user_response)

Modify

Source code in labridge\interact\collect\collector\select_collector.py

def single_modify(self, user_response: str):
	r""" Modify """
	for info in self._select_infos:
		self.modify_single_info(user_response=user_response, info=info)