Authorize

labridge.interact.authorize.authorize

labridge.interact.authorize.authorize.aoperation_authorize(user_id, op_name, kwargs_str, authorize_strict_mode=False, llm=None, embed_model=None, verbose=False) async

This function is used to query the user whether to execute a specific operation.

PARAMETER	DESCRIPTION
user_id	The user that will make decisions. TYPE: str
op_name	The operation to be executed. TYPE: str
kwargs_str	The keyword arguments of the operation function, which is dumped as a json string. TYPE: str
authorize_strict_mode	If it is set to True, the operation will be executed only when the user response the STRICT_AGREE_WORDS. Defaults to False. TYPE: bool DEFAULT: False
llm	The used LLM. Defaults to None. If set to None, the Settings.llm will be used. TYPE: LLM DEFAULT: None
embed_model	The used embedding model. Defaults to None. If set to None, the Settings.embed_model will be used. TYPE: BaseEmbedding DEFAULT: None
verbose	Whether to show the progress. TYPE: str DEFAULT: False

RETURNS	DESCRIPTION
callback_log	the log string. TYPE: str

Source code in labridge\interact\authorize\authorize.py

async def aoperation_authorize(
	user_id: str,
	op_name: str,
	kwargs_str: str,
	authorize_strict_mode: bool = False,
	llm: LLM = None,
	embed_model: BaseEmbedding = None,
	verbose: bool = False,
) -> OperationOutputLog:
	r"""
	This function is used to query the user whether to execute a specific operation.

	Args:
		user_id (str): The user that will make decisions.
		op_name (str): The operation to be executed.
		kwargs_str (str): The keyword arguments of the operation function, which is dumped as a json string.
		authorize_strict_mode (bool): If it is set to True, the operation will be executed only when the user response
			the `STRICT_AGREE_WORDS`. Defaults to False.
		llm (LLM): The used LLM. Defaults to None. If set to None, the Settings.llm will be used.
		embed_model (BaseEmbedding): The used embedding model. Defaults to None.
			If set to None, the Settings.embed_model will be used.
		verbose (str): Whether to show the progress.

	Returns:
		callback_log (str): the log string.

	"""
	if op_name not in CALL_BACK_OPS:
		raise ValueError(f"{op_name} is not a valid callback operation name.")

	operation_class = getattr(callback, op_name)
	if not issubclass(operation_class, CallBackOperationBase):
		raise ValueError(f"{op_name} should be a subclass of 'CallBackOperationBase'.")

	llm = llm or Settings.llm
	embed_model = embed_model or Settings.embed_model

	operation = operation_class(
		llm=llm,
		embed_model=embed_model,
		verbose=verbose,
		op_name=op_name,
	)
	kwargs = json.loads(kwargs_str)
	op_description = operation.operation_description(**kwargs)
	if authorize_strict_mode:
		query_str = STRICT_AUTHORIZE_QUERY_TMPL.format(
			strict_agree_str=",".join(STRICT_AGREE_WORDS),
			operation_description=op_description,
		)
	else:
		query_str = AUTHORIZE_QUERY_TMPL.format(operation_description=op_description)

	# TODO: send the operation description to the user.
	ChatBuffer.put_agent_reply(
		user_id=user_id,
		reply_str=query_str,
		inner_chat=True,
	)

	# TODO: wait for the user response.
	user_msg: PackedUserMessage = await ChatBuffer.get_user_msg(user_id=user_id)
	if user_msg is None:
		raise ValueError("Invalid user msg.")

	user_response = user_msg.user_msg

	agree = False
	if authorize_strict_mode:
		if user_response.encode("utf-8").isalpha():
			user_response = user_response.lower()
		agree = user_response in STRICT_AGREE_WORDS
	else:
		judgement = await llm.apredict(
			prompt=AUTHORIZATION_ANALYZE_PROMPT,
			agree_word=ANALYZE_AGREE_WORD,
			disagree_word=ANALYZE_DISAGREE_WORD,
			user_response=user_response,
		)
		agree = analyze_agree(llm_response=judgement)

	if agree:
		# TODO: I need an operation buffer to store operations. two operation class: 1. real time, 2. buffer.
		callback_log = operation.do_operation(**kwargs)
		return callback_log
	else:
		callback_log_str = (f"The assistant tries to obtain the authorization from user {user_id} to perform an operation."
						f"The user disagreed, so this operation does not be performed.\n"
						f"The operation is described as follows:\n{op_description}\n\n"
						f"The user's response is as follows:\n{user_response}")
		return OperationOutputLog(
			operation_name=op_name,
			operation_output=callback_log_str,
			log_to_user=None,
			log_to_system={
				OP_DESCRIPTION: callback_log_str,
				OP_REFERENCES: None,
			},
			operation_abort=True,
		)

labridge.interact.authorize.authorize.operation_authorize(user_id, op_name, kwargs_str, authorize_strict_mode=False, llm=None, embed_model=None, verbose=False)

This function is used to query the user whether to execute a specific operation.

PARAMETER	DESCRIPTION
user_id	The user that will make decisions. TYPE: str
op_name	The operation to be executed. TYPE: str
kwargs_str	The keyword arguments of the operation function, which is dumped as a json string. TYPE: str
authorize_strict_mode	If it is set to True, the operation will be executed only when the user response the STRICT_AGREE_WORDS. Defaults to False. TYPE: bool DEFAULT: False
llm	The used LLM. Defaults to None. If set to None, the Settings.llm will be used. TYPE: LLM DEFAULT: None
embed_model	The used embedding model. Defaults to None. If set to None, the Settings.embed_model will be used. TYPE: BaseEmbedding DEFAULT: None
verbose	Whether to show the progress. TYPE: str DEFAULT: False

RETURNS	DESCRIPTION
callback_log	the log string. TYPE: str

Source code in labridge\interact\authorize\authorize.py

def operation_authorize(
	user_id: str,
	op_name: str,
	kwargs_str: str,
	authorize_strict_mode: bool = False,
	llm: LLM = None,
	embed_model: BaseEmbedding = None,
	verbose: bool = False,
) -> OperationOutputLog:
	r"""
	This function is used to query the user whether to execute a specific operation.

	Args:
		user_id (str): The user that will make decisions.
		op_name (str): The operation to be executed.
		kwargs_str (str): The keyword arguments of the operation function, which is dumped as a json string.
		authorize_strict_mode (bool): If it is set to True, the operation will be executed only when the user response
			the `STRICT_AGREE_WORDS`. Defaults to False.
		llm (LLM): The used LLM. Defaults to None. If set to None, the Settings.llm will be used.
		embed_model (BaseEmbedding): The used embedding model. Defaults to None.
			If set to None, the Settings.embed_model will be used.
		verbose (str): Whether to show the progress.

	Returns:
		callback_log (str): the log string.

	"""
	if op_name not in CALL_BACK_OPS:
		raise ValueError(f"{op_name} is not a valid callback operation name.")

	operation_class = getattr(callback, op_name)
	if not issubclass(operation_class, CallBackOperationBase):
		raise ValueError(f"{op_name} should be a subclass of 'CallBackOperationBase'.")

	llm = llm or Settings.llm
	embed_model = embed_model or Settings.embed_model

	operation = operation_class(
		llm=llm,
		embed_model=embed_model,
		verbose=verbose,
		op_name=op_name,
	)

	kwargs = json.loads(kwargs_str)
	op_description = operation.operation_description(**kwargs)
	if authorize_strict_mode:
		query_str = STRICT_AUTHORIZE_QUERY_TMPL.format(
			strict_agree_str=",".join(STRICT_AGREE_WORDS),
			operation_description=op_description,
		)
	else:
		query_str = AUTHORIZE_QUERY_TMPL.format(operation_description=op_description)

	# TODO: send the operation description to the user.
	print(query_str)

	# TODO: wait for the user response.
	user_msg: PackedUserMessage = ChatBuffer.test_get_user_text(user_id=user_id)
	user_response = user_msg.user_msg

	agree = False
	print("Here 1 ....")
	if authorize_strict_mode:
		if user_response.encode("utf-8").isalpha():
			user_response = user_response.lower()
		agree = user_response in STRICT_AGREE_WORDS
	else:
		judgement = llm.predict(
			prompt=AUTHORIZATION_ANALYZE_PROMPT,
			agree_word=ANALYZE_AGREE_WORD,
			disagree_word=ANALYZE_DISAGREE_WORD,
			user_response=user_response,
		)
		agree = analyze_agree(llm_response=judgement)
		print("Here 2 ....")

	if agree:
		# TODO: I need an operation buffer to store operations.
		callback_log = operation.do_operation(**kwargs)
		print("Here", callback_log.dumps())
		return callback_log
	else:
		callback_log_str = (
			f"The assistant tries to obtain the authorization from user {user_id} to perform an operation."
			f"The user disagreed, so this operation does not be performed.\n"
			f"The operation is described as follows:\n{op_description}\n\n"
			f"The user's response is as follows:\n{user_response}"
		)
		return OperationOutputLog(
			operation_name=op_name,
			operation_output=callback_log_str,
			log_to_user=None,
			log_to_system={
				OP_DESCRIPTION: callback_log_str,
				OP_REFERENCES: None,
			},
			operation_abort=True,
		)