Spaces:
Sleeping
Sleeping
File size: 3,903 Bytes
53ae001 | 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 | """
Logging Configuration for RAG Agent API
Provides structured JSON logging for all components of the RAG Agent.
Logs include timestamp, level, component name, message, and contextual data.
Usage:
from config.logging_config import setup_logging, get_logger
setup_logging()
logger = get_logger('rag_agent')
logger.info("Message", extra={"query": "test", "latency_ms": 123})
"""
import logging
import json
import sys
from datetime import datetime
from typing import Any, Dict
class JSONFormatter(logging.Formatter):
"""
Custom formatter that outputs logs as JSON.
Format:
{
"timestamp": "2025-12-21T10:30:00Z",
"level": "INFO",
"component": "rag_agent",
"message": "Query processed",
"context": {"query": "...", "latency_ms": 123}
}
"""
def format(self, record: logging.LogRecord) -> str:
"""
Format log record as JSON string.
Args:
record: LogRecord to format
Returns:
JSON-formatted log string
"""
log_data: Dict[str, Any] = {
"timestamp": datetime.utcnow().isoformat() + "Z",
"level": record.levelname,
"component": record.name,
"message": record.getMessage(),
}
# Add any extra context from the log call
if hasattr(record, 'query'):
log_data['query'] = record.query
if hasattr(record, 'latency_ms'):
log_data['latency_ms'] = record.latency_ms
if hasattr(record, 'tokens_used'):
log_data['tokens_used'] = record.tokens_used
if hasattr(record, 'model'):
log_data['model'] = record.model
if hasattr(record, 'status_code'):
log_data['status_code'] = record.status_code
if hasattr(record, 'error_type'):
log_data['error_type'] = record.error_type
# Include exception info if present
if record.exc_info:
log_data['exception'] = self.formatException(record.exc_info)
return json.dumps(log_data)
def setup_logging(level: str = "INFO", json_format: bool = False) -> None:
"""
Configure logging for the application.
Args:
level: Logging level (DEBUG, INFO, WARNING, ERROR)
json_format: If True, use JSON formatter; otherwise use standard format
Example:
>>> setup_logging(level="DEBUG", json_format=True)
"""
log_level = getattr(logging, level.upper(), logging.INFO)
# Create handler for stdout
handler = logging.StreamHandler(sys.stdout)
if json_format:
handler.setFormatter(JSONFormatter())
else:
# Standard format for development
formatter = logging.Formatter(
'[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
handler.setFormatter(formatter)
# Configure root logger
logging.root.setLevel(log_level)
logging.root.handlers = [handler]
# Set levels for noisy libraries
logging.getLogger('httpx').setLevel(logging.WARNING)
logging.getLogger('httpcore').setLevel(logging.WARNING)
logging.getLogger('openai').setLevel(logging.WARNING)
logging.getLogger('urllib3').setLevel(logging.WARNING)
def get_logger(name: str) -> logging.Logger:
"""
Get a logger instance for a component.
Args:
name: Component name (e.g., 'rag_agent', 'retrieval', 'generation')
Returns:
Logger instance
Example:
>>> logger = get_logger('rag_agent')
>>> logger.info("Processing query", extra={"query": "test"})
"""
return logging.getLogger(name)
# Default setup for imports
if __name__ != "__main__":
setup_logging()
|