File size: 2,412 Bytes
cf5c1e2 c5563d6 cf5c1e2 c5563d6 cf5c1e2 c5563d6 cf5c1e2 c5563d6 cf5c1e2 c5563d6 cf5c1e2 c5563d6 cf5c1e2 c5563d6 cf5c1e2 |
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 |
from enum import Enum
from typing import Protocol
from mixpanel import Mixpanel
from knowlang.configs.chat_config import ChatbotAnalyticsConfig, AnalyticsProvider
class ChatFeedback(Enum):
POSITIVE = "positive"
NEGATIVE = "negative"
class AnalyticsProviderProtocol(Protocol):
"""Protocol defining what an analytics provider must implement"""
def track_event(self, event_name: str, distinct_id: str, properties: dict) -> None:
"""Track an event with the analytics provider"""
...
class NoopAnalyticsProvider:
"""Provider that does nothing - used when analytics are disabled"""
def track_event(self, event_name: str, distinct_id: str, properties: dict) -> None:
pass
class MixpanelProvider:
"""Concrete implementation for Mixpanel"""
def __init__(self, api_key: str):
self._mp = Mixpanel(api_key)
def track_event(self, event_name: str, distinct_id: str, properties: dict) -> None:
self._mp.track(
distinct_id=distinct_id,
event_name=event_name,
properties=properties
)
def create_analytics_provider(config: ChatbotAnalyticsConfig) -> AnalyticsProviderProtocol:
"""Factory function to create the appropriate analytics provider"""
if not config.enabled or not config.api_key:
return NoopAnalyticsProvider()
if config.provider == AnalyticsProvider.MIXPANEL:
return MixpanelProvider(config.api_key)
raise ValueError(f"Unsupported analytics provider: {config.provider}")
class ChatAnalytics:
def __init__(self, config: ChatbotAnalyticsConfig):
self._provider = create_analytics_provider(config)
def track_query(self, query: str, client_ip: str) -> None:
"""Track query event"""
self._provider.track_event(
event_name="chat_query",
distinct_id=str(hash(client_ip)), # Hash for privacy
properties={"query": query}
)
def track_feedback(self, like: bool, query: str, client_ip: str) -> None:
"""Track feedback event"""
self._provider.track_event(
event_name="chat_feedback",
distinct_id=str(hash(client_ip)), # Hash for privacy
properties={
"feedback": ChatFeedback.POSITIVE.value if like else ChatFeedback.NEGATIVE.value,
"query": query,
}
) |