Helicone LLM API网关与可观测性

# Helicone接入示例：仅需修改base_urlimport openai # 原始方式（无监控）# client = openai.OpenAI(api_key="sk-xxx") # 接入Helicone后（零侵入）client = openai.OpenAI(    api_key="sk-xxx",    base_url="https://oai.helicone.ai/v1",    default_headers={        "Helicone-Auth": "Bearer HELICONE_API_KEY",        "Helicone-Property-App": "my-llm-app",        "Helicone-Property-Environment": "production",        "Helicone-User-Id": "user-12345"    }) response = client.chat.completions.create(    model="gpt-4",    messages=[{"role": "user", "content": "Hello"}])print(response.choices[0].message.content)

import requestsimport time class HeliconeLogger:    def __init__(self):        self.logs = []        def log_request(self, request_data, response_data, latency_ms):        log_entry = {            "timestamp": time.time(),            "model": request_data.get("model", "unknown"),            "prompt_tokens": response_data.get("usage", {}).get("prompt_tokens", 0),            "completion_tokens": response_data.get("usage", {}).get("completion_tokens", 0),            "total_tokens": response_data.get("usage", {}).get("total_tokens", 0),            "latency_ms": latency_ms,            "status_code": response_data.get("status_code", 200),            "user_id": request_data.get("headers", {}).get("Helicone-User-Id", ""),            "properties": {                k.replace("Helicone-Property-", ""): v                for k, v in request_data.get("headers", {}).items()                if k.startswith("Helicone-Property-")            }        }        self.logs.append(log_entry)        return log_entry        def query_logs(self, model=None, user_id=None, min_latency=None):        results = self.logs        if model:            results = [l for l in results if l["model"] == model]        if user_id:            results = [l for l in results if l["user_id"] == user_id]        if min_latency:            results = [l for l in results if l["latency_ms"] >= min_latency]        return results logger = HeliconeLogger()entry = logger.log_request(    {"model": "gpt-4", "headers": {"Helicone-User-Id": "user-001", "Helicone-Property-App": "chatbot"}},    {"usage": {"prompt_tokens": 50, "completion_tokens": 120, "total_tokens": 170}, "status_code": 200},    latency_ms=1500)print(f"Logged: model={entry['model']}, tokens={entry['total_tokens']}, latency={entry['latency_ms']}ms")slow = logger.query_logs(user_id="user-001", min_latency=1000)print(f"Slow requests for user-001: {len(slow)}")

import hashlibimport mathfrom typing import Dict, List, Optional class SemanticCache:    def __init__(self, similarity_threshold: float = 0.95, ttl_seconds: int = 3600):        self.threshold = similarity_threshold        self.ttl = ttl_seconds        self.cache: Dict[str, dict] = {}        def _exact_hash(self, prompt: str) -> str:        return hashlib.sha256(prompt.encode()).hexdigest()        def _cosine_similarity(self, vec_a: List[float], vec_b: List[float]) -> float:        dot = sum(a * b for a, b in zip(vec_a, vec_b))        norm_a = math.sqrt(sum(a * a for a in vec_a))        norm_b = math.sqrt(sum(b * b for b in vec_b))        return dot / (norm_a * norm_b) if norm_a > 0 and norm_b > 0 else 0.0        def _mock_embed(self, text: str) -> List[float]:        h = hashlib.md5(text.encode()).hexdigest()        return [int(h[i:i+2], 16) / 255.0 for i in range(0, 32, 2)]        def get(self, prompt: str) -> Optional[str]:        import time        exact_key = self._exact_hash(prompt)        if exact_key in self.cache:            entry = self.cache[exact_key]            if time.time() - entry["timestamp"] < self.ttl:                return entry["response"]        query_emb = self._mock_embed(prompt)        best_match, best_sim = None, 0.0        for key, entry in self.cache.items():            if time.time() - entry["timestamp"] > self.ttl:                continue            sim = self._cosine_similarity(query_emb, entry["embedding"])            if sim > best_sim and sim >= self.threshold:                best_match, best_sim = entry, sim        return best_match["response"] if best_match else None        def put(self, prompt: str, response: str):        import time        key = self._exact_hash(prompt)        self.cache[key] = {            "response": response,            "embedding": self._mock_embed(prompt),            "timestamp": time.time()        } cache = SemanticCache(similarity_threshold=0.95)cache.put("什么是机器学习?", "机器学习是AI的一个分支...")result = cache.get("机器学习是什么?")print(f"Cache hit: {result is not None}, result: {result}")

import randomfrom typing import Dict, List class ModelRouter:    def __init__(self):        self.routes: List[Dict] = []        self.fallbacks: Dict[str, List[str]] = {}        def add_route(self, condition: str, model: str, weight: float = 1.0):        self.routes.append({"condition": condition, "model": model, "weight": weight})        def set_fallbacks(self, primary: str, fallback_list: List[str]):        self.fallbacks[primary] = fallback_list        def select_model(self, prompt: str) -> str:        for route in self.routes:            if route["condition"] == "*":                continue            if route["condition"] in prompt:                return route["model"]        weighted = [r for r in self.routes if r["condition"] == "*"]        if weighted:            total = sum(r["weight"] for r in weighted)            r = random.uniform(0, total)            cumulative = 0            for route in weighted:                cumulative += route["weight"]                if r <= cumulative:                    return route["model"]        return "gpt-3.5-turbo"        def call_with_failover(self, model: str, prompt: str) -> dict:        models_to_try = [model] + self.fallbacks.get(model, [])        for m in models_to_try:            try:                return {"model": m, "content": f"Response from {m}", "status": 200}            except Exception:                continue        return {"model": "none", "content": "All models failed", "status": 503} router = ModelRouter()router.add_route("代码生成", "codellama-34b")router.add_route("*", "gpt-4", weight=0.3)router.add_route("*", "gpt-3.5-turbo", weight=0.7)router.set_fallbacks("gpt-4", ["gpt-3.5-turbo", "claude-2"])model = router.select_model("请帮我做代码生成")print(f"Selected model: {model}")result = router.call_with_failover("gpt-4", "Hello")print(f"Result from: {result['model']}")

import hashlibimport mathimport timeimport randomfrom typing import Dict, List, Optional class HeliconeGateway:    """Helicone API网关模拟器：集成缓存、路由和监控"""        def __init__(self):        self.cache: Dict[str, dict] = {}        self.logs: List[dict] = []        self.routes: List[dict] = [            {"condition": "*", "model": "gpt-4", "weight": 0.3},            {"condition": "*", "model": "gpt-3.5-turbo", "weight": 0.7}        ]        self.cost_per_token = {"gpt-4": 0.00006, "gpt-3.5-turbo": 0.000002}        def _embed(self, text: str) -> List[float]:        h = hashlib.md5(text.encode()).hexdigest()        return [int(h[i:i+2], 16) / 255.0 for i in range(0, 32, 2)]        def _cosine_sim(self, a: List[float], b: List[float]) -> float:        dot = sum(x * y for x, y in zip(a, b))        na = math.sqrt(sum(x * x for x in a))        nb = math.sqrt(sum(x * x for x in b))        return dot / (na * nb) if na > 0 and nb > 0 else 0.0        def _semantic_cache_hit(self, prompt: str, threshold: float = 0.92) -> Optional[str]:        emb = self._embed(prompt)        for key, entry in self.cache.items():            if time.time() - entry["ts"] > entry.get("ttl", 3600):                continue            if self._cosine_sim(emb, entry["emb"]) >= threshold:                return entry["response"]        return None        def _select_model(self, prompt: str) -> str:        total_w = sum(r["weight"] for r in self.routes)        r = random.uniform(0, total_w)        cum = 0        for route in self.routes:            cum += route["weight"]            if r <= cum:                return route["model"]        return "gpt-3.5-turbo"        def process_request(self, prompt: str, user_id: str = "") -> dict:        start = time.time()        cached = self._semantic_cache_hit(prompt)        if cached:            latency = (time.time() - start) * 1000            log = {"model": "cache", "tokens": 0, "latency_ms": latency, "user": user_id, "cached": True}            self.logs.append(log)            return {"content": cached, "model": "cache", "cached": True}        model = self._select_model(prompt)        response = f"Response from {model} for: {prompt[:30]}"        prompt_tokens = len(prompt.split())        completion_tokens = len(response.split())        latency = (time.time() - start) * 1000 + random.uniform(100, 2000)        self.cache[hashlib.sha256(prompt.encode()).hexdigest()] = {            "response": response, "emb": self._embed(prompt), "ts": time.time(), "ttl": 3600        }        cost = completion_tokens * self.cost_per_token.get(model, 0)        log = {"model": model, "tokens": prompt_tokens + completion_tokens,               "latency_ms": latency, "user": user_id, "cost": cost, "cached": False}        self.logs.append(log)        return {"content": response, "model": model, "cached": False, "cost": cost}        def get_stats(self) -> dict:        total = len(self.logs)        cached = sum(1 for l in self.logs if l.get("cached"))        total_cost = sum(l.get("cost", 0) for l in self.logs)        avg_latency = sum(l["latency_ms"] for l in self.logs) / total if total else 0        return {"total_requests": total, "cache_hits": cached,                "cache_hit_rate": cached / total if total else 0,                "total_cost": total_cost, "avg_latency_ms": avg_latency} gw = HeliconeGateway()prompts = ["什么是深度学习?", "深度学习是什么?", "解释神经网络", "什么是深度学习?",           "Python如何安装?", "Python安装方法", "解释Transformer架构", "什么是注意力机制?"]for p in prompts:    result = gw.process_request(p, user_id="user-001")    status = "CACHED" if result["cached"] else f"MODEL={result['model']}"    print(f"[{status}] {p} -> cost={result.get('cost', 0):.6f}")stats = gw.get_stats()print(f"\nStats: requests={stats['total_requests']}, cache_rate={stats['cache_hit_rate']:.1%}, cost=${stats['total_cost']:.6f}")