跳转至

Add recent paper

labridge.callback.paper.add_paper

labridge.callback.paper.add_paper.AddNewRecentPaperOperation

Bases: CallBackOperationBase

This operation will add a new paper into a user's recent papers.

PARAMETER DESCRIPTION
llm

The used LLM.

TYPE: LLM DEFAULT: None

embed_model

The used embedding model.

TYPE: BaseEmbedding DEFAULT: None

verbose

Whether to show the inner progress.

TYPE: bool DEFAULT: False

Source code in labridge\callback\paper\add_paper.py
 18
 19
 20
 21
 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
class AddNewRecentPaperOperation(CallBackOperationBase):
	r"""
	This operation will add a new paper into a user's recent papers.

	Args:
		llm (LLM): The used LLM.
		embed_model (BaseEmbedding): The used embedding model.
		verbose (bool): Whether to show the inner progress.
	"""
	def __init__(
		self,
		llm: LLM = None,
		embed_model: BaseEmbedding = None,
		verbose: bool = False,
		op_name: str = None,
	):
		embed_model = embed_model or Settings.embed_model
		llm = llm or Settings.llm
		super().__init__(
			embed_model=embed_model,
			llm=llm,
			verbose=verbose,
			op_name=op_name or AddNewRecentPaperOperation.__name__,
		)

	def operation_description(self, **kwargs) -> str:
		r"""
		Return the operation description, this description will be sent to the user for authorization.

		Args:
			user_id (str): The user id of a lab member.
			paper_file_path (str): The file path of the new paper.

		Returns:
			str: The operation description.
		"""
		user_id = kwargs.get("user_id", None)
		paper_file_path = kwargs.get("paper_file_path", None)

		if None in [user_id, paper_file_path]:
			raise ValueError(f"Should provide these arguments: user_id, paper_file_path.")
		return ADD_NEW_RECENT_PAPER_TMPL.format(
			user_id=user_id,
			paper_file_path=paper_file_path,
		)

	def do_operation(
		self,
		**kwargs
	) -> OperationOutputLog:
		r"""
		Execute the operation after authorized to add a new paper to a user's recent papers.

		Args:
			user_id (str): The user id of a lab member.
			paper_file_path (str): The file path of the new paper.

		Returns:
			OperationOutputLog: The output and log of the operation.
		"""
		user_id = kwargs.get("user_id", None)
		paper_file_path = kwargs.get("paper_file_path", None)

		if None in [user_id, paper_file_path]:
			raise ValueError(f"Should provide these arguments: user_id, paper_file_path.")

		paper_store = RecentPaperStore.from_user_id(
			user_id=user_id,
			embed_model=self._embed_model,
		)
		try:
			paper_store.put(paper_file_path=paper_file_path)
			paper_store.persist()
			op_log = (
				f"Have put a new paper to the recent papers of the user {user_id}\n"
				f"Paper file path: {paper_file_path}"
			)
			paper_info = PaperInfo(
				file_path=paper_file_path,
				possessor=user_id,
				title=paper_file_path,
			)
			return OperationOutputLog(
				operation_name=self.op_name,
				operation_output=None,
				log_to_user=None,
				log_to_system={
					OP_DESCRIPTION: op_log,
					OP_REFERENCES: [paper_info.dumps()]
				}
			)

		except Exception as e:
			op_log = f"Error: {e}"
			return OperationOutputLog(
				operation_name=self.op_name,
				operation_output=None,
				log_to_user=None,
				log_to_system={
					OP_DESCRIPTION: op_log,
					OP_REFERENCES: None
				}
		)

	async def ado_operation(
		self,
		**kwargs
	) -> OperationOutputLog:
		return self.do_operation(**kwargs)

labridge.callback.paper.add_paper.AddNewRecentPaperOperation.do_operation(**kwargs)

Execute the operation after authorized to add a new paper to a user's recent papers.

PARAMETER DESCRIPTION
user_id

The user id of a lab member.

TYPE: str

paper_file_path

The file path of the new paper.

TYPE: str

RETURNS DESCRIPTION
OperationOutputLog

The output and log of the operation.

TYPE: OperationOutputLog

Source code in labridge\callback\paper\add_paper.py
 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
def do_operation(
	self,
	**kwargs
) -> OperationOutputLog:
	r"""
	Execute the operation after authorized to add a new paper to a user's recent papers.

	Args:
		user_id (str): The user id of a lab member.
		paper_file_path (str): The file path of the new paper.

	Returns:
		OperationOutputLog: The output and log of the operation.
	"""
	user_id = kwargs.get("user_id", None)
	paper_file_path = kwargs.get("paper_file_path", None)

	if None in [user_id, paper_file_path]:
		raise ValueError(f"Should provide these arguments: user_id, paper_file_path.")

	paper_store = RecentPaperStore.from_user_id(
		user_id=user_id,
		embed_model=self._embed_model,
	)
	try:
		paper_store.put(paper_file_path=paper_file_path)
		paper_store.persist()
		op_log = (
			f"Have put a new paper to the recent papers of the user {user_id}\n"
			f"Paper file path: {paper_file_path}"
		)
		paper_info = PaperInfo(
			file_path=paper_file_path,
			possessor=user_id,
			title=paper_file_path,
		)
		return OperationOutputLog(
			operation_name=self.op_name,
			operation_output=None,
			log_to_user=None,
			log_to_system={
				OP_DESCRIPTION: op_log,
				OP_REFERENCES: [paper_info.dumps()]
			}
		)

	except Exception as e:
		op_log = f"Error: {e}"
		return OperationOutputLog(
			operation_name=self.op_name,
			operation_output=None,
			log_to_user=None,
			log_to_system={
				OP_DESCRIPTION: op_log,
				OP_REFERENCES: None
			}
	)

labridge.callback.paper.add_paper.AddNewRecentPaperOperation.operation_description(**kwargs)

Return the operation description, this description will be sent to the user for authorization.

PARAMETER DESCRIPTION
user_id

The user id of a lab member.

TYPE: str

paper_file_path

The file path of the new paper.

TYPE: str

RETURNS DESCRIPTION
str

The operation description.

TYPE: str

Source code in labridge\callback\paper\add_paper.py
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def operation_description(self, **kwargs) -> str:
	r"""
	Return the operation description, this description will be sent to the user for authorization.

	Args:
		user_id (str): The user id of a lab member.
		paper_file_path (str): The file path of the new paper.

	Returns:
		str: The operation description.
	"""
	user_id = kwargs.get("user_id", None)
	paper_file_path = kwargs.get("paper_file_path", None)

	if None in [user_id, paper_file_path]:
		raise ValueError(f"Should provide these arguments: user_id, paper_file_path.")
	return ADD_NEW_RECENT_PAPER_TMPL.format(
		user_id=user_id,
		paper_file_path=paper_file_path,
	)