-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmemory.py
More file actions
92 lines (76 loc) · 2.63 KB
/
memory.py
File metadata and controls
92 lines (76 loc) · 2.63 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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
import numpy as np
import pandas as pd
from langchain_community.llms import OpenAI
from langchain.agents import initialize_agent, Tool, AgentType
# from langchain.agents.agent_toolkits import create_structured_tool
from typing import List
from langchain_core.tools import StructuredTool
import os
class MemoryWorker:
_instance = None
_initialized = False
def __new__(cls, *args, **kwargs): # Singleton design pattern. Only 1 memory is created
if not cls._instance:
cls._instance = super(MemoryWorker, cls).__new__(cls)
# Initialize your stuff here
# cls._instance.data = []
# cls._instance.glance = []
return cls._instance
def __init__(self):
if hasattr(self, 'initialized'):
return # Skip reinitializing
self.initialized = True
print(123)
self.data = [] # Placeholder for the loaded DataFrame
self.glance = []
self.tools = [
StructuredTool.from_function(
name="Get_data_at_index",
func=self.get_data_at_idx,
description="Retrieve the object located at the specified index"
),
StructuredTool.from_function(
name="Glance_data",
func=self.glance_data,
description="Return a list of all glance messages recorded"
),
]
def add_object(self, obj, description: str):
idx = len(self.data)
self.data.append(obj)
self.glance.append({
"index": idx,
"description": description
})
# print(f"object added, new glance:{self.glance}")
def get_data_at_idx(self, idx: int):
"""
Retrieve the DataFrame located at the specified index.
Args:
idx (int): The index of the DataFrame in self.data to retrieve.
Returns:
pd.DataFrame: The DataFrame at the specified index.
"""
return self.data[idx]
def glance_data(self):
"""
Return a list of all the description of the data stored. This can help to decide which data to retrieve
Args: None
Returns:
list: A list containing all the glance messages.
"""
tmp = self.glance.copy()
tmp.reverse()
return tmp
# def get_current_data(self, inp: str):
# """
# Return the most recently object
# Args: None
# Returns:
# object
# """
# return self.data[-1]
def clear_data(self):
pass
memoryworker = MemoryWorker()
# memoryworker.glance_data()