import dataclasses import json import logging import os import traceback from opentelemetry import context as context_api from opentelemetry.instrumentation.cohere.config import Config TRACELOOP_TRACE_CONTENT = "TRACELOOP_TRACE_CONTENT" def dont_throw(func): """ A decorator that wraps the passed in function and logs exceptions instead of throwing them. @param func: The function to wrap @return: The wrapper function """ # Obtain a logger specific to the function's module logger = logging.getLogger(func.__module__) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: logger.debug( "OpenLLMetry failed to trace in %s, error: %s", func.__name__, traceback.format_exc(), ) if Config.exception_logger: Config.exception_logger(e) return wrapper def should_send_prompts(): return ( os.getenv(TRACELOOP_TRACE_CONTENT) or "true" ).lower() == "true" or context_api.get_value("override_enable_content_tracing") def should_emit_events() -> bool: """ Checks if the instrumentation isn't using the legacy attributes and if the event logger is not None. """ return not Config.use_legacy_attributes def dump_object(obj): try: if hasattr(obj, "model_dump_json") and callable(obj.model_dump_json): return obj.model_dump_json() except Exception: pass try: return json.dumps(obj) except Exception: return "" def to_dict(obj): try: if hasattr(obj, "model_dump") and callable(obj.model_dump): return obj.model_dump() except Exception: pass if isinstance(obj, dict): return obj if dataclasses.is_dataclass(obj): return dataclasses.asdict(obj) try: return dict(obj) except Exception: return obj