Class for the blackboard, which stores the data and images which are visible to all the agents.
Initialize the blackboard.
Source code in agents/memory/blackboard.py
41
42
43
44
45
46
47
48
49
50
51
52
53 | def __init__(self) -> None:
"""
Initialize the blackboard.
"""
self._questions: Memory = Memory()
self._requests: Memory = Memory()
self._trajectories: Memory = Memory()
self._screenshots: Memory = Memory()
if configs.get("USE_CUSTOMIZATION", False):
self.load_questions(
configs.get("QA_PAIR_FILE", ""), configs.get("QA_PAIR_NUM", -1)
)
|
questions
property
Get the data from the blackboard.
Returns: |
-
Memory
–
The questions from the blackboard.
|
requests
property
Get the data from the blackboard.
Returns: |
-
Memory
–
The requests from the blackboard.
|
screenshots
property
Get the images from the blackboard.
Returns: |
-
Memory
–
The images from the blackboard.
|
trajectories
property
Get the data from the blackboard.
Returns: |
-
Memory
–
The trajectories from the blackboard.
|
add_data(data, memory)
Add the data to the a memory in the blackboard.
Parameters: |
-
data
(Union[MemoryItem, Dict[str, str], str] )
–
The data to be added. It can be a dictionary or a MemoryItem or a string.
-
memory
(Memory )
–
The memory to add the data to.
|
Source code in agents/memory/blackboard.py
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107 | def add_data(
self, data: Union[MemoryItem, Dict[str, str], str], memory: Memory
) -> None:
"""
Add the data to the a memory in the blackboard.
:param data: The data to be added. It can be a dictionary or a MemoryItem or a string.
:param memory: The memory to add the data to.
"""
if isinstance(data, dict):
data_memory = MemoryItem()
data_memory.add_values_from_dict(data)
memory.add_memory_item(data_memory)
elif isinstance(data, MemoryItem):
memory.add_memory_item(data)
elif isinstance(data, str):
data_memory = MemoryItem()
data_memory.add_values_from_dict({"text": data})
memory.add_memory_item(data_memory)
else:
print(f"Warning: Unsupported data type: {type(data)} when adding data.")
|
add_image(screenshot_path='', metadata=None)
Add the image to the blackboard.
Parameters: |
-
screenshot_path
(str , default:
''
)
–
-
metadata
(Optional[Dict[str, str]] , default:
None
)
–
The metadata of the image.
|
Source code in agents/memory/blackboard.py
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164 | def add_image(
self,
screenshot_path: str = "",
metadata: Optional[Dict[str, str]] = None,
) -> None:
"""
Add the image to the blackboard.
:param screenshot_path: The path of the image.
:param metadata: The metadata of the image.
"""
if os.path.exists(screenshot_path):
screenshot_str = PhotographerFacade().encode_image_from_path(
screenshot_path
)
else:
print(f"Screenshot path {screenshot_path} does not exist.")
screenshot_str = ""
image_memory_item = ImageMemoryItem()
image_memory_item.add_values_from_dict(
{
ImageMemoryItemNames.METADATA: metadata.get(
ImageMemoryItemNames.METADATA
),
ImageMemoryItemNames.IMAGE_PATH: screenshot_path,
ImageMemoryItemNames.IMAGE_STR: screenshot_str,
}
)
self.screenshots.add_memory_item(image_memory_item)
|
add_questions(questions)
Add the data to the blackboard.
Parameters: |
-
questions
(Union[MemoryItem, Dict[str, str]] )
–
The data to be added. It can be a dictionary or a MemoryItem or a string.
|
Source code in agents/memory/blackboard.py
109
110
111
112
113
114
115 | def add_questions(self, questions: Union[MemoryItem, Dict[str, str]]) -> None:
"""
Add the data to the blackboard.
:param questions: The data to be added. It can be a dictionary or a MemoryItem or a string.
"""
self.add_data(questions, self.questions)
|
add_requests(requests)
Add the data to the blackboard.
Parameters: |
-
requests
(Union[MemoryItem, Dict[str, str]] )
–
The data to be added. It can be a dictionary or a MemoryItem or a string.
|
Source code in agents/memory/blackboard.py
117
118
119
120
121
122
123 | def add_requests(self, requests: Union[MemoryItem, Dict[str, str]]) -> None:
"""
Add the data to the blackboard.
:param requests: The data to be added. It can be a dictionary or a MemoryItem or a string.
"""
self.add_data(requests, self.requests)
|
add_trajectories(trajectories)
Add the data to the blackboard.
Parameters: |
-
trajectories
(Union[MemoryItem, Dict[str, str]] )
–
The data to be added. It can be a dictionary or a MemoryItem or a string.
|
Source code in agents/memory/blackboard.py
125
126
127
128
129
130
131 | def add_trajectories(self, trajectories: Union[MemoryItem, Dict[str, str]]) -> None:
"""
Add the data to the blackboard.
:param trajectories: The data to be added. It can be a dictionary or a MemoryItem or a string.
"""
self.add_data(trajectories, self.trajectories)
|
blackboard_from_dict(blackboard_dict)
Convert the dictionary to the blackboard.
Parameters: |
-
blackboard_dict
(Dict[str, List[Dict[str, str]]] )
–
|
Source code in agents/memory/blackboard.py
264
265
266
267
268
269
270
271
272
273
274 | def blackboard_from_dict(
self, blackboard_dict: Dict[str, List[Dict[str, str]]]
) -> None:
"""
Convert the dictionary to the blackboard.
:param blackboard_dict: The dictionary.
"""
self.questions.from_list_of_dicts(blackboard_dict.get("questions", []))
self.requests.from_list_of_dicts(blackboard_dict.get("requests", []))
self.trajectories.from_list_of_dicts(blackboard_dict.get("trajectories", []))
self.screenshots.from_list_of_dicts(blackboard_dict.get("screenshots", []))
|
blackboard_to_dict()
Convert the blackboard to a dictionary.
Returns: |
-
Dict[str, List[Dict[str, str]]]
–
The blackboard in the dictionary format.
|
Source code in agents/memory/blackboard.py
243
244
245
246
247
248
249
250
251
252
253
254
255 | def blackboard_to_dict(self) -> Dict[str, List[Dict[str, str]]]:
"""
Convert the blackboard to a dictionary.
:return: The blackboard in the dictionary format.
"""
blackboard_dict = {
"questions": self.questions.to_list_of_dicts(),
"requests": self.requests.to_list_of_dicts(),
"trajectories": self.trajectories.to_list_of_dicts(),
"screenshots": self.screenshots.to_list_of_dicts(),
}
return blackboard_dict
|
blackboard_to_json()
Convert the blackboard to a JSON string.
Source code in agents/memory/blackboard.py
| def blackboard_to_json(self) -> str:
"""
Convert the blackboard to a JSON string.
:return: The JSON string.
"""
return json.dumps(self.blackboard_to_dict())
|
blackboard_to_prompt()
Convert the blackboard to a prompt.
Source code in agents/memory/blackboard.py
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298 | def blackboard_to_prompt(self) -> List[str]:
"""
Convert the blackboard to a prompt.
:return: The prompt.
"""
prefix = [
{
"type": "text",
"text": "[Blackboard:]",
}
]
blackboard_prompt = (
prefix
+ self.texts_to_prompt(self.questions, "[Questions & Answers:]")
+ self.texts_to_prompt(self.requests, "[Request History:]")
+ self.texts_to_prompt(
self.trajectories, "[Step Trajectories Completed Previously:]"
)
+ self.screenshots_to_prompt()
)
return blackboard_prompt
|
clear()
Clear the blackboard.
Source code in agents/memory/blackboard.py
312
313
314
315
316
317
318
319 | def clear(self) -> None:
"""
Clear the blackboard.
"""
self.questions.clear()
self.requests.clear()
self.trajectories.clear()
self.screenshots.clear()
|
is_empty()
Check if the blackboard is empty.
Returns: |
-
bool
–
True if the blackboard is empty, False otherwise.
|
Source code in agents/memory/blackboard.py
300
301
302
303
304
305
306
307
308
309
310 | def is_empty(self) -> bool:
"""
Check if the blackboard is empty.
:return: True if the blackboard is empty, False otherwise.
"""
return (
self.questions.is_empty()
and self.requests.is_empty()
and self.trajectories.is_empty()
and self.screenshots.is_empty()
)
|
load_questions(file_path, last_k=-1)
Load the data from a file.
Parameters: |
-
file_path
(str )
–
-
last_k
–
The number of lines to read from the end of the file. If -1, read all lines.
|
Source code in agents/memory/blackboard.py
194
195
196
197
198
199
200
201
202 | def load_questions(self, file_path: str, last_k=-1) -> None:
"""
Load the data from a file.
:param file_path: The path of the file.
:param last_k: The number of lines to read from the end of the file. If -1, read all lines.
"""
qa_list = self.read_json_file(file_path, last_k)
for qa in qa_list:
self.add_questions(qa)
|
questions_to_json()
Convert the data to a dictionary.
Returns: |
-
str
–
The data in the dictionary format.
|
Source code in agents/memory/blackboard.py
| def questions_to_json(self) -> str:
"""
Convert the data to a dictionary.
:return: The data in the dictionary format.
"""
return self.questions.to_json()
|
read_json_file(file_path, last_k=-1)
staticmethod
Read the json file.
Parameters: |
-
file_path
(str )
–
-
last_k
–
The number of lines to read from the end of the file. If -1, read all lines.
|
Source code in agents/memory/blackboard.py
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350 | @staticmethod
def read_json_file(file_path: str, last_k=-1) -> Dict[str, str]:
"""
Read the json file.
:param file_path: The path of the file.
:param last_k: The number of lines to read from the end of the file. If -1, read all lines.
:return: The data in the file.
"""
data_list = []
# Check if the file exists
if os.path.exists(file_path):
# Open the file and read the lines
with open(file_path, "r", encoding="utf-8") as file:
lines = file.readlines()
# If last_k is not -1, only read the last k lines
if last_k != -1:
lines = lines[-last_k:]
# Parse the lines as JSON
for line in lines:
try:
data = json.loads(line.strip())
data_list.append(data)
except json.JSONDecodeError:
print(f"Warning: Unable to parse line as JSON: {line}")
return data_list
|
requests_to_json()
Convert the data to a dictionary.
Returns: |
-
str
–
The data in the dictionary format.
|
Source code in agents/memory/blackboard.py
| def requests_to_json(self) -> str:
"""
Convert the data to a dictionary.
:return: The data in the dictionary format.
"""
return self.requests.to_json()
|
screenshots_to_json()
Convert the images to a dictionary.
Returns: |
-
str
–
The images in the dictionary format.
|
Source code in agents/memory/blackboard.py
| def screenshots_to_json(self) -> str:
"""
Convert the images to a dictionary.
:return: The images in the dictionary format.
"""
return self.screenshots.to_json()
|
screenshots_to_prompt()
Convert the images to a prompt.
Source code in agents/memory/blackboard.py
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241 | def screenshots_to_prompt(self) -> List[str]:
"""
Convert the images to a prompt.
:return: The prompt.
"""
user_content = []
for screenshot_dict in self.screenshots.list_content:
user_content.append(
{
"type": "text",
"text": json.dumps(
screenshot_dict.get(ImageMemoryItemNames.METADATA, "")
),
}
)
user_content.append(
{
"type": "image_url",
"image_url": {
"url": screenshot_dict.get(ImageMemoryItemNames.IMAGE_STR, "")
},
}
)
return user_content
|
texts_to_prompt(memory, prefix)
Convert the data to a prompt.
Source code in agents/memory/blackboard.py
204
205
206
207
208
209
210
211
212
213
214 | def texts_to_prompt(self, memory: Memory, prefix: str) -> List[str]:
"""
Convert the data to a prompt.
:return: The prompt.
"""
user_content = [
{"type": "text", "text": f"{prefix}\n {json.dumps(memory.list_content)}"}
]
return user_content
|
trajectories_to_json()
Convert the data to a dictionary.
Returns: |
-
str
–
The data in the dictionary format.
|
Source code in agents/memory/blackboard.py
| def trajectories_to_json(self) -> str:
"""
Convert the data to a dictionary.
:return: The data in the dictionary format.
"""
return self.trajectories.to_json()
|