langfuse-events.py

import uuid
from typing import Any, Dict, Optional

def get_iso_timestamp() -> str:
    """
    Returns the current time in ISO 8601 format.
    """
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())

def log_event(
    trace: Any,
    stage: str,
    details: Dict[str, Any],
    status: str = "success",
    input_data: Optional[Dict[str, Any]] = None,
    output_data: Optional[Dict[str, Any]] = None
) -> None:
    """
    Logs an event to Langfuse with a unique event ID, current timestamp,
    and optional input/output metadata.
    """
    event_id = "evt_" + str(uuid.uuid4())[:8]
    metadata = {
        "event_id": event_id,
        "timestamp": get_iso_timestamp(),
        "stage": stage,
        "details": details,
        "status": status,
    }

    if input_data:
        metadata["input_data"] = input_data
    if output_data:
        metadata["output_data"] = output_data

    try:
        trace.event(name=f"{stage}_audit", metadata=metadata)
    except Exception as exc:
        logger.error("Failed to log event: %s", exc)