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,
)
|