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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
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
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
|
"""Logging module for Auto-GPT."""
from __future__ import annotations
import enum
import logging
import os
import sys
from pathlib import Path
from typing import TYPE_CHECKING, Optional
from auto_gpt_plugin_template import AutoGPTPluginTemplate
from openai._base_client import log as openai_logger
if TYPE_CHECKING:
from autogpt.config import Config
from autogpt.speech import TTSConfig
from autogpt.core.configuration import SystemConfiguration, UserConfigurable
from autogpt.core.runner.client_lib.logging import BelowLevelFilter
from .formatters import AutoGptFormatter, StructuredLoggingFormatter
from .handlers import TTSHandler, TypingConsoleHandler
LOG_DIR = Path(__file__).parent.parent.parent / "logs"
LOG_FILE = "activity.log"
DEBUG_LOG_FILE = "debug.log"
ERROR_LOG_FILE = "error.log"
SIMPLE_LOG_FORMAT = "%(asctime)s %(levelname)s %(title)s%(message)s"
DEBUG_LOG_FORMAT = (
"%(asctime)s %(levelname)s %(filename)s:%(lineno)d" " %(title)s%(message)s"
)
SPEECH_OUTPUT_LOGGER = "VOICE"
USER_FRIENDLY_OUTPUT_LOGGER = "USER_FRIENDLY_OUTPUT"
_chat_plugins: list[AutoGPTPluginTemplate] = []
class LogFormatName(str, enum.Enum):
SIMPLE = "simple"
DEBUG = "debug"
STRUCTURED = "structured_google_cloud"
TEXT_LOG_FORMAT_MAP = {
LogFormatName.DEBUG: DEBUG_LOG_FORMAT,
LogFormatName.SIMPLE: SIMPLE_LOG_FORMAT,
}
class LoggingConfig(SystemConfiguration):
level: int = UserConfigurable(
default=logging.INFO,
from_env=lambda: logging.getLevelName(os.getenv("LOG_LEVEL", "INFO")),
)
# Console output
log_format: LogFormatName = UserConfigurable(
default=LogFormatName.SIMPLE, from_env="LOG_FORMAT"
)
plain_console_output: bool = UserConfigurable(
default=False,
from_env=lambda: os.getenv("PLAIN_OUTPUT", "False") == "True",
)
# File output
log_dir: Path = LOG_DIR
log_file_format: Optional[LogFormatName] = UserConfigurable(
default=LogFormatName.SIMPLE,
from_env=lambda: os.getenv(
"LOG_FILE_FORMAT", os.getenv("LOG_FORMAT", "simple")
),
)
def configure_logging(
level: int = logging.INFO,
log_dir: Path = LOG_DIR,
log_format: Optional[LogFormatName] = None,
log_file_format: Optional[LogFormatName] = None,
plain_console_output: bool = False,
tts_config: Optional[TTSConfig] = None,
) -> None:
"""Configure the native logging module.
Should be usable as `configure_logging(**config.logging.dict())`, where
`config.logging` is a `LoggingConfig` object.
"""
# Auto-adjust default log format based on log level
log_format = log_format or (
LogFormatName.SIMPLE if level != logging.DEBUG else LogFormatName.DEBUG
)
log_file_format = log_file_format or log_format
structured_logging = log_format == LogFormatName.STRUCTURED
if structured_logging:
plain_console_output = True
log_file_format = None
# create log directory if it doesn't exist
if not log_dir.exists():
log_dir.mkdir()
log_handlers: list[logging.Handler] = []
if log_format in (LogFormatName.DEBUG, LogFormatName.SIMPLE):
console_format_template = TEXT_LOG_FORMAT_MAP[log_format]
console_formatter = AutoGptFormatter(console_format_template)
else:
console_formatter = StructuredLoggingFormatter()
console_format_template = SIMPLE_LOG_FORMAT
# Console output handlers
stdout = logging.StreamHandler(stream=sys.stdout)
stdout.setLevel(level)
stdout.addFilter(BelowLevelFilter(logging.WARNING))
stdout.setFormatter(console_formatter)
stderr = logging.StreamHandler()
stderr.setLevel(logging.WARNING)
stderr.setFormatter(console_formatter)
log_handlers += [stdout, stderr]
# Console output handler which simulates typing
typing_console_handler = TypingConsoleHandler(stream=sys.stdout)
typing_console_handler.setLevel(logging.INFO)
typing_console_handler.setFormatter(console_formatter)
# User friendly output logger (text + speech)
user_friendly_output_logger = logging.getLogger(USER_FRIENDLY_OUTPUT_LOGGER)
user_friendly_output_logger.setLevel(logging.INFO)
user_friendly_output_logger.addHandler(
typing_console_handler if not plain_console_output else stdout
)
if tts_config:
user_friendly_output_logger.addHandler(TTSHandler(tts_config))
user_friendly_output_logger.addHandler(stderr)
user_friendly_output_logger.propagate = False
# File output handlers
if log_file_format is not None:
if level < logging.ERROR:
file_output_format_template = TEXT_LOG_FORMAT_MAP[log_file_format]
file_output_formatter = AutoGptFormatter(
file_output_format_template, no_color=True
)
# INFO log file handler
activity_log_handler = logging.FileHandler(log_dir / LOG_FILE, "a", "utf-8")
activity_log_handler.setLevel(level)
activity_log_handler.setFormatter(file_output_formatter)
log_handlers += [activity_log_handler]
user_friendly_output_logger.addHandler(activity_log_handler)
# ERROR log file handler
error_log_handler = logging.FileHandler(log_dir / ERROR_LOG_FILE, "a", "utf-8")
error_log_handler.setLevel(logging.ERROR)
error_log_handler.setFormatter(
AutoGptFormatter(DEBUG_LOG_FORMAT, no_color=True)
)
log_handlers += [error_log_handler]
user_friendly_output_logger.addHandler(error_log_handler)
# Configure the root logger
logging.basicConfig(
format=console_format_template,
level=level,
handlers=log_handlers,
)
# Speech output
speech_output_logger = logging.getLogger(SPEECH_OUTPUT_LOGGER)
speech_output_logger.setLevel(logging.INFO)
if tts_config:
speech_output_logger.addHandler(TTSHandler(tts_config))
speech_output_logger.propagate = False
# JSON logger with better formatting
json_logger = logging.getLogger("JSON_LOGGER")
json_logger.setLevel(logging.DEBUG)
json_logger.propagate = False
# Disable debug logging from OpenAI library
openai_logger.setLevel(logging.WARNING)
def configure_chat_plugins(config: Config) -> None:
"""Configure chat plugins for use by the logging module"""
logger = logging.getLogger(__name__)
# Add chat plugins capable of report to logger
if config.chat_messages_enabled:
if _chat_plugins:
_chat_plugins.clear()
for plugin in config.plugins:
if hasattr(plugin, "can_handle_report") and plugin.can_handle_report():
logger.debug(f"Loaded plugin into logger: {plugin.__class__.__name__}")
_chat_plugins.append(plugin)
|