Spaces:
Running
Running
File size: 3,473 Bytes
73c6377 |
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 |
import os
import uuid
import contextvars
from typing import Optional, Dict, Any
# LangSmith for Tracing and Monitoring
try:
from langsmith import traceable, Client
from langsmith import trace
except Exception as e:
# Provide fallbacks if langsmith is not installed to avoid runtime crashes
traceable = lambda *args, **kwargs: (lambda f: f)
class _Noop:
def __call__(self, *args, **kwargs):
class _Ctx:
def __enter__(self):
class _Run:
id = None
return _Run()
def __exit__(self, exc_type, exc, tb):
return False
return _Ctx()
trace = _Noop()
Client = None
# -----------------------------------------------------------------------------
# Environment Setup (non-destructive: set defaults if not present)
# -----------------------------------------------------------------------------
DEFAULT_LANGSMITH_API_KEY = "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"
DEFAULT_PROJECT = "medical_chatbot"
os.environ.setdefault("LANGSMITH_TRACING_V2", "true")
os.environ.setdefault("LANGSMITH_API_KEY", DEFAULT_LANGSMITH_API_KEY)
os.environ.setdefault("LANGCHAIN_PROJECT", DEFAULT_PROJECT) # tracing
os.environ.setdefault("LANGSMITH_PROJECT", DEFAULT_PROJECT) # feedback
# -----------------------------------------------------------------------------
# LangSmith Client
# -----------------------------------------------------------------------------
_langsmith_client: Optional[Client] = None
try:
if Client is not None:
_langsmith_client = Client(
api_url="https://api.smith.langchain.com",
api_key=os.environ.get("LANGSMITH_API_KEY"),
)
except Exception:
_langsmith_client = None
# -----------------------------------------------------------------------------
# Conversation Tracker for session metadata
# -----------------------------------------------------------------------------
class ConversationTracker:
def __init__(self) -> None:
self.session_id = str(uuid.uuid4())
self.conversation_count = 0
def start_new_session(self) -> None:
self.session_id = str(uuid.uuid4())
self.conversation_count = 0
def get_session_metadata(self, increment: bool = False) -> Dict[str, Any]:
if increment:
self.conversation_count += 1
return {
"session_id": self.session_id,
"conversation_count": self.conversation_count,
"application": os.environ.get("LANGSMITH_PROJECT", DEFAULT_PROJECT),
"version": "1.0",
}
conversation_tracker = ConversationTracker()
# -----------------------------------------------------------------------------
# Helper function to safely log feedback to LangSmith
# -----------------------------------------------------------------------------
def log_to_langsmith(key: str, value: dict, run_id: Optional[str] = None) -> None:
client = _langsmith_client
if not client:
return
try:
client.create_feedback(
run_id=run_id,
key=key,
value=value,
project=os.environ.get("LANGSMITH_PROJECT", DEFAULT_PROJECT),
)
except Exception:
# Swallow logging errors to avoid breaking the app
pass
__all__ = [
"traceable",
"trace",
"Client",
"conversation_tracker",
"log_to_langsmith",
]
|