Bases: RetrieverBaseTool
This tool is used to retrieve in the permanent chat memory of a user or a chat group.
| PARAMETER |
DESCRIPTION |
chat_memory_retriever |
The chat memory retriever.
TYPE:
ChatMemoryRetriever
DEFAULT:
None
|
metadata_mode |
The metadata mode, defaults to MetadataMode.LLM.
TYPE:
MetadataMode
DEFAULT:
LLM
|
Source code in labridge/tools/memory/chat/retrieve.py
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 | class ChatMemoryRetrieverTool(RetrieverBaseTool):
r"""
This tool is used to retrieve in the permanent chat memory of a user or a chat group.
Args:
chat_memory_retriever (ChatMemoryRetriever): The chat memory retriever.
metadata_mode (MetadataMode): The metadata mode, defaults to `MetadataMode.LLM`.
"""
def __init__(
self,
chat_memory_retriever: ChatMemoryRetriever = None,
metadata_mode: MetadataMode = MetadataMode.LLM,
):
self.metadata_mode = metadata_mode
chat_memory_retriever = chat_memory_retriever or ChatMemoryRetriever()
super().__init__(
retriever=chat_memory_retriever,
name=ChatMemoryRetrieverTool.__name__,
retrieve_fn=ChatMemoryRetriever.retrieve,
)
def get_ref_info(self, nodes: List[NodeWithScore]) -> List[RefInfoBase]:
r""" Get the reference infos from the retrieved nodes. """
return []
def log(self, log_dict) -> ToolLog:
r""" tool log """
item = log_dict["item_to_be_retrieved"]
memory_id = log_dict["memory_id"]
start_date = log_dict["start_date"]
end_date = log_dict["end_date"]
log_string = (
f"Using {self.metadata.name} to retrieve '{item}' in the chat history memory with memory_id: '{memory_id}'\n"
f"Retrieve date is ranging from {start_date} to {end_date}\n"
)
log_to_user = None
log_to_system = {
TOOL_OP_DESCRIPTION: log_string,
TOOL_REFERENCES: None,
}
return ToolLog(
tool_name=self.metadata.name,
log_to_user=log_to_user,
log_to_system=log_to_system,
)
def _retrieve(self, retrieve_kwargs: dict) -> List[NodeWithScore]:
r""" Use the retriever to retrieve relevant nodes. """
nodes = self._retriever.retrieve(**retrieve_kwargs)
return nodes
async def _aretrieve(self, retrieve_kwargs: dict) -> List[NodeWithScore]:
r""" Asynchronously use the retriever to retrieve relevant nodes. """
nodes = await self._retriever.aretrieve(**retrieve_kwargs)
return nodes
def _nodes_to_tool_output(self, nodes: List[NodeWithScore]) -> Tuple[str, dict]:
r""" output the retrieved contents in a specific format. """
if nodes:
output = f"Have retrieved relevant conversations in the chat memory\n\n"
contents = []
for idx, node in enumerate(nodes):
node_content = (
f"Conversation {idx + 1}:\n"
f"{node.node.get_content(metadata_mode=self.metadata_mode)}"
)
contents.append(node_content.strip())
output += "\n\n".join(contents)
else:
output = "No Relevant chat history found."
output_log = dict()
return output, output_log
|
Get the reference infos from the retrieved nodes.
Source code in labridge/tools/memory/chat/retrieve.py
| def get_ref_info(self, nodes: List[NodeWithScore]) -> List[RefInfoBase]:
r""" Get the reference infos from the retrieved nodes. """
return []
|
tool log
Source code in labridge/tools/memory/chat/retrieve.py
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66 | def log(self, log_dict) -> ToolLog:
r""" tool log """
item = log_dict["item_to_be_retrieved"]
memory_id = log_dict["memory_id"]
start_date = log_dict["start_date"]
end_date = log_dict["end_date"]
log_string = (
f"Using {self.metadata.name} to retrieve '{item}' in the chat history memory with memory_id: '{memory_id}'\n"
f"Retrieve date is ranging from {start_date} to {end_date}\n"
)
log_to_user = None
log_to_system = {
TOOL_OP_DESCRIPTION: log_string,
TOOL_REFERENCES: None,
}
return ToolLog(
tool_name=self.metadata.name,
log_to_user=log_to_user,
log_to_system=log_to_system,
)
|