200 lines
8.5 KiB
Python
200 lines
8.5 KiB
Python
#!/usr/bin/env python3
|
||
"""llm_client.py — 共享 LLM 客户端（直连上游 + gateway 兜底）
|
||
|
||
⚠️ 架构说明（2026-07-21 事故后重写）：
|
||
hermes gateway(:8643) 的 /v1/chat/completions 不是透传，而是完整 agent 运行时——
|
||
每个请求都会创建一个带工具（terminal/websearch/patch 等）的 agent 会话。
|
||
曾导致：一次重评请求螺旋 35 分钟、44 次 terminal 调用、输入累积到 153k token，
|
||
客户端 150s 超时后服务端继续空转，重试又叠加新会话，网关被自己人打满。
|
||
|
||
因此重评等批量分析调用【直连 OCG 上游】（裸 completion，无 agent），
|
||
hermes gateway 仅作兜底。监控仍可用 gateway_alive() 观察网关健康。
|
||
|
||
用法:
|
||
from llm_client import call_llm, REASSESS_MODEL, gateway_alive, ocg_alive
|
||
result = call_llm(prompt)
|
||
if result["ok"]:
|
||
text = result["content"]
|
||
"""
|
||
|
||
import json
|
||
import time
|
||
import urllib.request
|
||
import urllib.error
|
||
|
||
# ── 常量：所有重评调用统一使用 ──
|
||
# 2026-07-21 A/B 实测：flash 与 pro 在新 prompt 下质量差距微弱，flash 快 ~40%
|
||
REASSESS_MODEL = "deepseek-v4-flash"
|
||
# flash 对某些 prompt 会稳定返回空内容（688617 实测三连空），pro 能正常输出。
|
||
# 空输出时自动升级到 pro 重试一次。
|
||
FALLBACK_MODEL = "deepseek-v4-pro"
|
||
|
||
# 主通道：OCG 上游直连（与 hermes providers.ocg-key6 同源）
|
||
# key 运行时从 hermes config 读取（SSOT，不落盘到代码库）
|
||
OCG_URL = "https://opencode.ai/zen/go/v1/chat/completions"
|
||
_HERMES_CONFIG = "/home/hmo/.hermes/profiles/position-analyst/config.yaml"
|
||
|
||
|
||
def _load_ocg_key():
|
||
"""从 hermes config.yaml 读取 ocg-key6（单点数据源）。读不到则禁用直连通道。"""
|
||
import re
|
||
try:
|
||
with open(_HERMES_CONFIG, encoding="utf-8") as f:
|
||
text = f.read()
|
||
m = re.search(r'ocg-key6:\s*\n\s*api_key:\s*(\S+)', text)
|
||
if m:
|
||
return m.group(1)
|
||
except Exception as e:
|
||
print(f" [LLM] 无法从 hermes config 读取 ocg-key6: {e}", flush=True)
|
||
return None
|
||
|
||
|
||
_OCG_KEY = _load_ocg_key()
|
||
OCG_HEADERS = {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {_OCG_KEY}",
|
||
"User-Agent": "curl/8.5.0", # OCG UA 风控要求（与 hermes 配置一致）
|
||
} if _OCG_KEY else None
|
||
|
||
# 兜底通道：hermes gateway（注意：会走 agent 运行时，仅应急）
|
||
GATEWAY = "http://127.0.0.1:8643/v1/chat/completions"
|
||
GATEWAY_MODELS = "http://127.0.0.1:8643/v1/models"
|
||
GATEWAY_AUTH = "Bearer hermes123"
|
||
|
||
|
||
def _post(url, headers, payload, timeout):
|
||
"""单次 POST，返回 (ok, content_or_error)。"""
|
||
req = urllib.request.Request(url, data=payload, headers=headers)
|
||
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
||
resp = opener.open(req, timeout=timeout)
|
||
body = json.loads(resp.read().decode())
|
||
if "choices" not in body:
|
||
return False, f"API响应无choices字段: {str(body)[:200]}"
|
||
return True, body["choices"][0]["message"]["content"]
|
||
|
||
|
||
def ocg_alive(timeout=8):
|
||
"""OCG 上游可达性快检（极小请求）。key 缺失直接 False。"""
|
||
if not OCG_HEADERS:
|
||
return False
|
||
try:
|
||
payload = json.dumps({
|
||
"model": REASSESS_MODEL,
|
||
"messages": [{"role": "user", "content": "ping"}],
|
||
"max_tokens": 1,
|
||
}).encode()
|
||
_post(OCG_URL, OCG_HEADERS, payload, timeout)
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def gateway_alive(timeout=5):
|
||
"""快速预检 hermes gateway 是否存活。失败立刻返回 False，不阻塞。"""
|
||
try:
|
||
req = urllib.request.Request(GATEWAY_MODELS,
|
||
headers={"Authorization": GATEWAY_AUTH})
|
||
urllib.request.build_opener(urllib.request.ProxyHandler({})).open(req, timeout=timeout)
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def call_llm(prompt, model=None, max_tokens=4096, timeout=150,
|
||
retries=1, backoff=20, system=None):
|
||
"""调用 LLM：OCG 直连优先，hermes gateway 兜底。带重试和结构化日志。
|
||
|
||
Args:
|
||
prompt: 用户消息内容
|
||
model: 模型名（默认 REASSESS_MODEL）
|
||
max_tokens: 最大输出 token 数
|
||
timeout: 单次调用超时（秒）
|
||
retries: 每个通道的失败重试次数
|
||
backoff: 重试间隔（秒）
|
||
system: 可选 system message
|
||
|
||
Returns:
|
||
{ok, content, error, model, elapsed, attempts, channel}
|
||
永远不抛异常到调用方。
|
||
"""
|
||
model_name = model or REASSESS_MODEL
|
||
messages = []
|
||
if system:
|
||
messages.append({"role": "system", "content": system})
|
||
messages.append({"role": "user", "content": prompt})
|
||
payload = json.dumps({
|
||
"model": model_name,
|
||
"messages": messages,
|
||
"max_tokens": max_tokens,
|
||
}).encode()
|
||
|
||
channels = []
|
||
if OCG_HEADERS:
|
||
channels.append(("ocg", OCG_URL, OCG_HEADERS))
|
||
channels.append(("gateway", GATEWAY, {
|
||
"Content-Type": "application/json",
|
||
"Authorization": GATEWAY_AUTH,
|
||
}))
|
||
|
||
total_attempts = 0
|
||
last_err = "未知错误"
|
||
t_start = time.monotonic()
|
||
|
||
for ch_name, ch_url, ch_headers in channels:
|
||
for attempt in range(retries + 1):
|
||
total_attempts += 1
|
||
t0 = time.monotonic()
|
||
try:
|
||
ok, result = _post(ch_url, ch_headers, payload, timeout)
|
||
elapsed = time.monotonic() - t0
|
||
if ok:
|
||
# ── 空输出升级：flash 对部分 prompt 稳定返回空（688617 实测），
|
||
# 换 FALLBACK_MODEL(pro) 重试一次 ──
|
||
if not result.strip() and model_name != FALLBACK_MODEL and OCG_HEADERS:
|
||
print(f" [LLM] {ch_name} {model_name} 空输出({elapsed:.1f}s)，"
|
||
f"升级 {FALLBACK_MODEL} 重试...", flush=True)
|
||
esc_payload = json.dumps({
|
||
"model": FALLBACK_MODEL, "messages": messages,
|
||
"max_tokens": max_tokens,
|
||
}).encode()
|
||
try:
|
||
ok2, result2 = _post(OCG_URL, OCG_HEADERS, esc_payload, timeout)
|
||
total_attempts += 1
|
||
if ok2 and result2.strip():
|
||
print(f" [LLM] 升级 {FALLBACK_MODEL} 成功, 输出{len(result2)}字", flush=True)
|
||
return {
|
||
"ok": True, "content": result2, "error": None,
|
||
"model": FALLBACK_MODEL,
|
||
"elapsed": time.monotonic() - t_start,
|
||
"attempts": total_attempts, "channel": ch_name + "+esc",
|
||
}
|
||
print(f" [LLM] 升级 {FALLBACK_MODEL} 仍空/失败", flush=True)
|
||
except Exception as e2:
|
||
print(f" [LLM] 升级 {FALLBACK_MODEL} 异常: {str(e2)[:100]}", flush=True)
|
||
print(f" [LLM] {ch_name} 尝试{attempt+1}/{retries+1} 成功, "
|
||
f"{elapsed:.1f}s, 输出{len(result)}字, model={model_name}", flush=True)
|
||
return {
|
||
"ok": True, "content": result, "error": None,
|
||
"model": model_name, "elapsed": time.monotonic() - t_start,
|
||
"attempts": total_attempts, "channel": ch_name,
|
||
}
|
||
last_err = result
|
||
print(f" [LLM] {ch_name} 尝试{attempt+1}/{retries+1} 失败({elapsed:.1f}s): {result[:120]}", flush=True)
|
||
except Exception as e:
|
||
elapsed = time.monotonic() - t0
|
||
last_err = str(e)
|
||
print(f" [LLM] {ch_name} 尝试{attempt+1}/{retries+1} 异常({elapsed:.1f}s): {last_err[:120]}", flush=True)
|
||
if attempt < retries:
|
||
print(f" [LLM] 等待{backoff}s后重试...", flush=True)
|
||
time.sleep(backoff)
|
||
# 当前通道重试耗尽 → 切下一通道
|
||
if (ch_name, ch_url, ch_headers) != channels[-1]:
|
||
print(f" [LLM] {ch_name} 通道不可用，切换兜底通道...", flush=True)
|
||
|
||
return {
|
||
"ok": False, "content": "",
|
||
"error": f"双通道均失败(共{total_attempts}次): {last_err[:200]}",
|
||
"model": model_name, "elapsed": time.monotonic() - t_start,
|
||
"attempts": total_attempts, "channel": None,
|
||
}
|