mirror of
https://github.com/Significant-Gravitas/AutoGPT.git
synced 2026-04-08 03:00:28 -04:00
Updated logging
This commit is contained in:
SwiftyOS
@@ -1,15 +1,12 @@
|
||||
"""Logging module for Auto-GPT."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import enum
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from .crazyconf import SystemConfiguration, UserConfigurable
|
||||
from pydantic import Field, field_validator
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
from .filters import BelowLevelFilter
|
||||
from .formatters import AGPTFormatter, StructuredLoggingFormatter
|
||||
|
||||
@@ -19,173 +16,183 @@ 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"
|
||||
|
||||
class LoggingConfig(BaseSettings):
|
||||
|
||||
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")),
|
||||
level: str = Field(
|
||||
default="INFO",
|
||||
description="Logging level",
|
||||
validation_alias="LOG_LEVEL",
|
||||
)
|
||||
|
||||
# Console output
|
||||
log_format: LogFormatName = UserConfigurable(
|
||||
default=LogFormatName.SIMPLE, from_env="LOG_FORMAT"
|
||||
)
|
||||
plain_console_output: bool = UserConfigurable(
|
||||
enable_cloud_logging: bool = Field(
|
||||
default=False,
|
||||
from_env=lambda: os.getenv("PLAIN_OUTPUT", "False") == "True",
|
||||
description="Enable logging to Google Cloud Logging",
|
||||
validation_alias="ENABLE_CLOUD_LOGGING",
|
||||
)
|
||||
|
||||
enable_file_logging: bool = Field(
|
||||
default=True,
|
||||
description="Enable logging to file",
|
||||
validation_alias="ENABLE_FILE_LOGGING",
|
||||
)
|
||||
|
||||
enable_console_logging: bool = Field(
|
||||
default=True,
|
||||
description="Enable logging to console",
|
||||
validation_alias="ENABLE_CONSOLE_LOGGING",
|
||||
)
|
||||
|
||||
enable_json_logging: bool = Field(
|
||||
default=False,
|
||||
description="Enable logging to JSON file",
|
||||
validation_alias="ENABLE_JSON_LOGGING",
|
||||
)
|
||||
|
||||
# File output
|
||||
log_dir: Path = LOG_DIR
|
||||
log_file_format: Optional[LogFormatName] = UserConfigurable(
|
||||
default=LogFormatName.SIMPLE,
|
||||
from_env=lambda: os.getenv( # type: ignore
|
||||
"LOG_FILE_FORMAT", os.getenv("LOG_FORMAT", "simple")
|
||||
),
|
||||
log_dir: Path = Field(
|
||||
default=LOG_DIR,
|
||||
description="Log directory",
|
||||
)
|
||||
|
||||
model_config = SettingsConfigDict(
|
||||
env_prefix="",
|
||||
env_file=".env",
|
||||
env_file_encoding="utf-8",
|
||||
extra="ignore",
|
||||
)
|
||||
|
||||
@field_validator("level", mode="before")
|
||||
@classmethod
|
||||
def parse_log_level(cls, v):
|
||||
if isinstance(v, str):
|
||||
v = v.upper()
|
||||
if v not in ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]:
|
||||
raise ValueError(f"Invalid log level: {v}")
|
||||
return v
|
||||
return v
|
||||
|
||||
|
||||
def configure_logging(
|
||||
debug: bool = False,
|
||||
level: Optional[int | str] = None,
|
||||
log_dir: Optional[Path] = None,
|
||||
log_format: Optional[LogFormatName | str] = None,
|
||||
log_file_format: Optional[LogFormatName | str] = None,
|
||||
plain_console_output: Optional[bool] = None,
|
||||
config: Optional[LoggingConfig] = None,
|
||||
force_cloud_logging: bool = False
|
||||
) -> None:
|
||||
"""Configure the native logging module, based on the environment config and any
|
||||
specified overrides.
|
||||
"""Configure the native logging module based on the LoggingConfig settings.
|
||||
|
||||
Arguments override values specified in the environment.
|
||||
Overrides are also applied to `config`, if passed.
|
||||
This function sets up logging handlers and formatters according to the
|
||||
configuration specified in the LoggingConfig object. It supports various
|
||||
logging outputs including console, file, cloud, and JSON logging.
|
||||
|
||||
Should be usable as `configure_logging(**config.logging.dict())`, where
|
||||
`config.logging` is a `LoggingConfig` object.
|
||||
The function uses the LoggingConfig object to determine which logging
|
||||
features to enable and how to configure them. This includes setting
|
||||
log levels, log formats, and output destinations.
|
||||
|
||||
No arguments are required as the function creates its own LoggingConfig
|
||||
instance internally.
|
||||
|
||||
Note: This function is typically called at the start of the application
|
||||
to set up the logging infrastructure.
|
||||
"""
|
||||
if debug and level:
|
||||
raise ValueError("Only one of either 'debug' and 'level' arguments may be set")
|
||||
|
||||
# Parse arguments
|
||||
if isinstance(level, str):
|
||||
if type(_level := logging.getLevelName(level.upper())) is int:
|
||||
level = _level
|
||||
else:
|
||||
raise ValueError(f"Unknown log level '{level}'")
|
||||
if isinstance(log_format, str):
|
||||
if log_format in LogFormatName._value2member_map_:
|
||||
log_format = LogFormatName(log_format)
|
||||
elif not isinstance(log_format, LogFormatName):
|
||||
raise ValueError(f"Unknown log format '{log_format}'")
|
||||
if isinstance(log_file_format, str):
|
||||
if log_file_format in LogFormatName._value2member_map_:
|
||||
log_file_format = LogFormatName(log_file_format)
|
||||
elif not isinstance(log_file_format, LogFormatName):
|
||||
raise ValueError(f"Unknown log format '{log_format}'")
|
||||
|
||||
config = config or LoggingConfig.from_env()
|
||||
|
||||
# Aggregate env config + arguments
|
||||
config.level = logging.DEBUG if debug else level or config.level
|
||||
config.log_dir = log_dir or config.log_dir
|
||||
config.log_format = log_format or (
|
||||
LogFormatName.DEBUG if debug else config.log_format
|
||||
)
|
||||
config.log_file_format = log_file_format or log_format or config.log_file_format
|
||||
config.plain_console_output = (
|
||||
plain_console_output
|
||||
if plain_console_output is not None
|
||||
else config.plain_console_output
|
||||
)
|
||||
|
||||
# Structured logging is used for cloud environments,
|
||||
# where logging to a file makes no sense.
|
||||
if config.log_format == LogFormatName.STRUCTURED:
|
||||
config.plain_console_output = True
|
||||
config.log_file_format = None
|
||||
|
||||
# create log directory if it doesn't exist
|
||||
if not config.log_dir.exists():
|
||||
config.log_dir.mkdir()
|
||||
config = LoggingConfig()
|
||||
|
||||
log_handlers: list[logging.Handler] = []
|
||||
|
||||
if config.log_format in (LogFormatName.DEBUG, LogFormatName.SIMPLE):
|
||||
console_format_template = TEXT_LOG_FORMAT_MAP[config.log_format]
|
||||
console_formatter = AGPTFormatter(console_format_template)
|
||||
else:
|
||||
console_formatter = StructuredLoggingFormatter()
|
||||
console_format_template = SIMPLE_LOG_FORMAT
|
||||
# Cloud logging setup
|
||||
if config.enable_cloud_logging or force_cloud_logging:
|
||||
try:
|
||||
import google.cloud.logging
|
||||
from google.cloud.logging.handlers import CloudLoggingHandler
|
||||
from google.cloud.logging_v2.handlers.transports.sync import SyncTransport
|
||||
|
||||
# Console output handlers
|
||||
stdout = logging.StreamHandler(stream=sys.stdout)
|
||||
stdout.setLevel(config.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]
|
||||
|
||||
# File output handlers
|
||||
if config.log_file_format is not None:
|
||||
if config.level < logging.ERROR:
|
||||
file_output_format_template = TEXT_LOG_FORMAT_MAP[config.log_file_format]
|
||||
file_output_formatter = AGPTFormatter(
|
||||
file_output_format_template, no_color=True
|
||||
client = google.cloud.logging.Client()
|
||||
cloud_handler = CloudLoggingHandler(
|
||||
client,
|
||||
name="autogpt_logs",
|
||||
transport=SyncTransport,
|
||||
)
|
||||
|
||||
# INFO log file handler
|
||||
activity_log_handler = logging.FileHandler(
|
||||
config.log_dir / LOG_FILE, "a", "utf-8"
|
||||
cloud_handler.setLevel(config.level)
|
||||
cloud_handler.setFormatter(StructuredLoggingFormatter())
|
||||
log_handlers.append(cloud_handler)
|
||||
print("Cloud logging enabled")
|
||||
except ImportError:
|
||||
print(
|
||||
"Cloud logging requested but google-cloud-logging is not installed. "
|
||||
"Please install it with 'pip install google-cloud-logging'."
|
||||
)
|
||||
activity_log_handler.setLevel(config.level)
|
||||
activity_log_handler.setFormatter(file_output_formatter)
|
||||
log_handlers += [activity_log_handler]
|
||||
except Exception as e:
|
||||
print(f"Failed to set up cloud logging: {str(e)}")
|
||||
|
||||
# ERROR log file handler
|
||||
# File logging setup
|
||||
if config.enable_file_logging:
|
||||
# create log directory if it doesn't exist
|
||||
if not config.log_dir.exists():
|
||||
config.log_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Activity log handler (INFO and above)
|
||||
activity_log_handler = logging.FileHandler(
|
||||
config.log_dir / LOG_FILE, "a", "utf-8"
|
||||
)
|
||||
activity_log_handler.setLevel(config.level)
|
||||
activity_log_handler.setFormatter(
|
||||
AGPTFormatter(SIMPLE_LOG_FORMAT, no_color=True)
|
||||
)
|
||||
log_handlers.append(activity_log_handler)
|
||||
|
||||
# Debug log handler (all levels)
|
||||
debug_log_handler = logging.FileHandler(
|
||||
config.log_dir / DEBUG_LOG_FILE, "a", "utf-8"
|
||||
)
|
||||
debug_log_handler.setLevel(logging.DEBUG)
|
||||
debug_log_handler.setFormatter(AGPTFormatter(DEBUG_LOG_FORMAT, no_color=True))
|
||||
log_handlers.append(debug_log_handler)
|
||||
|
||||
# Error log handler (ERROR and above)
|
||||
error_log_handler = logging.FileHandler(
|
||||
config.log_dir / ERROR_LOG_FILE, "a", "utf-8"
|
||||
)
|
||||
error_log_handler.setLevel(logging.ERROR)
|
||||
error_log_handler.setFormatter(AGPTFormatter(DEBUG_LOG_FORMAT, no_color=True))
|
||||
log_handlers += [error_log_handler]
|
||||
log_handlers.append(error_log_handler)
|
||||
|
||||
print("File logging enabled")
|
||||
|
||||
if config.enable_console_logging:
|
||||
# Console output handlers
|
||||
stdout = logging.StreamHandler(stream=sys.stdout)
|
||||
stdout.setLevel(config.level)
|
||||
stdout.addFilter(BelowLevelFilter(logging.WARNING))
|
||||
if config.level == logging.DEBUG:
|
||||
stdout.setFormatter(AGPTFormatter(DEBUG_LOG_FORMAT))
|
||||
else:
|
||||
stdout.setFormatter(AGPTFormatter(SIMPLE_LOG_FORMAT))
|
||||
|
||||
stderr = logging.StreamHandler()
|
||||
stderr.setLevel(logging.WARNING)
|
||||
if config.level == logging.DEBUG:
|
||||
stderr.setFormatter(AGPTFormatter(DEBUG_LOG_FORMAT))
|
||||
else:
|
||||
stderr.setFormatter(AGPTFormatter(SIMPLE_LOG_FORMAT))
|
||||
|
||||
log_handlers += [stdout, stderr]
|
||||
print("Console logging enabled")
|
||||
|
||||
# Configure the root logger
|
||||
logging.basicConfig(
|
||||
format=console_format_template,
|
||||
format=DEBUG_LOG_FORMAT if config.level == logging.DEBUG else SIMPLE_LOG_FORMAT,
|
||||
level=config.level,
|
||||
handlers=log_handlers,
|
||||
)
|
||||
|
||||
# Speech output
|
||||
speech_output_logger = logging.getLogger(SPEECH_OUTPUT_LOGGER)
|
||||
speech_output_logger.setLevel(logging.INFO)
|
||||
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
|
||||
if config.enable_json_logging:
|
||||
# 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 = logging.getLogger("openai")
|
||||
|
||||
@@ -1,370 +0,0 @@
|
||||
import os
|
||||
import typing
|
||||
from typing import Any, Callable, Generic, Optional, Type, TypeVar, get_args
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field, ValidationError
|
||||
from pydantic._internal._model_construction import (
|
||||
ModelMetaclass, # HACK shouldn't be used
|
||||
)
|
||||
from pydantic.fields import FieldInfo
|
||||
from pydantic_core import PydanticUndefined, PydanticUndefinedType
|
||||
|
||||
T = TypeVar("T")
|
||||
M = TypeVar("M", bound=BaseModel)
|
||||
|
||||
|
||||
def UserConfigurable(
|
||||
default: T | PydanticUndefinedType = PydanticUndefined,
|
||||
*args,
|
||||
default_factory: Optional[Callable[[], T]] = None,
|
||||
from_env: Optional[str | Callable[[], T | None]] = None,
|
||||
description: str = "",
|
||||
exclude: bool = False,
|
||||
**kwargs,
|
||||
) -> T:
|
||||
# TODO: use this to auto-generate docs for the application configuration
|
||||
field_info: FieldInfo = Field(
|
||||
default,
|
||||
*args,
|
||||
default_factory=default_factory,
|
||||
description=description,
|
||||
exclude=exclude,
|
||||
**kwargs,
|
||||
)
|
||||
field_info.metadata.append(("user_configurable", True))
|
||||
field_info.metadata.append(("from_env", from_env))
|
||||
|
||||
return field_info # type: ignore
|
||||
|
||||
|
||||
def _get_field_metadata(field: FieldInfo, key: str, default: Any = None) -> Any:
|
||||
for item in field.metadata:
|
||||
if isinstance(item, tuple) and item[0] == key:
|
||||
return item[1]
|
||||
if isinstance(item, str) and item == key:
|
||||
return True
|
||||
return default
|
||||
|
||||
|
||||
class SystemConfiguration(BaseModel):
|
||||
def get_user_config(self) -> dict[str, Any]:
|
||||
return _recurse_user_config_values(self)
|
||||
|
||||
@classmethod
|
||||
def from_env(cls):
|
||||
"""
|
||||
Initializes the config object from environment variables.
|
||||
|
||||
Environment variables are mapped to UserConfigurable fields using the from_env
|
||||
attribute that can be passed to UserConfigurable.
|
||||
"""
|
||||
|
||||
def infer_field_value(field: FieldInfo):
|
||||
default_value = (
|
||||
field.default
|
||||
if field.default not in (None, PydanticUndefined)
|
||||
else (
|
||||
field.default_factory()
|
||||
if field.default_factory
|
||||
else PydanticUndefined
|
||||
)
|
||||
)
|
||||
if from_env := _get_field_metadata(field, "from_env"):
|
||||
val_from_env = (
|
||||
os.getenv(from_env) if type(from_env) is str else from_env()
|
||||
)
|
||||
if val_from_env is not None:
|
||||
return val_from_env
|
||||
return default_value
|
||||
|
||||
return _recursive_init_model(cls, infer_field_value)
|
||||
|
||||
model_config = ConfigDict(
|
||||
extra="forbid", use_enum_values=True, validate_assignment=True
|
||||
)
|
||||
|
||||
|
||||
SC = TypeVar("SC", bound=SystemConfiguration)
|
||||
|
||||
|
||||
class SystemSettings(BaseModel):
|
||||
"""A base class for all system settings."""
|
||||
|
||||
name: str
|
||||
description: str
|
||||
|
||||
model_config = ConfigDict(
|
||||
extra="forbid", use_enum_values=True, validate_assignment=True
|
||||
)
|
||||
|
||||
|
||||
S = TypeVar("S", bound=SystemSettings)
|
||||
|
||||
|
||||
class Configurable(Generic[S]):
|
||||
"""A base class for all configurable objects."""
|
||||
|
||||
prefix: str = ""
|
||||
default_settings: typing.ClassVar[S] # type: ignore
|
||||
|
||||
@classmethod
|
||||
def get_user_config(cls) -> dict[str, Any]:
|
||||
return _recurse_user_config_values(cls.default_settings)
|
||||
|
||||
@classmethod
|
||||
def build_agent_configuration(cls, overrides: dict = {}) -> S:
|
||||
"""Process the configuration for this object."""
|
||||
|
||||
base_config = _update_user_config_from_env(cls.default_settings)
|
||||
final_configuration = deep_update(base_config, overrides)
|
||||
|
||||
return cls.default_settings.__class__.model_validate(final_configuration)
|
||||
|
||||
|
||||
def _update_user_config_from_env(instance: BaseModel) -> dict[str, Any]:
|
||||
"""
|
||||
Update config fields of a Pydantic model instance from environment variables.
|
||||
|
||||
Precedence:
|
||||
1. Non-default value already on the instance
|
||||
2. Value returned by `from_env()`
|
||||
3. Default value for the field
|
||||
|
||||
Params:
|
||||
instance: The Pydantic model instance.
|
||||
|
||||
Returns:
|
||||
The user config fields of the instance.
|
||||
"""
|
||||
|
||||
def infer_field_value(field: FieldInfo, value):
|
||||
default_value = (
|
||||
field.default
|
||||
if field.default not in (None, PydanticUndefined)
|
||||
else (field.default_factory() if field.default_factory else None)
|
||||
)
|
||||
if value == default_value and (
|
||||
from_env := _get_field_metadata(field, "from_env")
|
||||
):
|
||||
val_from_env = os.getenv(from_env) if type(from_env) is str else from_env()
|
||||
if val_from_env is not None:
|
||||
return val_from_env
|
||||
return value
|
||||
|
||||
def init_sub_config(model: Type[SC]) -> SC | None:
|
||||
try:
|
||||
return model.model_validate(model.from_env(), strict=True)
|
||||
except ValidationError as e:
|
||||
# Gracefully handle missing fields
|
||||
if all(e["type"] == "missing" for e in e.errors()):
|
||||
return None
|
||||
raise
|
||||
|
||||
return _recurse_user_config_fields(instance, infer_field_value, init_sub_config)
|
||||
|
||||
|
||||
def _recursive_init_model(
|
||||
model: Type[M],
|
||||
infer_field_value: Callable[[FieldInfo], Any],
|
||||
) -> M:
|
||||
"""
|
||||
Recursively initialize the user configuration fields of a Pydantic model.
|
||||
|
||||
Parameters:
|
||||
model: The Pydantic model type.
|
||||
infer_field_value: A callback function to infer the value of each field.
|
||||
Parameters:
|
||||
ModelField: The Pydantic ModelField object describing the field.
|
||||
|
||||
Returns:
|
||||
BaseModel: An instance of the model with the initialized configuration.
|
||||
"""
|
||||
user_config_fields = {}
|
||||
for name, field in model.model_fields.items():
|
||||
if _get_field_metadata(field, "user_configurable"):
|
||||
user_config_fields[name] = infer_field_value(field)
|
||||
elif isinstance(field.annotation, ModelMetaclass) and issubclass(
|
||||
field.annotation, SystemConfiguration
|
||||
):
|
||||
try:
|
||||
user_config_fields[name] = _recursive_init_model(
|
||||
model=field.annotation,
|
||||
infer_field_value=infer_field_value,
|
||||
)
|
||||
except ValidationError as e:
|
||||
# Gracefully handle missing fields
|
||||
if all(e["type"] == "missing" for e in e.errors()):
|
||||
user_config_fields[name] = None
|
||||
raise
|
||||
|
||||
user_config_fields = remove_none_items(user_config_fields)
|
||||
|
||||
return model.model_validate(user_config_fields)
|
||||
|
||||
|
||||
def _recurse_user_config_fields(
|
||||
model: BaseModel,
|
||||
infer_field_value: Callable[[FieldInfo, Any], Any],
|
||||
init_sub_config: Optional[
|
||||
Callable[[Type[SystemConfiguration]], SystemConfiguration | None]
|
||||
] = None,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Recursively process the user configuration fields of a Pydantic model instance.
|
||||
|
||||
Params:
|
||||
model: The Pydantic model to iterate over.
|
||||
infer_field_value: A callback function to process each field.
|
||||
Params:
|
||||
ModelField: The Pydantic ModelField object describing the field.
|
||||
Any: The current value of the field.
|
||||
init_sub_config: An optional callback function to initialize a sub-config.
|
||||
Params:
|
||||
Type[SystemConfiguration]: The type of the sub-config to initialize.
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: The processed user configuration fields of the instance.
|
||||
"""
|
||||
user_config_fields = {}
|
||||
|
||||
for name, field in model.model_fields.items():
|
||||
value = getattr(model, name)
|
||||
|
||||
# Handle individual field
|
||||
if _get_field_metadata(field, "user_configurable"):
|
||||
user_config_fields[name] = infer_field_value(field, value)
|
||||
|
||||
# Recurse into nested config object
|
||||
elif isinstance(value, SystemConfiguration):
|
||||
user_config_fields[name] = _recurse_user_config_fields(
|
||||
model=value,
|
||||
infer_field_value=infer_field_value,
|
||||
init_sub_config=init_sub_config,
|
||||
)
|
||||
|
||||
# Recurse into optional nested config object
|
||||
elif value is None and init_sub_config:
|
||||
field_type = get_args(field.annotation)[0] # Optional[T] -> T
|
||||
if type(field_type) is ModelMetaclass and issubclass(
|
||||
field_type, SystemConfiguration
|
||||
):
|
||||
sub_config = init_sub_config(field_type)
|
||||
if sub_config:
|
||||
user_config_fields[name] = _recurse_user_config_fields(
|
||||
model=sub_config,
|
||||
infer_field_value=infer_field_value,
|
||||
init_sub_config=init_sub_config,
|
||||
)
|
||||
|
||||
elif isinstance(value, list) and all(
|
||||
isinstance(i, SystemConfiguration) for i in value
|
||||
):
|
||||
user_config_fields[name] = [
|
||||
_recurse_user_config_fields(i, infer_field_value, init_sub_config)
|
||||
for i in value
|
||||
]
|
||||
elif isinstance(value, dict) and all(
|
||||
isinstance(i, SystemConfiguration) for i in value.values()
|
||||
):
|
||||
user_config_fields[name] = {
|
||||
k: _recurse_user_config_fields(v, infer_field_value, init_sub_config)
|
||||
for k, v in value.items()
|
||||
}
|
||||
|
||||
return user_config_fields
|
||||
|
||||
|
||||
def _recurse_user_config_values(
|
||||
instance: BaseModel,
|
||||
get_field_value: Callable[[FieldInfo, T], T] = lambda _, v: v,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
This function recursively traverses the user configuration values in a Pydantic
|
||||
model instance.
|
||||
|
||||
Params:
|
||||
instance: A Pydantic model instance.
|
||||
get_field_value: A callback function to process each field. Parameters:
|
||||
ModelField: The Pydantic ModelField object that describes the field.
|
||||
Any: The current value of the field.
|
||||
|
||||
Returns:
|
||||
A dictionary containing the processed user configuration fields of the instance.
|
||||
"""
|
||||
user_config_values = {}
|
||||
|
||||
for name, value in instance.__dict__.items():
|
||||
field = instance.model_fields[name]
|
||||
if _get_field_metadata(field, "user_configurable"):
|
||||
user_config_values[name] = get_field_value(field, value)
|
||||
elif isinstance(value, SystemConfiguration):
|
||||
user_config_values[name] = _recurse_user_config_values(
|
||||
instance=value, get_field_value=get_field_value
|
||||
)
|
||||
elif isinstance(value, list) and all(
|
||||
isinstance(i, SystemConfiguration) for i in value
|
||||
):
|
||||
user_config_values[name] = [
|
||||
_recurse_user_config_values(i, get_field_value) for i in value
|
||||
]
|
||||
elif isinstance(value, dict) and all(
|
||||
isinstance(i, SystemConfiguration) for i in value.values()
|
||||
):
|
||||
user_config_values[name] = {
|
||||
k: _recurse_user_config_values(v, get_field_value)
|
||||
for k, v in value.items()
|
||||
}
|
||||
|
||||
return user_config_values
|
||||
|
||||
|
||||
def _get_non_default_user_config_values(instance: BaseModel) -> dict[str, Any]:
|
||||
"""
|
||||
Get the non-default user config fields of a Pydantic model instance.
|
||||
|
||||
Params:
|
||||
instance: The Pydantic model instance.
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: The non-default user config values on the instance.
|
||||
"""
|
||||
|
||||
def get_field_value(field: FieldInfo, value):
|
||||
default = field.default_factory() if field.default_factory else field.default
|
||||
if value != default:
|
||||
return value
|
||||
|
||||
return remove_none_items(_recurse_user_config_values(instance, get_field_value))
|
||||
|
||||
|
||||
def deep_update(original_dict: dict, update_dict: dict) -> dict:
|
||||
"""
|
||||
Recursively update a dictionary.
|
||||
|
||||
Params:
|
||||
original_dict (dict): The dictionary to be updated.
|
||||
update_dict (dict): The dictionary to update with.
|
||||
|
||||
Returns:
|
||||
dict: The updated dictionary.
|
||||
"""
|
||||
for key, value in update_dict.items():
|
||||
if (
|
||||
key in original_dict
|
||||
and isinstance(original_dict[key], dict)
|
||||
and isinstance(value, dict)
|
||||
):
|
||||
original_dict[key] = deep_update(original_dict[key], value)
|
||||
else:
|
||||
original_dict[key] = value
|
||||
return original_dict
|
||||
|
||||
|
||||
def remove_none_items(d):
|
||||
if isinstance(d, dict):
|
||||
return {
|
||||
k: remove_none_items(v)
|
||||
for k, v in d.items()
|
||||
if v not in (None, PydanticUndefined)
|
||||
}
|
||||
return d
|
||||
@@ -25,9 +25,3 @@ def print_attribute(
|
||||
"color": value_color,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def speak(message: str, level: int = logging.INFO) -> None:
|
||||
from .config import SPEECH_OUTPUT_LOGGER
|
||||
|
||||
logging.getLogger(SPEECH_OUTPUT_LOGGER).log(level, message)
|
||||
|
||||
22
rnd/autogpt_libs/poetry.lock
generated
22
rnd/autogpt_libs/poetry.lock
generated
@@ -1171,6 +1171,26 @@ files = [
|
||||
[package.dependencies]
|
||||
typing-extensions = ">=4.6.0,<4.7.0 || >4.7.0"
|
||||
|
||||
[[package]]
|
||||
name = "pydantic-settings"
|
||||
version = "2.5.2"
|
||||
description = "Settings management using Pydantic"
|
||||
optional = false
|
||||
python-versions = ">=3.8"
|
||||
files = [
|
||||
{file = "pydantic_settings-2.5.2-py3-none-any.whl", hash = "sha256:2c912e55fd5794a59bf8c832b9de832dcfdf4778d79ff79b708744eed499a907"},
|
||||
{file = "pydantic_settings-2.5.2.tar.gz", hash = "sha256:f90b139682bee4d2065273d5185d71d37ea46cfe57e1b5ae184fc6a0b2484ca0"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
pydantic = ">=2.7.0"
|
||||
python-dotenv = ">=0.21.0"
|
||||
|
||||
[package.extras]
|
||||
azure-key-vault = ["azure-identity (>=1.16.0)", "azure-keyvault-secrets (>=4.8.0)"]
|
||||
toml = ["tomli (>=2.0.1)"]
|
||||
yaml = ["pyyaml (>=6.0.1)"]
|
||||
|
||||
[[package]]
|
||||
name = "pyjwt"
|
||||
version = "2.9.0"
|
||||
@@ -1670,4 +1690,4 @@ type = ["pytest-mypy"]
|
||||
[metadata]
|
||||
lock-version = "2.0"
|
||||
python-versions = ">=3.10,<4.0"
|
||||
content-hash = "e6b4bc9df9bad143369092e5f91bc3fe4f038d72fe8bd83e02dfe0eeaf11d5c0"
|
||||
content-hash = "e9b6e5d877eeb9c9f1ebc69dead1985d749facc160afbe61f3bf37e9a6e35aa5"
|
||||
|
||||
@@ -14,6 +14,7 @@ python-dotenv = "^1.0.1"
|
||||
supabase = "^2.7.2"
|
||||
colorama = "^0.4.6"
|
||||
google-cloud-logging = "^3.8.0"
|
||||
pydantic-settings = "^2.5.2"
|
||||
|
||||
|
||||
[build-system]
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import os
|
||||
|
||||
from autogpt_libs.logging.config import LogFormatName
|
||||
from autogpt_libs.logging.config import configure_logging
|
||||
|
||||
|
||||
def configure_logging():
|
||||
@@ -11,7 +11,7 @@ def configure_logging():
|
||||
if os.getenv("APP_ENV") != "cloud":
|
||||
configure_logging()
|
||||
else:
|
||||
configure_logging(log_format=LogFormatName.STRUCTURED)
|
||||
configure_logging(force_cloud_logging=True)
|
||||
|
||||
# Silence httpx logger
|
||||
logging.getLogger("httpx").setLevel(logging.WARNING)
|
||||
|
||||
Reference in New Issue
Block a user