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
|
import json
import os
from typing import Any, Dict, Union
from .logger import logger
DEFAULT_PREFIX = "agent"
FULL_MESSAGE_HISTORY_FILE_NAME = "full_message_history.json"
CURRENT_CONTEXT_FILE_NAME = "current_context.json"
NEXT_ACTION_FILE_NAME = "next_action.json"
PROMPT_SUMMARY_FILE_NAME = "prompt_summary.json"
SUMMARY_FILE_NAME = "summary.txt"
SUPERVISOR_FEEDBACK_FILE_NAME = "supervisor_feedback.txt"
PROMPT_SUPERVISOR_FEEDBACK_FILE_NAME = "prompt_supervisor_feedback.json"
USER_INPUT_FILE_NAME = "user_input.txt"
class LogCycleHandler:
"""
A class for logging cycle data.
"""
def __init__(self):
self.log_count_within_cycle = 0
@staticmethod
def create_directory_if_not_exists(directory_path: str) -> None:
if not os.path.exists(directory_path):
os.makedirs(directory_path, exist_ok=True)
def create_outer_directory(self, ai_name: str, created_at: str) -> str:
log_directory = logger.get_log_directory()
if os.environ.get("OVERWRITE_DEBUG") == "1":
outer_folder_name = "auto_gpt"
else:
ai_name_short = self.get_agent_short_name(ai_name)
outer_folder_name = f"{created_at}_{ai_name_short}"
outer_folder_path = os.path.join(log_directory, "DEBUG", outer_folder_name)
self.create_directory_if_not_exists(outer_folder_path)
return outer_folder_path
def get_agent_short_name(self, ai_name: str) -> str:
return ai_name[:15].rstrip() if ai_name else DEFAULT_PREFIX
def create_inner_directory(self, outer_folder_path: str, cycle_count: int) -> str:
nested_folder_name = str(cycle_count).zfill(3)
nested_folder_path = os.path.join(outer_folder_path, nested_folder_name)
self.create_directory_if_not_exists(nested_folder_path)
return nested_folder_path
def create_nested_directory(
self, ai_name: str, created_at: str, cycle_count: int
) -> str:
outer_folder_path = self.create_outer_directory(ai_name, created_at)
nested_folder_path = self.create_inner_directory(outer_folder_path, cycle_count)
return nested_folder_path
def log_cycle(
self,
ai_name: str,
created_at: str,
cycle_count: int,
data: Union[Dict[str, Any], Any],
file_name: str,
) -> None:
"""
Log cycle data to a JSON file.
Args:
data (Any): The data to be logged.
file_name (str): The name of the file to save the logged data.
"""
nested_folder_path = self.create_nested_directory(
ai_name, created_at, cycle_count
)
json_data = json.dumps(data, ensure_ascii=False, indent=4)
log_file_path = os.path.join(
nested_folder_path, f"{self.log_count_within_cycle}_{file_name}"
)
logger.log_json(json_data, log_file_path)
self.log_count_within_cycle += 1
|
