PostHog：开源产品分析与LLM可观测性

import jsonimport timeimport hashlib class PostHogEvent:    def __init__(self, project_api_key):        self.api_key = project_api_key        self.events_buffer = []        self.feature_flags = {}     def capture(self, event_name, user_id, properties=None):        event = {            "event": event_name,            "distinct_id": user_id,            "timestamp": time.time(),            "properties": properties or {}        }        self.events_buffer.append(event)        return event     def capture_llm_call(self, user_id, model_name, input_tokens,                         output_tokens, latency_ms, error=None):        properties = {            "model": model_name,            "input_tokens": input_tokens,            "output_tokens": output_tokens,            "total_tokens": input_tokens + output_tokens,            "latency_ms": latency_ms,            "error": error,            "is_llm_call": True        }        return self.capture("llm_call", user_id, properties)     def capture_user_action(self, user_id, action, page=None):        properties = {"action": action, "page": page}        return self.capture("user_action", user_id, properties)     def set_feature_flag(self, flag_key, enabled=True, rollout_percentage=100):        self.feature_flags[flag_key] = {            "enabled": enabled,            "rollout_percentage": rollout_percentage        }     def is_feature_enabled(self, flag_key, user_id):        flag = self.feature_flags.get(flag_key)        if not flag or not flag["enabled"]:            return False        hash_val = int(hashlib.md5(            f"{flag_key}:{user_id}".encode()).hexdigest(), 16) % 100        return hash_val < flag["rollout_percentage"]     def get_llm_metrics(self):        llm_events = [e for e in self.events_buffer                      if e.get("properties", {}).get("is_llm_call")]        if not llm_events:            return {"total_calls": 0}        total_tokens = sum(            e["properties"]["total_tokens"] for e in llm_events)        avg_latency = sum(            e["properties"]["latency_ms"] for e in llm_events) / len(llm_events)        error_count = sum(            1 for e in llm_events if e["properties"].get("error"))        return {            "total_calls": len(llm_events),            "total_tokens": total_tokens,            "avg_latency_ms": round(avg_latency, 2),            "error_rate": round(error_count / len(llm_events) * 100, 2)        }     def get_user_journey(self, user_id):        user_events = [e for e in self.events_buffer                       if e["distinct_id"] == user_id]        user_events.sort(key=lambda e: e["timestamp"])        journey = []        for e in user_events:            journey.append({                "event": e["event"],                "time": e["timestamp"],                "props": e.get("properties", {})            })        return journey     def correlate_user_llm(self, user_id):        journey = self.get_user_journey(user_id)        metrics = self.get_llm_metrics()        user_llm_events = [            step for step in journey            if step["props"].get("is_llm_call")        ]        return {            "user_id": user_id,            "total_events": len(journey),            "llm_calls": len(user_llm_events),            "llm_metrics": metrics,            "journey": journey        } ph = PostHogEvent(project_api_key="phc_xxx")ph.set_feature_flag("new_ai_assistant", enabled=True, rollout_percentage=50)for i in range(100):    uid = f"user_{i}"    ph.capture_user_action(uid, "page_view", page="chat")    if ph.is_feature_enabled("new_ai_assistant", uid):        ph.capture_user_action(uid, "click_send", page="chat")        latency = 200 + (i % 10) * 50        err = "timeout" if i % 20 == 0 else None        ph.capture_llm_call(uid, "gpt-4", input_tokens=150,                           output_tokens=300, latency_ms=latency, error=err)    else:        ph.capture_user_action(uid, "page_exit", page="chat")metrics = ph.get_llm_metrics()print(f"LLM Metrics: {json.dumps(metrics, indent=2)}")user_corr = ph.correlate_user_llm("user_5")print(f"User 5 LLM calls: {user_corr['llm_calls']}")