Paper summarize

labridge.callback.paper.paper_summarize

labridge.callback.paper.paper_summarize.PaperSummarizeOperation

Bases: CallBackOperationBase

This operation will summarize a paper.

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\paper_summarize.py

class PaperSummarizeOperation(CallBackOperationBase):
	r"""
	This operation will summarize a paper.

	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,
	):
		root = Path(__file__)

		for idx in range(4):
			root = root.parent

		self.root = root
		self._fs = fsspec.filesystem("file")
		embed_model = embed_model or Settings.embed_model
		llm = llm or Settings.llm
		self._summarizer = PaperBatchSummarize(llm=llm)
		super().__init__(
			llm=llm,
			embed_model=embed_model,
			verbose=verbose,
			op_name=op_name or PaperSummarizeOperation.__name__,
		)

	def operation_description(self, **kwargs) -> str:
		r"""
		Describe the operation.

		Args:
			user_id (str): the user id.
			paper_file_path (str): The file path of the 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("should provide valid user_id, paper_infos.")

		description = SUMMARIZE_DESCRIPTION_TMPL.format(user_id=user_id, paper_file_path=paper_file_path)
		return description

	def do_operation(
		self,
		user_id: str,
		paper_file_path: str,
	) -> OperationOutputLog:
		r"""
		Execute the operation to summarize a paper in 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 paper.

		Returns:
			OperationOutputLog:
				The operation output and log.
		"""
		recent_paper_store = RecentPaperStore.from_user_id(
			user_id=user_id,
			embed_model=self._embed_model,
		)
		recent_paper_store.check_valid_paper(paper_file_path=paper_file_path)
		summary_node = recent_paper_store.get_summary_node(paper_file_path=paper_file_path)
		if summary_node is not None:
			return summary_node.text

		# TODO: Send to the user
		print("Assistant: 正在为您总结中，请稍候。")
		paper_nodes = recent_paper_store.get_paper_nodes(paper_file_path=paper_file_path)
		nodes_with_scores = [NodeWithScore(node=n) for n in paper_nodes]
		# get the summary for each doc_id
		summary_response = self._summarizer.synthesize(
			nodes=nodes_with_scores,
			query=""
		)
		summary_response = cast(Response, summary_response)
		summary_node = TextNode(text=summary_response.response)
		recent_paper_store.insert_summary_node(
			paper_file_path=paper_file_path,
			summary_node=summary_node,
		)
		recent_paper_store.persist()
		paper_info = PaperInfo(
			title=paper_file_path,
			possessor=user_id,
			file_path=paper_file_path,
		)
		op_log = f"Have summarized the paper {paper_file_path} for the user {user_id}."
		return OperationOutputLog(
			operation_name=self.op_name,
			operation_output=summary_node.text,
			log_to_user=None,
			log_to_system={
				OP_DESCRIPTION: op_log,
				OP_REFERENCES: [paper_info.dumps()]
			},
		)

	async def ado_operation(
		self,
		user_id: str,
		paper_file_path: str,
	) -> OperationOutputLog:
		r"""
		Asynchronously execute the operation to summarize a paper in 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 paper.

		Returns:
			OperationOutputLog:
				The output and log.
		"""
		recent_paper_store = RecentPaperStore.from_user_id(
			user_id=user_id,
			embed_model=self._embed_model,
		)
		recent_paper_store.check_valid_paper(paper_file_path=paper_file_path)
		summary_node = recent_paper_store.get_summary_node(paper_file_path=paper_file_path)
		if summary_node is not None:
			return summary_node.text

		# TODO: Send to the user
		print("Assistant: 正在为您总结中，请稍候。")
		paper_nodes = recent_paper_store.get_paper_nodes(paper_file_path=paper_file_path)
		nodes_with_scores = [NodeWithScore(node=n) for n in paper_nodes]
		# get the summary for each doc_id
		summary_response = await self._summarizer.asynthesize(
			nodes=nodes_with_scores,
			query=""
		)
		summary_response = cast(Response, summary_response)
		summary_node = TextNode(text=summary_response.response)
		recent_paper_store.insert_summary_node(
			paper_file_path=paper_file_path,
			summary_node=summary_node,
		)
		recent_paper_store.persist()
		paper_info = PaperInfo(
			title=paper_file_path,
			possessor=user_id,
			file_path=paper_file_path,
		)
		op_log = f"Have summarized the paper {paper_file_path} for the user {user_id}."
		return OperationOutputLog(
			operation_name=self.op_name,
			operation_output=summary_node.text,
			log_to_user=None,
			log_to_system={
				OP_DESCRIPTION: op_log,
				OP_REFERENCES: [paper_info.dumps()]
			},
		)

labridge.callback.paper.paper_summarize.PaperSummarizeOperation.ado_operation(user_id, paper_file_path) async

Asynchronously execute the operation to summarize a paper in 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 paper. TYPE: str

RETURNS	DESCRIPTION
OperationOutputLog	The output and log. TYPE: OperationOutputLog

Source code in labridge\callback\paper\paper_summarize.py

async def ado_operation(
	self,
	user_id: str,
	paper_file_path: str,
) -> OperationOutputLog:
	r"""
	Asynchronously execute the operation to summarize a paper in 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 paper.

	Returns:
		OperationOutputLog:
			The output and log.
	"""
	recent_paper_store = RecentPaperStore.from_user_id(
		user_id=user_id,
		embed_model=self._embed_model,
	)
	recent_paper_store.check_valid_paper(paper_file_path=paper_file_path)
	summary_node = recent_paper_store.get_summary_node(paper_file_path=paper_file_path)
	if summary_node is not None:
		return summary_node.text

	# TODO: Send to the user
	print("Assistant: 正在为您总结中，请稍候。")
	paper_nodes = recent_paper_store.get_paper_nodes(paper_file_path=paper_file_path)
	nodes_with_scores = [NodeWithScore(node=n) for n in paper_nodes]
	# get the summary for each doc_id
	summary_response = await self._summarizer.asynthesize(
		nodes=nodes_with_scores,
		query=""
	)
	summary_response = cast(Response, summary_response)
	summary_node = TextNode(text=summary_response.response)
	recent_paper_store.insert_summary_node(
		paper_file_path=paper_file_path,
		summary_node=summary_node,
	)
	recent_paper_store.persist()
	paper_info = PaperInfo(
		title=paper_file_path,
		possessor=user_id,
		file_path=paper_file_path,
	)
	op_log = f"Have summarized the paper {paper_file_path} for the user {user_id}."
	return OperationOutputLog(
		operation_name=self.op_name,
		operation_output=summary_node.text,
		log_to_user=None,
		log_to_system={
			OP_DESCRIPTION: op_log,
			OP_REFERENCES: [paper_info.dumps()]
		},
	)

labridge.callback.paper.paper_summarize.PaperSummarizeOperation.do_operation(user_id, paper_file_path)

Execute the operation to summarize a paper in 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 paper. TYPE: str

RETURNS	DESCRIPTION
OperationOutputLog	The operation output and log. TYPE: OperationOutputLog

Source code in labridge\callback\paper\paper_summarize.py

def do_operation(
	self,
	user_id: str,
	paper_file_path: str,
) -> OperationOutputLog:
	r"""
	Execute the operation to summarize a paper in 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 paper.

	Returns:
		OperationOutputLog:
			The operation output and log.
	"""
	recent_paper_store = RecentPaperStore.from_user_id(
		user_id=user_id,
		embed_model=self._embed_model,
	)
	recent_paper_store.check_valid_paper(paper_file_path=paper_file_path)
	summary_node = recent_paper_store.get_summary_node(paper_file_path=paper_file_path)
	if summary_node is not None:
		return summary_node.text

	# TODO: Send to the user
	print("Assistant: 正在为您总结中，请稍候。")
	paper_nodes = recent_paper_store.get_paper_nodes(paper_file_path=paper_file_path)
	nodes_with_scores = [NodeWithScore(node=n) for n in paper_nodes]
	# get the summary for each doc_id
	summary_response = self._summarizer.synthesize(
		nodes=nodes_with_scores,
		query=""
	)
	summary_response = cast(Response, summary_response)
	summary_node = TextNode(text=summary_response.response)
	recent_paper_store.insert_summary_node(
		paper_file_path=paper_file_path,
		summary_node=summary_node,
	)
	recent_paper_store.persist()
	paper_info = PaperInfo(
		title=paper_file_path,
		possessor=user_id,
		file_path=paper_file_path,
	)
	op_log = f"Have summarized the paper {paper_file_path} for the user {user_id}."
	return OperationOutputLog(
		operation_name=self.op_name,
		operation_output=summary_node.text,
		log_to_user=None,
		log_to_system={
			OP_DESCRIPTION: op_log,
			OP_REFERENCES: [paper_info.dumps()]
		},
	)

labridge.callback.paper.paper_summarize.PaperSummarizeOperation.operation_description(**kwargs)

Describe the operation.

PARAMETER	DESCRIPTION
user_id	the user id. TYPE: str
paper_file_path	The file path of the paper. TYPE: str

RETURNS	DESCRIPTION
str	the operation description. TYPE: str

Source code in labridge\callback\paper\paper_summarize.py

def operation_description(self, **kwargs) -> str:
	r"""
	Describe the operation.

	Args:
		user_id (str): the user id.
		paper_file_path (str): The file path of the 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("should provide valid user_id, paper_infos.")

	description = SUMMARIZE_DESCRIPTION_TMPL.format(user_id=user_id, paper_file_path=paper_file_path)
	return description