File size: 2,408 Bytes
42d88ae | 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 | import logging
import sys
from contextvars import ContextVar
from datetime import datetime
from typing import Any, Optional
from uuid import UUID
import json
correlation_id: ContextVar[Optional[str]] = ContextVar("correlation_id", default=None)
class JSONFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
log_data = {
"timestamp": datetime.utcnow().isoformat(),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
"correlation_id": correlation_id.get(),
}
if hasattr(record, "issue_id"):
log_data["issue_id"] = str(record.issue_id)
if hasattr(record, "agent"):
log_data["agent"] = record.agent
if hasattr(record, "decision"):
log_data["decision"] = record.decision
if record.exc_info:
log_data["exception"] = self.formatException(record.exc_info)
return json.dumps(log_data)
class AgentLogger(logging.LoggerAdapter):
def __init__(self, logger: logging.Logger, agent_name: str):
super().__init__(logger, {"agent": agent_name})
def process(self, msg: str, kwargs: dict[str, Any]) -> tuple[str, dict[str, Any]]:
extra = kwargs.get("extra", {})
extra["agent"] = self.extra["agent"]
kwargs["extra"] = extra
return msg, kwargs
def log_decision(
self,
issue_id: UUID,
decision: str,
reasoning: str,
level: int = logging.INFO
) -> None:
self.log(
level,
f"Decision: {decision} | Reasoning: {reasoning}",
extra={"issue_id": issue_id, "decision": decision}
)
def setup_logging(debug: bool = False) -> None:
root = logging.getLogger()
root.setLevel(logging.DEBUG if debug else logging.INFO)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
root.addHandler(handler)
logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
def get_logger(name: str, agent_name: Optional[str] = None) -> logging.Logger | AgentLogger:
logger = logging.getLogger(name)
if agent_name:
return AgentLogger(logger, agent_name)
return logger
|