Common collector

labridge.interact.collect.collector.common_collector

labridge.interact.collect.collector.common_collector.CommonInfoCollector

Collect CommonInfo from the user. refer to ..types.common_info.CollectingCommonInfo for the detail of CommonInfo.

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

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

class CommonInfoCollector:
	r"""
	Collect CommonInfo from the user.
	refer to `..types.common_info.CollectingCommonInfo` for the detail of CommonInfo.

	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._common_infos = self.get_common_infos(required_infos=required_infos)
		self._collect_manager = CollectManager(llm=llm)

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

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

		Returns:
			CollectingCommonInfo: All required CollectingCommonInfo are aggregated in a  CollectingCommonInfo.
				If no CollectingCommonInfo required, return None.
		"""
		common_info = None
		for info in required_infos:
			if isinstance(info, CollectingCommonInfo):
				if common_info is None:
					common_info = copy.deepcopy(info)
				else:
					common_info.insert_info(info)
		return common_info

	@property
	def collecting_keys(self) -> Optional[List[str]]:
		r""" The information to be collected currently. """
		if self._common_infos is None:
			return None

		return self._common_infos.collecting_keys

	@property
	def collected_infos(self) -> Optional[Dict[str, str]]:
		r""" The Collected Common information. """
		if self._common_infos is None:
			return None

		return self._common_infos.collected_infos

	@property
	def collected(self):
		r""" Whether all Common information are collected or not. """
		if self._common_infos is None:
			return True

		return self._common_infos.collected

	@property
	def collecting_query(self) -> str:
		r""" This query will be sent to user to collect rest Common information. """
		query_to_user = f"{COLLECT_COMMON_INFO_QUERY}\n"
		for key in self.collecting_keys:
			query_to_user += f"\t{key}\n"
		return query_to_user

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

		Returns:
			bool: Whether the user aborts the collecting process.
		"""
		if self._common_infos is None:
			return False

		query_to_user = self.collecting_query
		if query_str:
			query_to_user = "\n".join([query_str, query_to_user])

		# 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:
			return abort

		for batch_info_dict in self._common_infos.info_content():
			predict_kwargs = {
				"prompt": COLLECT_COMMON_INFO_PROMPT,
				CollectPromptKeys.user_response_key: user_response,
			}
			predict_kwargs.update(batch_info_dict)
			extract_info = self._llm.predict(**predict_kwargs)
			new_info_dict = parse_common_collected_info(
				extract_info=extract_info,
				info_keys=self._common_infos.collecting_keys,
			)
			self._common_infos.update_collected_info(collected_info_dict=new_info_dict)
		return abort

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

		Returns:
			bool: Whether the user aborts the collecting process.
		"""
		if self._common_infos is None:
			return False

		query_to_user = self.collecting_query
		if query_str:
			query_to_user = "\n".join([query_str, query_to_user])

		# 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:
			return abort

		for batch_info_dict in self._common_infos.info_content():
			predict_kwargs = {
				"prompt": COLLECT_COMMON_INFO_PROMPT,
				CollectPromptKeys.user_response_key: user_response,
			}
			predict_kwargs.update(batch_info_dict)
			extract_info = await self._llm.apredict(**predict_kwargs)
			new_info_dict = parse_common_collected_info(
				extract_info=extract_info,
				info_keys=self._common_infos.collecting_keys,
			)
			self._common_infos.update_collected_info(collected_info_dict=new_info_dict)
		return abort

	def modify(self, user_id: str) -> Tuple[bool, bool]:
		r"""
		Modify the collected information 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 self._common_infos is None:
			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 information 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 self._common_infos is None:
			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 the collected information according to the user's comment.

		Args:
			user_response (str): The user's comment.

		Returns:
			None
		"""
		for batch_info_dict in self._common_infos.modify_info_content():
			predict_kwargs = {
				"prompt": MODIFY_COMMON_INFO_PROMPT,
				ModifyPromptKeys.user_comment_key: user_response,
			}
			predict_kwargs.update(batch_info_dict)
			extract_info = self._llm.predict(**predict_kwargs)
			new_info_dict = parse_common_collected_info(
				extract_info=extract_info,
				info_keys=list(self._common_infos.required_infos.keys()),
			)
			print("modified: ", new_info_dict)
			self._common_infos.update_collected_info(collected_info_dict=new_info_dict)

	async def asingle_modify(self, user_response: str):
		r"""
		Asynchronously modify the collected information according to the user's comment.

		Args:
			user_response (str): The user's comment.

		Returns:
			None
		"""
		for batch_info_dict in self._common_infos.modify_info_content():
			predict_kwargs = {
				"prompt": MODIFY_COMMON_INFO_PROMPT,
				ModifyPromptKeys.user_comment_key: user_response,
			}
			predict_kwargs.update(batch_info_dict)
			extract_info = await self._llm.apredict(**predict_kwargs)
			new_info_dict = parse_common_collected_info(
				extract_info=extract_info,
				info_keys=list(self._common_infos.required_infos.keys()),
			)
			print("modified: ", new_info_dict)
			self._common_infos.update_collected_info(collected_info_dict=new_info_dict)

labridge.interact.collect.collector.common_collector.CommonInfoCollector.collected property

Whether all Common information are collected or not.

labridge.interact.collect.collector.common_collector.CommonInfoCollector.collected_infos: Optional[Dict[str, str]] property

The Collected Common information.

labridge.interact.collect.collector.common_collector.CommonInfoCollector.collecting_keys: Optional[List[str]] property

The information to be collected currently.

labridge.interact.collect.collector.common_collector.CommonInfoCollector.collecting_query: str property

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

labridge.interact.collect.collector.common_collector.CommonInfoCollector.acollect(user_id, query_str=None) async

Asynchronously collect the Common information.

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

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

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

	Returns:
		bool: Whether the user aborts the collecting process.
	"""
	if self._common_infos is None:
		return False

	query_to_user = self.collecting_query
	if query_str:
		query_to_user = "\n".join([query_str, query_to_user])

	# 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:
		return abort

	for batch_info_dict in self._common_infos.info_content():
		predict_kwargs = {
			"prompt": COLLECT_COMMON_INFO_PROMPT,
			CollectPromptKeys.user_response_key: user_response,
		}
		predict_kwargs.update(batch_info_dict)
		extract_info = await self._llm.apredict(**predict_kwargs)
		new_info_dict = parse_common_collected_info(
			extract_info=extract_info,
			info_keys=self._common_infos.collecting_keys,
		)
		self._common_infos.update_collected_info(collected_info_dict=new_info_dict)
	return abort

labridge.interact.collect.collector.common_collector.CommonInfoCollector.amodify(user_id) async

Asynchronously modify the collected information 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\common_collector.py

async def amodify(self, user_id: str) -> Tuple[bool, bool]:
	r"""
	Asynchronously modify the collected information 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 self._common_infos is None:
		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.common_collector.CommonInfoCollector.asingle_modify(user_response) async

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

PARAMETER	DESCRIPTION
user_response	The user's comment. TYPE: str

RETURNS	DESCRIPTION
	None

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

async def asingle_modify(self, user_response: str):
	r"""
	Asynchronously modify the collected information according to the user's comment.

	Args:
		user_response (str): The user's comment.

	Returns:
		None
	"""
	for batch_info_dict in self._common_infos.modify_info_content():
		predict_kwargs = {
			"prompt": MODIFY_COMMON_INFO_PROMPT,
			ModifyPromptKeys.user_comment_key: user_response,
		}
		predict_kwargs.update(batch_info_dict)
		extract_info = await self._llm.apredict(**predict_kwargs)
		new_info_dict = parse_common_collected_info(
			extract_info=extract_info,
			info_keys=list(self._common_infos.required_infos.keys()),
		)
		print("modified: ", new_info_dict)
		self._common_infos.update_collected_info(collected_info_dict=new_info_dict)

labridge.interact.collect.collector.common_collector.CommonInfoCollector.collect(user_id, query_str=None)

Collect the Common information.

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

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

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

	Returns:
		bool: Whether the user aborts the collecting process.
	"""
	if self._common_infos is None:
		return False

	query_to_user = self.collecting_query
	if query_str:
		query_to_user = "\n".join([query_str, query_to_user])

	# 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:
		return abort

	for batch_info_dict in self._common_infos.info_content():
		predict_kwargs = {
			"prompt": COLLECT_COMMON_INFO_PROMPT,
			CollectPromptKeys.user_response_key: user_response,
		}
		predict_kwargs.update(batch_info_dict)
		extract_info = self._llm.predict(**predict_kwargs)
		new_info_dict = parse_common_collected_info(
			extract_info=extract_info,
			info_keys=self._common_infos.collecting_keys,
		)
		self._common_infos.update_collected_info(collected_info_dict=new_info_dict)
	return abort

labridge.interact.collect.collector.common_collector.CommonInfoCollector.get_common_infos(required_infos)

Choose the CollectingCommonInfo from the required_infos.

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

RETURNS	DESCRIPTION
CollectingCommonInfo	All required CollectingCommonInfo are aggregated in a CollectingCommonInfo. If no CollectingCommonInfo required, return None. TYPE: CollectingCommonInfo

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

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

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

	Returns:
		CollectingCommonInfo: All required CollectingCommonInfo are aggregated in a  CollectingCommonInfo.
			If no CollectingCommonInfo required, return None.
	"""
	common_info = None
	for info in required_infos:
		if isinstance(info, CollectingCommonInfo):
			if common_info is None:
				common_info = copy.deepcopy(info)
			else:
				common_info.insert_info(info)
	return common_info

labridge.interact.collect.collector.common_collector.CommonInfoCollector.modify(user_id)

Modify the collected information 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\common_collector.py

def modify(self, user_id: str) -> Tuple[bool, bool]:
	r"""
	Modify the collected information 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 self._common_infos is None:
		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.common_collector.CommonInfoCollector.single_modify(user_response)

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

PARAMETER	DESCRIPTION
user_response	The user's comment. TYPE: str

RETURNS	DESCRIPTION
	None

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

def single_modify(self, user_response: str):
	r"""
	Modify the collected information according to the user's comment.

	Args:
		user_response (str): The user's comment.

	Returns:
		None
	"""
	for batch_info_dict in self._common_infos.modify_info_content():
		predict_kwargs = {
			"prompt": MODIFY_COMMON_INFO_PROMPT,
			ModifyPromptKeys.user_comment_key: user_response,
		}
		predict_kwargs.update(batch_info_dict)
		extract_info = self._llm.predict(**predict_kwargs)
		new_info_dict = parse_common_collected_info(
			extract_info=extract_info,
			info_keys=list(self._common_infos.required_infos.keys()),
		)
		print("modified: ", new_info_dict)
		self._common_infos.update_collected_info(collected_info_dict=new_info_dict)