Spaces:
Sleeping
Sleeping
| import os | |
| from functools import lru_cache | |
| from typing import Any, Dict, List, Optional | |
| from app.core.config import get_env_bool | |
| from app.core.logging import get_logger | |
| logger = get_logger(__name__) | |
| def is_tracing_enabled() -> bool: | |
| """Return True if LangSmith / LangChain tracing is enabled via environment.""" | |
| tracing_flag = get_env_bool("LANGCHAIN_TRACING_V2", False) | |
| api_key = os.getenv("LANGCHAIN_API_KEY") | |
| return tracing_flag and bool(api_key) | |
| def get_langsmith_project() -> Optional[str]: | |
| """Return the LangSmith project name, if configured.""" | |
| return os.getenv("LANGCHAIN_PROJECT") | |
| def get_tracing_callbacks() -> List[Any]: | |
| """Return LangChain callback handlers for tracing, if available. | |
| When LANGCHAIN_TRACING_V2=true and LANGCHAIN_API_KEY is set, this will | |
| attempt to create a LangChainTracer instance. If tracing is not enabled | |
| or the tracer is unavailable, an empty list is returned. | |
| """ | |
| if not is_tracing_enabled(): | |
| logger.info( | |
| "LangSmith tracing disabled (set LANGCHAIN_TRACING_V2=true and " | |
| "LANGCHAIN_API_KEY to enable)." | |
| ) | |
| return [] | |
| try: | |
| from langchain_core.tracers import LangChainTracer # type: ignore[import] | |
| except Exception as exc: # noqa: BLE001 | |
| logger.warning( | |
| "LangSmith tracing requested but LangChainTracer is unavailable: %s", exc | |
| ) | |
| return [] | |
| project = get_langsmith_project() | |
| tracer = LangChainTracer(project_name=project) | |
| logger.info( | |
| "LangSmith tracing enabled for project='%s'", | |
| project or "(default)", | |
| ) | |
| return [tracer] | |
| def get_tracing_response_metadata() -> Dict[str, Any]: | |
| """Return trace metadata suitable for API responses.""" | |
| return { | |
| "langsmith_project": get_langsmith_project(), | |
| "trace_enabled": is_tracing_enabled(), | |
| } |