跳转至

Collect manager

labridge.interact.collect.manager.collect_manager

labridge.interact.collect.manager.collect_manager.CollectManager

This manager judges whether to abort the collecting process according to the user's response, and whether the collected information need modification.

PARAMETER DESCRIPTION
llm

The used LLM.

TYPE: LLM

Source code in labridge\interact\collect\manager\collect_manager.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
class CollectManager:
	r"""
	This manager judges whether to abort the collecting process according to the user's response,
	and whether the collected information need modification.

	Args:
		llm (LLM): The used LLM.
	"""
	def __init__(
		self,
		llm: LLM,
	):
		self._llm = llm

	def analyze_whether_abort(self, user_response: str) -> bool:
		r"""
		Whether the user tends to abort.

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

		Returns:
			bool: Whether to abort or not.
		"""
		abort = condition_analyze(
			llm=self._llm,
			prompt=COLLECT_ABORT_PROMPT,
			condition_true_word=COLLECT_ABORT_WORD,
			abort_word=COLLECT_ABORT_WORD,
			continue_word=COLLECT_CONTINUE_WORD,
			user_response=user_response,
		)
		return abort

	async def async_analyze_whether_abort(self, user_response: str) -> bool:
		r"""
		Async version.
		Whether the user tends to abort.

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

		Returns:
			bool: Whether to abort or not.
		"""
		abort = await acondition_analyze(
			llm=self._llm,
			prompt=COLLECT_ABORT_PROMPT,
			condition_true_word=COLLECT_ABORT_WORD,
			abort_word=COLLECT_ABORT_WORD,
			continue_word=COLLECT_CONTINUE_WORD,
			user_response=user_response,
		)
		return abort

	async def async_analyze_whether_modify(
		self,
		user_response: str,
		collected_info_dict: Dict[str, str],
	) -> bool:
		r"""
		Async version.
		Whether the user thinks the collected information need modification.

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

		Returns:
			bool: Whether to modify or not.
		"""
		do_modify = await acondition_analyze(
			llm=self._llm,
			prompt=WHETHER_MODIFY_INFO_PROMPT,
			condition_true_word=DO_MODIFY_WORD,
			do_modify_word=DO_MODIFY_WORD,
			not_modify_word=NOT_MODIFY_WORD,
			collected_infos_str=json.dumps(collected_info_dict),
			user_comment_str=user_response,
		)
		return do_modify

	def analyze_whether_modify(
		self,
		user_response: str,
		collected_info_dict: Dict[str, str],
	) -> bool:
		r"""
		Whether the user thinks the collected information need modification.

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

		Returns:
			bool: Whether to modify or not.
		"""
		do_modify = condition_analyze(
			llm=self._llm,
			prompt=WHETHER_MODIFY_INFO_PROMPT,
			condition_true_word=DO_MODIFY_WORD,
			do_modify_word=DO_MODIFY_WORD,
			not_modify_word=NOT_MODIFY_WORD,
			collected_infos_str=json.dumps(collected_info_dict),
			user_comment_str=user_response,
		)
		return do_modify

	def verify_query(self, collected_info_dict: Dict[str, str]) -> str:
		r""" This query will be sent to the user to verify the correctness of the collected information. """
		verify_str = f"{VERIFY_COLLECTED_INFO_QUERY}\n"
		for key in collected_info_dict.keys():
			verify_str += f"{key}:\n\t{collected_info_dict[key]}\n"
		return verify_str

labridge.interact.collect.manager.collect_manager.CollectManager.analyze_whether_abort(user_response)

Whether the user tends to abort.

PARAMETER DESCRIPTION
user_response

The user's response.

TYPE: str

RETURNS DESCRIPTION
bool

Whether to abort or not.

TYPE: bool

Source code in labridge\interact\collect\manager\collect_manager.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def analyze_whether_abort(self, user_response: str) -> bool:
	r"""
	Whether the user tends to abort.

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

	Returns:
		bool: Whether to abort or not.
	"""
	abort = condition_analyze(
		llm=self._llm,
		prompt=COLLECT_ABORT_PROMPT,
		condition_true_word=COLLECT_ABORT_WORD,
		abort_word=COLLECT_ABORT_WORD,
		continue_word=COLLECT_CONTINUE_WORD,
		user_response=user_response,
	)
	return abort

labridge.interact.collect.manager.collect_manager.CollectManager.analyze_whether_modify(user_response, collected_info_dict)

Whether the user thinks the collected information need modification.

PARAMETER DESCRIPTION
user_response

The user's response.

TYPE: str

RETURNS DESCRIPTION
bool

Whether to modify or not.

TYPE: bool

Source code in labridge\interact\collect\manager\collect_manager.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def analyze_whether_modify(
	self,
	user_response: str,
	collected_info_dict: Dict[str, str],
) -> bool:
	r"""
	Whether the user thinks the collected information need modification.

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

	Returns:
		bool: Whether to modify or not.
	"""
	do_modify = condition_analyze(
		llm=self._llm,
		prompt=WHETHER_MODIFY_INFO_PROMPT,
		condition_true_word=DO_MODIFY_WORD,
		do_modify_word=DO_MODIFY_WORD,
		not_modify_word=NOT_MODIFY_WORD,
		collected_infos_str=json.dumps(collected_info_dict),
		user_comment_str=user_response,
	)
	return do_modify

labridge.interact.collect.manager.collect_manager.CollectManager.async_analyze_whether_abort(user_response) async

Async version. Whether the user tends to abort.

PARAMETER DESCRIPTION
user_response

The user's response.

TYPE: str

RETURNS DESCRIPTION
bool

Whether to abort or not.

TYPE: bool

Source code in labridge\interact\collect\manager\collect_manager.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
async def async_analyze_whether_abort(self, user_response: str) -> bool:
	r"""
	Async version.
	Whether the user tends to abort.

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

	Returns:
		bool: Whether to abort or not.
	"""
	abort = await acondition_analyze(
		llm=self._llm,
		prompt=COLLECT_ABORT_PROMPT,
		condition_true_word=COLLECT_ABORT_WORD,
		abort_word=COLLECT_ABORT_WORD,
		continue_word=COLLECT_CONTINUE_WORD,
		user_response=user_response,
	)
	return abort

labridge.interact.collect.manager.collect_manager.CollectManager.async_analyze_whether_modify(user_response, collected_info_dict) async

Async version. Whether the user thinks the collected information need modification.

PARAMETER DESCRIPTION
user_response

The user's response.

TYPE: str

RETURNS DESCRIPTION
bool

Whether to modify or not.

TYPE: bool

Source code in labridge\interact\collect\manager\collect_manager.py 
 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
async def async_analyze_whether_modify(
	self,
	user_response: str,
	collected_info_dict: Dict[str, str],
) -> bool:
	r"""
	Async version.
	Whether the user thinks the collected information need modification.

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

	Returns:
		bool: Whether to modify or not.
	"""
	do_modify = await acondition_analyze(
		llm=self._llm,
		prompt=WHETHER_MODIFY_INFO_PROMPT,
		condition_true_word=DO_MODIFY_WORD,
		do_modify_word=DO_MODIFY_WORD,
		not_modify_word=NOT_MODIFY_WORD,
		collected_infos_str=json.dumps(collected_info_dict),
		user_comment_str=user_response,
	)
	return do_modify

labridge.interact.collect.manager.collect_manager.CollectManager.verify_query(collected_info_dict)

This query will be sent to the user to verify the correctness of the collected information.

Source code in labridge\interact\collect\manager\collect_manager.py 
128
129
130
131
132
133
def verify_query(self, collected_info_dict: Dict[str, str]) -> str:
	r""" This query will be sent to the user to verify the correctness of the collected information. """
	verify_str = f"{VERIFY_COLLECTED_INFO_QUERY}\n"
	for key in collected_info_dict.keys():
		verify_str += f"{key}:\n\t{collected_info_dict[key]}\n"
	return verify_str