observability/logging_config.py

"""
Observability layer — matches "Pydantic Logfire / Span tracing" and
"LangSmith / Agent traces" in the architecture diagram, plus Cloud Logging /
Cloud Monitoring / Cloud Trace / Alerts in the GCP box.

- Structured console + Cloud Logging handler (auto-detected when running on GCP).
- Optional Logfire instrumentation for span-level tracing of the LangGraph run.
- Optional LangSmith tracing (enable via LANGCHAIN_TRACING_V2=true).
"""
from __future__ import annotations

import logging
import os


def configure_logging() -> None:
    level = os.getenv("LOG_LEVEL", "INFO").upper()
    logging.basicConfig(
        level=level,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    )

    # Google Cloud Logging — active automatically on Cloud Run, no-op locally if
    # credentials aren't available.
    if os.getenv("GCP_PROJECT_ID"):
        try:
            import google.cloud.logging as gcp_logging

            client = gcp_logging.Client()
            client.setup_logging(log_level=getattr(logging, level, logging.INFO))
        except Exception as exc:  # noqa: BLE001
            logging.getLogger(__name__).warning("Cloud Logging not configured: %s", exc)

    # Pydantic Logfire — span/trace instrumentation for the LangGraph pipeline.
    logfire_token = os.getenv("LOGFIRE_TOKEN")
    if logfire_token:
        try:
            import logfire

            logfire.configure(token=logfire_token)
            logfire.instrument_httpx()
        except Exception as exc:  # noqa: BLE001
            logging.getLogger(__name__).warning("Logfire not configured: %s", exc)

    # LangSmith tracing for the agentic core (set LANGCHAIN_TRACING_V2=true + LANGCHAIN_API_KEY).
    if os.getenv("LANGCHAIN_TRACING_V2", "false").lower() == "true":
        os.environ.setdefault("LANGCHAIN_PROJECT", "production-rag-langgraph")
        logging.getLogger(__name__).info("LangSmith tracing enabled for project=%s", os.environ["LANGCHAIN_PROJECT"])