#!/usr/bin/env python3 """ Proxy Health Check Tool - 评估本地代理和 OpenRouter 的健康状况 """ import os import sys sys.path.append(os.getcwd()) import time from openai import OpenAI from slack_bot.llm.gemini import get_system_instruction from slack_bot.tools.groups import get_tool_preset import json print("=" * 80) print("Gemini 代理健康检查工具") print("=" * 80) # ========== 配置 ========== LOCAL_CONFIG = { "name": "本地代理", "base_url": "http://127.0.0.1:8045", "api_key": "sk-457cbcd2e0a4467e90db1af0ae65748e", "model": "gemini-3-pro-high" } OPENROUTER_CONFIG = { "name": "OpenRouter", "base_url": "https://openrouter.ai/api", "api_key": "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", "model": "google/gemini-3-flash-preview" } # ========== 测试用例 ========== TESTS = [ { "name": "简单对话（无工具）", "system": "You are a helpful assistant.", "user": "Hello", "tools": None, "expected": "text" }, { "name": "健康查询（无工具）", "system": get_system_instruction(), "user": "今天睡得怎么样？", "tools": None, "expected": "text" }, { "name": "简单工具调用（1工具）", "system": "You are a health assistant. Call tools to get data.", "user": "今天睡得怎么样？", "tools": [{ "type": "function", "function": { "name": "get_health_data", "description": "Get health data", "parameters": { "type": "object", "properties": {"date": {"type": "string"}}, "required": ["date"] } } }], "expected": "tool_call" }, { "name": "完整健康查询（6工具）", "system": get_system_instruction(), "user": "今天睡得怎么样？", "tools": get_tool_preset("light")[0], "expected": "tool_call" }, { "name": "完整健康查询（21工具）", "system": get_system_instruction(), "user": "今天睡得怎么样？", "tools": get_tool_preset("all")[0], "expected": "tool_call" } ] def run_test(config, test): """运行单个测试""" headers = {} if "openrouter" in config["base_url"]: headers = { "HTTP-Referer": "https://github.com/butler", "X-Title": "Butler Health Check" } client = OpenAI( api_key=config["api_key"], base_url=f"{config['base_url']}/v1", default_headers=headers ) try: start = time.time() kwargs = { "model": config["model"], "messages": [ {"role": "system", "content": test["system"]}, {"role": "user", "content": test["user"]} ] } if test["tools"]: kwargs["tools"] = test["tools"] response = client.chat.completions.create(**kwargs) elapsed = time.time() - start msg = response.choices[0].message finish = response.choices[0].finish_reason has_content = bool(msg.content and msg.content.strip()) has_tools = bool(msg.tool_calls) # 判断结果 if test["expected"] == "tool_call": success = has_tools status = "✅" if has_tools else "❌" detail = f"tool_calls={len(msg.tool_calls)}" if has_tools else f"no tool call (text={len(msg.content or '')} chars)" else: # expected == "text" success = has_content status = "✅" if has_content else "❌" detail = f"text={len(msg.content)} chars" if has_content else "empty" return { "success": success, "status": status, "detail": detail, "elapsed": elapsed, "finish": finish } except Exception as e: return { "success": False, "status": "❌", "detail": f"ERROR: {str(e)[:50]}", "elapsed": 0, "finish": "error" } def evaluate_config(config): """评估一个配置""" print(f"\n{'=' * 80}") print(f"测试: {config['name']}") print(f"{'=' * 80}") print(f"Base URL: {config['base_url']}") print(f"Model: {config['model']}") print() results = [] for i, test in enumerate(TESTS, 1): tool_count = len(test["tools"]) if test["tools"] else 0 print(f"{i}. {test['name']} (工具数: {tool_count}, 期望: {test['expected']})") result = run_test(config, test) results.append(result) print(f" {result['status']} {result['detail']} ({result['elapsed']:.2f}s)") # 统计 total = len(results) success_count = sum(1 for r in results if r["success"]) success_rate = success_count / total * 100 print(f"\n{'-' * 80}") print(f"总体成功率: {success_count}/{total} ({success_rate:.1f}%)") print(f"平均响应时间: {sum(r['elapsed'] for r in results) / total:.2f}s") return { "config": config, "results": results, "success_rate": success_rate, "total": total, "success_count": success_count } # ========== 执行测试 ========== print("\n开始健康检查...") print() local_report = evaluate_config(LOCAL_CONFIG) openrouter_report = evaluate_config(OPENROUTER_CONFIG) # ========== 对比分析 ========== print(f"\n{'=' * 80}") print("对比分析") print(f"{'=' * 80}") print(f"\n本地代理:") print(f" 成功率: {local_report['success_rate']:.1f}%") print(f" Function Calling: {'✅ 可用' if local_report['results'][2]['success'] else '❌ 不可用'}") print(f"\nOpenRouter:") print(f" 成功率: {openrouter_report['success_rate']:.1f}%") print(f" Function Calling: {'✅ 可用' if openrouter_report['results'][2]['success'] else '❌ 不可用'}") # ========== 健康评级 ========== print(f"\n{'=' * 80}") print("健康评级") print(f"{'=' * 80}") def grade_health(report): rate = report['success_rate'] fc_works = report['results'][2]['success'] # Test 3 是 function calling if rate >= 80 and fc_works: return "A", "优秀", "完全可用，推荐使用" elif rate >= 60 and fc_works: return "B", "良好", "基本可用，可能有小问题" elif rate >= 40: return "C", "一般", "部分功能可用，需要 SAFETY OVERRIDE" elif rate >= 20: return "D", "较差", "严重依赖 SAFETY OVERRIDE" else: return "F", "不可用", "不推荐使用" local_grade, local_label, local_desc = grade_health(local_report) or_grade, or_label, or_desc = grade_health(openrouter_report) print(f"\n本地代理: {local_grade} ({local_label})") print(f" {local_desc}") print(f"\nOpenRouter: {or_grade} ({or_label})") print(f" {or_desc}") # ========== 推荐 ========== print(f"\n{'=' * 80}") print("推荐建议") print(f"{'=' * 80}") if local_report['success_rate'] >= 80: print("\n✅ 本地代理健康状况良好，可以继续使用") elif local_report['success_rate'] >= 40: print("\n⚠️ 本地代理可用，但建议:") print(" 1. 保持 SAFETY OVERRIDE 开启") print(" 2. 监控日志中 🔧 标记的频率") print(" 3. 如果 >50% 的请求需要 SAFETY OVERRIDE，考虑切换") else: print("\n❌ 本地代理健康状况较差，强烈建议切换到 OpenRouter") if openrouter_report['success_rate'] > local_report['success_rate']: improvement = openrouter_report['success_rate'] - local_report['success_rate'] print(f"\n💡 切换到 OpenRouter 可提升 {improvement:.1f}% 的成功率") # 成本估算 print(f"\n💰 OpenRouter 成本估算:") print(f" - Gemini 3 Flash: ~$0.15/1M tokens") print(f" - 假设每天 100 次对话，平均 2K tokens") print(f" - 月成本约: $0.90 (~6元人民币)") print(f"\n{'=' * 80}") print("检查完成！") print(f"{'=' * 80}")