diff --git a/.github/dependabot.yml b/.github/dependabot.yml index 4f44831c..62899f72 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -11,7 +11,6 @@ updates: separator: "/" commit-message: prefix: "chore(deps)" - prefix-major: "chore(deps)!" include: "scope" labels: - "dependencies" @@ -20,11 +19,6 @@ updates: - "sheepdestroyer" allow: - dependency-type: "direct" - ignore: - # Ignore major version updates for critical services - # Uncomment and adjust as needed - # - dependency-name: "postgres" - # update-types: ["version-update:semver-major"] # GitHub Actions dependency updates (optional - add if using GH Actions) - package-ecosystem: "github-actions" diff --git a/.gitignore b/.gitignore index d0279282..79502281 100644 --- a/.gitignore +++ b/.gitignore @@ -26,12 +26,3 @@ router/__pycache__/ router/router_stats.json router/quantization_report.txt router/router_timeline.json -router/best_free_model.json -router/free_models_roster.json - -# Hermes agent artifacts -.hermes/ -workdirs/ - -# Dataset work in progress -data/ diff --git a/.jules/bolt.md b/.jules/bolt.md new file mode 100644 index 00000000..f2935c10 --- /dev/null +++ b/.jules/bolt.md @@ -0,0 +1,3 @@ +## 2024-06-16 - Synchronous I/O in Async API Handlers +**Learning:** `save_persisted_stats()` was being called synchronously on every API request, cache hit, and tool usage log, triggering blocking disk I/O in the main event loop. +**Action:** Always throttle or batch background telemetry writes in async Python applications to prevent blocking the event loop under load. diff --git a/litellm/config.yaml b/litellm/config.yaml index c27b47c7..fba489f9 100644 --- a/litellm/config.yaml +++ b/litellm/config.yaml @@ -143,19 +143,10 @@ redis_settings: redis_port: 6379 router_settings: allowed_fails: 2 - cooldown_time: 300 + cooldown_time: 30 enable_pre_call_checks: false num_retries: 1 routing_strategy: latency-based-routing - # Per-error-type cooldown thresholds (overrides allowed_fails for specific errors). - # Upstream rate limits ("temporarily rate-limited upstream") can last minutes — - # a 30s cooldown just wastes retries. 300s gives providers time to clear the limit. - # RateLimitError: cooldown on FIRST 429 (upstream rate limits persist for minutes — - # allowing even 1 retry wastes a request against a provider that's still throttling) - allowed_fails_policy: - RateLimitErrorAllowedFails: 0 - TimeoutErrorAllowedFails: 3 - BadRequestErrorAllowedFails: 1 vector_store_settings: collection_name: litellm_semantic_cache connection_string: postgresql://postgres:***@127.0.0.1:5432/postgres diff --git a/router/Containerfile b/router/Containerfile index b21bf253..a874e1c1 100644 --- a/router/Containerfile +++ b/router/Containerfile @@ -6,7 +6,7 @@ WORKDIR /app RUN pip install --no-cache-dir fastapi uvicorn httpx pyyaml python-multipart asyncpg langfuse # Copy all source in one layer — removes dead config COPY (volume-mounted at runtime) -COPY main.py agy_proxy.py circuit_breaker.py aa_scores.json free_models_roster.json best_free_model.json /app/ +COPY main.py agy_proxy.py circuit_breaker.py aa_scores.json /app/ EXPOSE 5000 diff --git a/router/best_free_model.json b/router/best_free_model.json new file mode 100644 index 00000000..b816c45f --- /dev/null +++ b/router/best_free_model.json @@ -0,0 +1,8 @@ +{ + "id": "nvidia/nemotron-3-ultra-550b-a55b:free", + "name": "NVIDIA: Nemotron 3 Ultra (free)", + "score": 48.0, + "context_length": 1000000, + "is_fallback": false, + "updated_at": "2026-06-16T15:43:39.475769Z" +} \ No newline at end of file diff --git a/router/config.yaml b/router/config.yaml index 5606ae8e..6da414b0 100644 --- a/router/config.yaml +++ b/router/config.yaml @@ -7,19 +7,17 @@ router: router_model: api_base: "http://127.0.0.1:8080/v1" api_key: "local-token" - model: "gemma4-26a4b-routing" + model: "qwen-2b-routing" classification_rules: system_prompt: | - Classify the coding task complexity. Output ONLY the tier name. - - agent-simple-core: trivial one-liners, syntax fixes, single-line edits - agent-medium-core: single-function changes, light refactoring, simple tests - agent-complex-core: multi-file changes, algorithmic work, data pipelines - agent-reasoning-core: deep analysis, architecture decisions, debugging complex systems - agent-advanced-core: system-level architecture, cross-cutting concerns, novel design - - Task: + Analyze the request complexity. Respond with exactly one of: + - simple boilerplate: agent-simple-core + - moderate complexity: agent-medium-core + - deep algorithms: agent-complex-core + - heavy reasoning, advanced architecture, deep multi-step analysis: agent-reasoning-core + - extremely complex across many files: agent-advanced-core + Do not add markdown formatting. Only the model name string. # 5-tier classifier dispatch backends — all proxied through LiteLLM (port 4000). # Tiers (complexity ↑): @@ -51,9 +49,6 @@ backends: - name: "ollama-deepseek-v4-pro" api_base: "http://127.0.0.1:4000/v1" api_key: "DYNAMIC_LITELLM_MASTER_KEY_PLACEHOLDER" - - name: "ollama-deepseek-v4-flash" - api_base: "http://127.0.0.1:4000/v1" - api_key: "DYNAMIC_LITELLM_MASTER_KEY_PLACEHOLDER" - name: "openrouter-auto" api_base: "http://127.0.0.1:4000/v1" api_key: "DYNAMIC_LITELLM_MASTER_KEY_PLACEHOLDER" diff --git a/router/free_models_roster.json b/router/free_models_roster.json index f7cedb5b..a14ce1ea 100644 --- a/router/free_models_roster.json +++ b/router/free_models_roster.json @@ -120,6 +120,24 @@ "score": 25.0, "context_length": 32768 }, + { + "id": "meta-llama/llama-3.3-70b-instruct:free", + "name": "Meta: Llama 3.3 70B Instruct (free)", + "score": 25.0, + "context_length": 131072 + }, + { + "id": "meta-llama/llama-3.2-3b-instruct:free", + "name": "Meta: Llama 3.2 3B Instruct (free)", + "score": 25.0, + "context_length": 131072 + }, + { + "id": "nousresearch/hermes-3-llama-3.1-405b:free", + "name": "Nous: Hermes 3 405B Instruct (free)", + "score": 25.0, + "context_length": 131072 + }, { "id": "poolside/laguna-xs.2:free", "name": "Poolside: Laguna XS.2 (free)", @@ -139,6 +157,6 @@ "context_length": 32768 } ], - "updated_at": "2026-06-17T20:17:56.981195Z", - "count": 23 + "updated_at": "2026-06-16T15:40:59.687282Z", + "count": 26 } \ No newline at end of file diff --git a/router/main.py b/router/main.py index 64efce64..69bf9b09 100644 --- a/router/main.py +++ b/router/main.py @@ -151,8 +151,20 @@ def load_persisted_stats(): except Exception as e: logger.error(f"Failed to load persisted stats: {e}") -def save_persisted_stats(): +_last_stats_save = 0.0 + +def save_persisted_stats(force=False): """Persists current statistics in-memory structure to disk securely.""" + global _last_stats_save + import time + now = time.time() + + # Throttle disk writes to max once per 2 seconds, unless forced + if not force and (now - _last_stats_save < 2.0): + return + + _last_stats_save = now + try: os.makedirs(os.path.dirname(STATS_JSON_PATH), exist_ok=True) with open(STATS_JSON_PATH, "w") as f: @@ -191,23 +203,6 @@ async def sync_adaptive_router_roster(master_key: str): # Skip internal OpenRouter encoded IDs that LiteLLM can't map to a provider if not mid or (len(mid) > 64 and "/" not in mid): continue - - # 1. Enforce Tool/Function Calling Support - supported_params = m.get('supported_parameters') or [] - if 'tools' not in supported_params: - logger.info(f"🚫 Skipping {mid} — Model does not support tool calling.") - continue - - # 2. Denylist: skip models known to be problematic (stale, wrong context_length, etc.) - # llama-3.3-70b reports 131K ctx but actual endpoint enforces 65K → context_limit errors. - # All meta-llama and llama-derived models are too old and unreliable on free tier. - _denylist_prefixes = ( - "meta-llama/", "nousresearch/hermes-3-llama", - ) - if any(mid.startswith(p) for p in _denylist_prefixes): - logger.info(f"🚫 Skipping {mid} — denylisted (stale/unreliable free tier model)") - continue - pricing = m.get("pricing", {}) if pricing.get("prompt") in ("0", 0, "0.0", 0.0) and pricing.get("completion") in ("0", 0, "0.0", 0.0): try: @@ -358,12 +353,8 @@ async def check_http_endpoint(url: str) -> bool: except Exception: return False -async def classify_request(prompt: str, bypass_cache: bool = False, langfuse_trace_id: str | None = None) -> tuple[str, float, bool, str]: - """Queries the local fast Qwen instance to classify request complexity with TTL caching. - - When langfuse_trace_id is provided, the classifier HTTP call is wrapped in a child - observation (span) so latency and output appear as a nested span in Langfuse traces. - """ +async def classify_request(prompt: str, bypass_cache: bool = False) -> tuple[str, float, bool, str]: + """Queries the local fast Qwen instance to classify request complexity with TTL caching.""" global triage_cache, stats # Normalize the prompt text for cache mapping @@ -396,31 +387,16 @@ async def classify_request(prompt: str, bypass_cache: bool = False, langfuse_tra payload = { "model": router_model_name, "messages": [ - {"role": "user", "content": system_prompt + prompt} + {"role": "system", "content": system_prompt}, + {"role": "user", "content": prompt} ], "temperature": 0.0, "max_tokens": 15, + "grammar": 'root ::= "agent-simple-core" | "agent-medium-core" | "agent-complex-core" | "agent-reasoning-core" | "agent-advanced-core"' } headers = {"Authorization": f"Bearer {router_api_key}"} logger.info(f"Classifying intent via {router_api_base} using model {router_model_name}...") - - # --- Langfuse child span: classifier call --- - class_span_obj = None - if langfuse_trace_id: - lf_cls = get_langfuse() - if lf_cls: - try: - class_span_obj = lf_cls.start_observation( - trace_context={"trace_id": langfuse_trace_id}, - name="classifier-qwen", - input=prompt[:200], - metadata={"model": router_model_name}, - level="DEFAULT", - ) - except Exception: - pass - response = await client.post( f"{router_api_base}/chat/completions", json=payload, @@ -430,14 +406,6 @@ async def classify_request(prompt: str, bypass_cache: bool = False, langfuse_tra latency = (time.time() - start_time) * 1000.0 if response.status_code != 200: - if class_span_obj: - try: - class_span_obj.end( - output={"status": response.status_code, "error": "classification_failed"}, - metadata={"latency_ms": latency}, - ) - except Exception: - pass logger.error(f"Classification failed with status {response.status_code}: {response.text}") return "agent-advanced-core", latency, False, "advanced (fallback)" @@ -457,16 +425,6 @@ async def classify_request(prompt: str, bypass_cache: bool = False, langfuse_tra decision = content_clean else: decision = "agent-advanced-core" - - # Finalize classifier child span - if class_span_obj: - try: - class_span_obj.end( - output={"tier": decision, "raw": raw_result}, - metadata={"latency_ms": latency}, - ) - except Exception: - pass # Store in cache triage_cache[normalized_prompt] = (decision, time.time()) @@ -626,12 +584,16 @@ def record_tool_usage(tool_name: str, prompt_tokens: int, completion_tokens: int if len(stats["timeline"]) > 15: stats["timeline"].pop(0) save_persisted_stats() - try: - timeline_path = os.path.join(os.path.dirname(CONFIG_PATH), "router_timeline.json") - with open(timeline_path, "w") as f: - json.dump(stats["timeline"], f) - except Exception: - pass # disk persistence failure is non-fatal + # Throttled timeline save (handled together with stats save internally, + # but we also explicitly throttle this specific file) + if time.time() - getattr(record_tool_usage, "_last_save", 0.0) >= 2.0: + try: + timeline_path = os.path.join(os.path.dirname(CONFIG_PATH), "router_timeline.json") + with open(timeline_path, "w") as f: + json.dump(stats["timeline"], f) + record_tool_usage._last_save = time.time() + except Exception: + pass # disk persistence failure is non-fatal def get_goose_sessions() -> list: """Queries the live mounted SQLite goose database to fetch the latest agentic sessions.""" @@ -795,12 +757,6 @@ async def get_best_free_model() -> dict: for m in data: mid = m.get("id", "") - # Denylist: skip stale/unreliable free tier models - _denylist_prefixes = ( - "meta-llama/", "nousresearch/hermes-3-llama", - ) - if any(mid.startswith(p) for p in _denylist_prefixes): - continue pricing = m.get("pricing", {}) # Standard pricing is string or float p_prompt = pricing.get("prompt") @@ -982,29 +938,11 @@ async def chat_completions(request: Request): client_model = body.get("model", "llm-routing-auto-free") - # --- Langfuse parent trace: create early so child spans can reference it --- - langfuse_trace_id = None - parent_obs = None - lf = get_langfuse() - if lf: - try: - langfuse_trace_id = lf.create_trace_id(seed=f"triage_{stats['total_requests']}") - parent_obs = lf.start_observation( - trace_context={"trace_id": langfuse_trace_id}, - name=f"triage-{client_model}", - input=last_user_message[:200], - level="DEFAULT", - ) - except Exception as e: - logger.warning(f"Langfuse trace init failed (non-fatal): {e}") - langfuse_trace_id = None - parent_obs = None - if client_model in AUTO_MODELS: # Full pipeline: classify → route to best tier bypass_cache = request.headers.get("x-bypass-cache") == "true" target_model, triage_latency, was_cache_hit, raw_classification = await classify_request( - last_user_message, bypass_cache=bypass_cache, langfuse_trace_id=langfuse_trace_id + last_user_message, bypass_cache=bypass_cache ) logger.info(f"Triage decision (auto): Routing to -> '{target_model}'") elif client_model in DIRECT_TIERS: @@ -1039,19 +977,29 @@ async def chat_completions(request: Request): stats["advanced_requests"] = stats.get("advanced_requests", 0) + 1 save_persisted_stats() - # Update the parent Langfuse observation with classification results - if parent_obs: + # Push classification trace to Langfuse (v4 API: start_observation) + langfuse_trace_id = None + lf = get_langfuse() + if lf: try: - parent_obs.update( + # Create a trace ID first, then start an observation with it + trace_id = lf.create_trace_id(seed=f"triage_{stats['total_requests']}") + lf.start_observation( + trace_context={"trace_id": trace_id}, + name=f"triage-{target_model}", + input=last_user_message[:200], output={"tier": target_model, "raw": raw_classification}, metadata={ "triage_latency_ms": round(triage_latency, 2), "cache_hit": was_cache_hit, "total_requests": stats["total_requests"], }, + level="DEFAULT", ) + lf.flush() + langfuse_trace_id = trace_id except Exception as e: - logger.warning(f"Langfuse trace update failed (non-fatal): {e}") + logger.warning(f"Langfuse trace push failed (non-fatal): {e}") # --- PREMIUM PROXY ROUTES --- # agy: triggered unconditionally for llm-routing-agy (direct). @@ -1103,22 +1051,6 @@ async def chat_completions(request: Request): session_id = hashlib.md5(fingerprint.encode()).hexdigest() if last_prompt: - # --- Langfuse child span: agy proxy --- - agy_span_obj = None - if langfuse_trace_id: - lf_agy = get_langfuse() - if lf_agy: - try: - agy_span_obj = lf_agy.start_observation( - trace_context={"trace_id": langfuse_trace_id}, - name="agy-proxy", - input=last_prompt[:200], - metadata={"tier": target_model}, - level="DEFAULT", - ) - except Exception: - pass - agy_response = await try_agy_proxy( prompt=last_prompt, messages=messages, @@ -1138,16 +1070,6 @@ async def chat_completions(request: Request): ) logger.info(f"✅ agy proxy succeeded: {model_name}, {latency_ms:.0f}ms") - # Finalize agy span - if agy_span_obj: - try: - agy_span_obj.end( - output={"model": model_name, "tokens": completion_tokens}, - metadata={"latency_ms": latency_ms, "tier": target_model}, - ) - except Exception: - pass - if body.get("stream", False): content = agy_response.get("choices", [{}])[0].get("message", {}).get("content", "") async def agy_stream_generator(): @@ -1188,24 +1110,8 @@ async def agy_stream_generator(): else: return agy_response except ImportError: - if agy_span_obj: - try: - agy_span_obj.end( - output={"error": "module_not_available"}, - metadata={"status": "skipped"}, - ) - except Exception: - pass logger.warning("agy_proxy module not available, falling back to LiteLLM") except Exception as e: - if agy_span_obj: - try: - agy_span_obj.end( - output={"error": str(e)[:200]}, - metadata={"status": "failed"}, - ) - except Exception: - pass logger.error(f"agy proxy failed: {e}, falling back to LiteLLM") # --- OLLAMA (via LiteLLM) --- @@ -1237,22 +1143,6 @@ async def agy_stream_generator(): # LiteLLM maps this to Nemotron → Kimi → GPT-OSS → local Qwen logger.info(f"Proxying to LiteLLM as model={target_model}") - # --- Langfuse child span: LiteLLM proxy --- - litellm_span_obj = None - if langfuse_trace_id: - lf_litellm = get_langfuse() - if lf_litellm: - try: - litellm_span_obj = lf_litellm.start_observation( - trace_context={"trace_id": langfuse_trace_id}, - name="litellm-proxy", - input=target_model, - metadata={"model": target_model}, - level="DEFAULT", - ) - except Exception: - pass - # Set up outgoing proxy request client = httpx.AsyncClient(timeout=3600.0) headers = {"Authorization": f"Bearer {backend_api_key}"} @@ -1263,41 +1153,9 @@ async def agy_stream_generator(): proxy_start = time.time() model_name = target_model # LiteLLM handles fallback internally - # --- Pre-screening: clamp max_tokens to fit within downstream model context limits --- - # Hermes agents set max_tokens=65536 unconditionally, but some free models have - # context limits as low as 32K. Without clamping, input+output exceeds the limit - # and the request fails with a 400 BadRequest (context_length exceeded). - # We estimate input tokens and clamp max_tokens to leave a 2K safety margin. try: body_to_send = body.copy() body_to_send["model"] = model_name - requested_max_tokens = body_to_send.get("max_tokens", 4096) - if requested_max_tokens > 32768: # Only intervene for unusually large max_tokens - # Tier-aware minimum context length (from actual roster data): - # - agent-simple-core: 32K (includes tiny liquid/dolphin models) - # - agent-medium-core+: 256K (smallest non-tiny model is nemotron-nano-omni at 256K) - # - ollama-deepseek-v4-*: 1M (DeepSeek V4 native context) - _tier_min_ctx = { - "agent-simple-core": 32768, - "ollama-deepseek-v4-pro": 1000000, - "ollama-deepseek-v4-flash": 1000000, - } - _min_ctx = _tier_min_ctx.get(model_name, 262144) - # Rough input token estimate: 1 token ≈ 4 chars of JSON - _est_input = len(json.dumps(body_to_send)) // 4 - _safe_max = max(1024, _min_ctx - _est_input - 2048) # 2K safety margin - if requested_max_tokens > _safe_max: - logger.warning( - f"⛔ Clamping max_tokens: {requested_max_tokens} → {_safe_max} " - f"(est_input={_est_input}, min_ctx={_min_ctx}, tier={model_name})" - ) - body_to_send["max_tokens"] = _safe_max - except Exception as e: - logger.warning(f"Pre-screening failed (non-fatal): {e}") - body_to_send = body.copy() - body_to_send["model"] = model_name - - try: if "metadata" not in body_to_send or not isinstance(body_to_send["metadata"], dict): body_to_send["metadata"] = {} body_to_send["metadata"]["trace_name"] = "agent-completion" @@ -1318,15 +1176,6 @@ async def stream_generator(): stats["total_proxy_time_ms"] += proxy_latency stats["avg_proxy_latency_ms"] = stats["total_proxy_time_ms"] / stats["total_requests"] record_tool_usage(active_tool, request_tokens, completion_chars // 4, model_name, proxy_latency, route="litellm_fallback") - # Finalize LiteLLM span (streaming path) - if litellm_span_obj: - try: - litellm_span_obj.end( - output={"model": model_name, "stream": True}, - metadata={"latency_ms": proxy_latency, "tokens": completion_chars // 4}, - ) - except Exception: - pass except Exception as ex: logger.error(f"Stream error: {ex}") finally: @@ -1351,15 +1200,6 @@ async def stream_generator(): prompt_tokens = usage.get("prompt_tokens", len(json.dumps(body_to_send)) // 4) completion_tokens = usage.get("completion_tokens", len(json.dumps(resp_json)) // 4) record_tool_usage(active_tool, prompt_tokens, completion_tokens, model_name, proxy_latency, route="litellm_fallback") - # Finalize LiteLLM span (non-streaming path) - if litellm_span_obj: - try: - litellm_span_obj.end( - output={"model": model_name, "tokens": completion_tokens}, - metadata={"latency_ms": proxy_latency}, - ) - except Exception: - pass return resp_json else: logger.warning(f"LiteLLM failed ({response.status_code}): {response.text[:300]}") @@ -1514,43 +1354,32 @@ async def get_dashboard(): # 2b. Fetch live llama.cpp metrics llamacpp = await get_llamacpp_metrics() - # 3. Calculative metrics — 5-tier triage table - tier_data = [ - {"tier": "agent-simple-core", "count": stats.get("simple_requests", 0), "color": "#34d399"}, - {"tier": "agent-medium-core", "count": stats.get("medium_requests", 0), "color": "#fbbf24"}, - {"tier": "agent-complex-core", "count": stats.get("complex_requests", 0), "color": "#a78bfa"}, - {"tier": "agent-reasoning-core", "count": stats.get("reasoning_requests", 0), "color": "#60a5fa"}, - {"tier": "agent-advanced-core", "count": stats.get("advanced_requests", 0), "color": "#f472b6"}, - ] - total_tier = sum(t["count"] for t in tier_data) - for t in tier_data: - t["ratio"] = (t["count"] / total_tier * 100.0) if total_tier > 0 else 0.0 - - # Build tier table rows - tier_table_rows = "" - for t in tier_data: - tier_table_rows += f""" - - - - {t['tier']} - - {t['count']} - {t['ratio']:.1f}% - """ - tier_table_html = f""" - - - - - - - - - - {tier_table_rows} - -
TierRequestsShare
""" + # 3. Calculative metrics — 5-tier triage ratios + tier_data = { + "simple": {"count": stats.get("simple_requests", 0), "label": "Simple Core (Lite/Gemma)", "color": "#34d399"}, + "medium": {"count": stats.get("medium_requests", 0), "label": "Medium Core", "color": "#fbbf24"}, + "complex": {"count": stats.get("complex_requests", 0), "label": "Complex Core (Qwen)", "color": "#a78bfa"}, + "reasoning":{"count": stats.get("reasoning_requests", 0),"label": "Reasoning Core (Claude)", "color": "#60a5fa"}, + "advanced":{"count": stats.get("advanced_requests", 0), "label": "Advanced Core (Nemotron)", "color": "#f472b6"}, + } + for k, v in tier_data.items(): + v["ratio"] = (v["count"] / stats["total_requests"] * 100.0) if stats["total_requests"] > 0 else 0.0 + + # Build 5-segment stacked bar widths for inline CSS + tier_bar_segments = "" + for k, v in tier_data.items(): + if v["ratio"] > 0: + tier_bar_segments += f'
' + + # Build 5-item legend rows + tier_legend_rows = "" + for k, v in tier_data.items(): + tier_legend_rows += f""" +
+ + {v['label']}: + {v['count']} requests ({v['ratio']:.1f}%) +
""" # 4. Generate dynamic conic-gradient CSS background for the Pie Chart pie_gradient = get_pie_chart_gradient() @@ -2199,7 +2028,12 @@ def src_badge(label, color):
{src_badge('ROUTER', '#818cf8')} Triage Routing Split
- {tier_table_html} +
+ {tier_bar_segments} +
+
+ {tier_legend_rows} +
diff --git a/scripts/extract_prompts.py b/scripts/extract_prompts.py deleted file mode 100644 index 8c3e454d..00000000 --- a/scripts/extract_prompts.py +++ /dev/null @@ -1,136 +0,0 @@ -"""Extract 300 meaningful coding prompts from Langfuse observations.""" -import os, base64, json, urllib.request, urllib.error, sys, time -from pathlib import Path - -env = {} -env_path = Path(__file__).resolve().parent.parent / ".env" -with open(env_path) as f: - for line in f: - line = line.strip() - if line and not line.startswith('#') and '=' in line: - k, v = line.split('=', 1) - env[k.strip()] = v.strip().strip('"').strip("'") - -pk = env['LANGFUSE_PUBLIC_KEY'] -sk = env['LANGFUSE_SECRET_KEY'] -auth = base64.b64encode(f"{pk}:{sk}".encode()).decode() -base_url = "http://localhost:3001" - -# Trivial test prompts to skip -TRIVIAL_PATTERNS = [ - "say hello", "hi", "test", "ping", "hello", "hey", "what's up", - "how are you", "good morning", "good afternoon", "good evening", - "what model are you", "what llm are you", "who are you", - "tell me a joke", "what is 2+2", "what is the capital", -] - -def is_trivial(prompt): - """Filter out test pings and trivial prompts.""" - lower = prompt.strip().lower() - if len(lower) < 20: - return True - for pat in TRIVIAL_PATTERNS: - if pat in lower and len(lower) < 50: - return True - return False - -def fetch_observations(page=1, limit=50): - """Fetch one page of DEFAULT-level litellm-acompletion observations.""" - url = f"{base_url}/api/public/observations?limit={limit}&page={page}&orderBy=timestamp.desc&level=DEFAULT&name=litellm-acompletion" - req = urllib.request.Request(url) - req.add_header("Authorization", f"Basic {auth}") - with urllib.request.urlopen(req, timeout=15) as resp: - return json.loads(resp.read()) - -def extract_user_prompt(obs): - """Extract the last user message from an observation's input.""" - inp = obs.get('input') - if not inp: - return None - if isinstance(inp, str): - try: - inp = json.loads(inp) - except: - return None - if not isinstance(inp, dict): - return None - messages = inp.get('messages', []) - if not messages: - return None - for msg in reversed(messages): - if isinstance(msg, dict) and msg.get('role') == 'user': - content = msg.get('content', '') - if isinstance(content, str) and len(content.strip()) > 3: - return content.strip() - return None - -print("Extracting meaningful coding prompts from Langfuse observations...") -prompts = [] -seen = set() -page = 1 -target = 300 -max_pages = 50 # 50 pages × 50 = 2500 observations - -while len(prompts) < target and page <= max_pages: - try: - data = fetch_observations(page=page, limit=50) - except Exception as e: - print(f" Page {page} failed: {e}") - break - - obs_list = data.get('data', []) - total_available = data.get('meta', {}).get('totalItems', 0) - - if not obs_list: - print(f" Page {page}: empty, stopping") - break - - added_this_page = 0 - for obs in obs_list: - if len(prompts) >= target: - break - - prompt = extract_user_prompt(obs) - if not prompt: - continue - - # Skip trivial test pings - if is_trivial(prompt): - continue - - # Deduplicate - norm = prompt.strip().lower() - if norm in seen: - continue - seen.add(norm) - - prompts.append({ - "prompt": prompt, - "observation_id": obs.get('id', ''), - "trace_id": obs.get('traceId', ''), - "timestamp": obs.get('startTime', ''), - "model": obs.get('model', ''), - }) - added_this_page += 1 - - print(f" Page {page}: {len(obs_list)} obs, +{added_this_page} new → {len(prompts)} total (of {total_available} available)") - page += 1 - time.sleep(0.1) # gentle rate limit - -# Save -out_dir = Path(__file__).resolve().parent.parent / "data" -out_dir.mkdir(exist_ok=True) -out_path = out_dir / "raw_prompts.json" - -with open(out_path, 'w') as f: - json.dump(prompts, f, indent=2, ensure_ascii=False) - -print(f"\nSaved {len(prompts)} prompts to {out_path}") - -# Stats -lengths = [len(p['prompt']) for p in prompts] -print(f"Length range: {min(lengths)}-{max(lengths)} chars") -print(f"Avg length: {sum(lengths)/len(lengths):.0f} chars") -print(f"\nSample (first 10):") -for p in prompts[:10]: - print(f" [{p['timestamp'][:19]}] {p['prompt'][:120]}...")