337 lines
15 KiB
Python
337 lines
15 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"
|
||
# 本地 OCG 路由代理（2026-08-13 老莫修复bug后启用）：池化多key自动选最空闲+故障冷却
|
||
# 与 hermes ocg-router provider 同源，MoFin 重评主通道
|
||
OCG_ROUTER = "http://127.0.0.1:19878/v1/chat/completions"
|
||
OCG_ROUTER_AUTH = "Bearer ocg-router-local"
|
||
_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
|
||
|
||
|
||
def _load_key_pool():
|
||
"""从 hermes config.yaml 解析全部 OCG key，按 AgentsMeeting /api/keys 的
|
||
实时状态排序（可用优先、用量低优先）。失败回退到单 key6。
|
||
返回 [(key_id, api_key), ...]，按优先级排序。"""
|
||
import re
|
||
fallback = [("key6", _load_ocg_key())] if _load_ocg_key() else []
|
||
try:
|
||
with open(_HERMES_CONFIG, encoding="utf-8") as f:
|
||
text = f.read()
|
||
# 解析 providers 下的 ocg 块: name + api_key
|
||
blocks = re.findall(r'^\s{2}(ocg[\w\-]*):\s*\n\s*api_key:\s*(\S+)', text, re.M)
|
||
if not blocks:
|
||
return fallback
|
||
# provider名 → key_id 映射（ocg-keyN→keyN, ocg-3→key3, ocg-hy3→key1, ocg-new→key2）
|
||
name2id = {"ocg-key6": "key6", "ocg-key5": "key5", "ocg-key4": "key4",
|
||
"ocg-3": "key3", "ocg-hy3": "key1", "ocg-new": "key2"}
|
||
kv = {}
|
||
for name, key in blocks:
|
||
kid = name2id.get(name)
|
||
if kid and kid != "key7":
|
||
kv[kid] = key
|
||
if not kv:
|
||
return fallback
|
||
# 查询 AgentsMeeting /api/keys 实时状态
|
||
import json as _json
|
||
status = {}
|
||
try:
|
||
req = urllib.request.Request("http://127.0.0.1:5803/api/keys")
|
||
with urllib.request.build_opener(urllib.request.ProxyHandler({})).open(req, timeout=5) as r:
|
||
data = _json.loads(r.read().decode())
|
||
for k in data.get("keys", []):
|
||
m_ok = k.get("monthly", {}).get("status") == "ok"
|
||
r_ok = k.get("rolling", {}).get("status") == "ok"
|
||
usage = k.get("monthly", {}).get("usage_percent", 99)
|
||
status[k["key_id"]] = (m_ok and r_ok, usage)
|
||
except Exception:
|
||
# API 不可达时默认全可用、key6 优先
|
||
order = ["key6", "key5", "key4", "key2", "key1", "key3"]
|
||
return [(k, kv[k]) for k in order if k in kv]
|
||
# 可用且用量低优先；key6 在用量相同时优先（沿用现有配置）
|
||
scored = []
|
||
for kid, key in kv.items():
|
||
ok, usage = status.get(kid, (True, 50))
|
||
if ok:
|
||
scored.append((usage - (2 if kid == "key6" else 0), kid, key))
|
||
scored.sort()
|
||
pool = [(kid, key) for _, kid, key in scored]
|
||
return pool or fallback
|
||
except Exception as e:
|
||
print(f" [LLM] key池加载失败，回退单key: {e}", flush=True)
|
||
return fallback
|
||
|
||
|
||
_KEY_POOL = None
|
||
_KEY_POOL_TS = 0
|
||
|
||
|
||
def _get_key_pool():
|
||
"""key 池缓存 5 分钟。LLM_KEY_OFFSET 环境变量：并发 worker 起始 key 错开，
|
||
避免 N 个进程同时压同一个首选 key（2026-07-24 并发分片配套）。"""
|
||
global _KEY_POOL, _KEY_POOL_TS
|
||
import time as _t, os as _os
|
||
if not _KEY_POOL or (_t.time() - _KEY_POOL_TS) > 300:
|
||
_KEY_POOL = _load_key_pool()
|
||
_KEY_POOL_TS = _t.time()
|
||
if _KEY_POOL:
|
||
_off = int(_os.environ.get("LLM_KEY_OFFSET", "0") or 0)
|
||
if _off and len(_KEY_POOL) > 1:
|
||
_off = _off % len(_KEY_POOL)
|
||
_KEY_POOL = _KEY_POOL[_off:] + _KEY_POOL[:_off]
|
||
print(f" [LLM] key池: {[k for k, _ in _KEY_POOL]}" + (f" (offset={_off})" if _off else ""), flush=True)
|
||
return _KEY_POOL
|
||
|
||
|
||
_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"
|
||
|
||
# ── SenseNova 备用通道（2026-07-24 老爸：并发要快）──
|
||
# token.sensenova.cn 托管同款 deepseek-v4-flash，实测 3.2s vs OCG 45s（开盘高峰）。
|
||
# 公测免费，与 OCG 不同上游、不共享拥塞。不支持 deepseek-v4-pro 升级（404）。
|
||
# key 从 hermes config 的 sense-nova provider 读取（SSOT，不落盘）。
|
||
SN_URL = "https://token.sensenova.cn/v1/chat/completions"
|
||
|
||
|
||
def _load_sn_key():
|
||
"""从 hermes config.yaml 读 sense-nova 的 api_key。读不到则禁用该通道。"""
|
||
import re
|
||
try:
|
||
with open(_HERMES_CONFIG, encoding="utf-8") as f:
|
||
text = f.read()
|
||
m = re.search(r'sense-nova:\s*\n\s*api_key:\s*(\S+)', text)
|
||
if m:
|
||
return m.group(1)
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
|
||
_SN_KEY = _load_sn_key()
|
||
SN_HEADERS = {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {_SN_KEY}",
|
||
"User-Agent": "curl/8.5.0",
|
||
} if _SN_KEY else None
|
||
|
||
|
||
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 可达性快检（极小请求）。优先本地路由代理，回退直连。"""
|
||
payload = json.dumps({
|
||
"model": REASSESS_MODEL,
|
||
"messages": [{"role": "user", "content": "ping"}],
|
||
"max_tokens": 1,
|
||
}).encode()
|
||
# 本地路由代理
|
||
try:
|
||
_post(OCG_ROUTER, {"Content-Type": "application/json",
|
||
"Authorization": OCG_ROUTER_AUTH,
|
||
"User-Agent": "curl/8.5.0"}, payload, timeout)
|
||
return True
|
||
except Exception:
|
||
pass
|
||
# 直连兜底
|
||
if not OCG_HEADERS:
|
||
return False
|
||
try:
|
||
_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=None, timeout=180,
|
||
retries=1, backoff=20, system=None, concurrent=False):
|
||
# 2026-08-17 OCG-Router文档对齐：
|
||
# max_tokens 默认 None=不写入 payload（deepseek-v4-flash 是推理模型，指定 max_tokens
|
||
# 会被 reasoning tokens 烧光 → finish=length 空输出。不指定用模型默认 1M 上下文）。
|
||
# timeout 150→180（文档普通请求建议）。
|
||
# retries/backoff 保持 1/20：6 通道链下多重试=疯狂重试（文档禁止项），20s 退避防限流。
|
||
"""调用 LLM：OCG 直连优先，hermes gateway 兜底。带重试和结构化日志。
|
||
|
||
Args:
|
||
prompt: 用户消息内容
|
||
model: 模型名（默认 REASSESS_MODEL）
|
||
max_tokens: 最大输出 token 数
|
||
timeout: 单次调用超时（秒）
|
||
retries: 每个通道的失败重试次数
|
||
backoff: 重试间隔（秒）
|
||
system: 可选 system message
|
||
concurrent: 并发模式（2026-08-13）：True 时 OCG router 用 round-robin 锁 key 轮询，
|
||
避免并发压同一 key（header X-OCG-Mode: concurrent）
|
||
|
||
Returns:
|
||
{ok, content, error, model, elapsed, attempts, channel}
|
||
永远不抛异常到调用方。
|
||
"""
|
||
model_name = model or REASSESS_MODEL
|
||
# 仅显式传入时才钳制下限（防过小短路，2026-07-22 实测 max_tokens=64 必空）；
|
||
# None = 不指定（遵循文档：推理模型不设上限）
|
||
if max_tokens is not None and max_tokens < 512:
|
||
max_tokens = 512
|
||
messages = []
|
||
if system:
|
||
messages.append({"role": "system", "content": system})
|
||
messages.append({"role": "user", "content": prompt})
|
||
_req = {"model": model_name, "messages": messages}
|
||
if max_tokens is not None:
|
||
_req["max_tokens"] = max_tokens
|
||
payload = json.dumps(_req).encode()
|
||
|
||
# ── 构建通道列表：SenseNova(快,同款model) → OCG路由代理(本地,自动选key) → key池直连 → gateway 兜底 ──
|
||
# 2026-08-13 老莫修复 ocg_router bug 后，路由代理作为 OCG 主通道（自动选最空闲key+故障冷却）。
|
||
# key池直连保留为 fallback（router 不可达时回退）。
|
||
channels = []
|
||
# OCG 路由代理第一（2026-08-13 老莫：重评配置到 OCG 路由代理，支持pro升级+多key池）
|
||
_ocg_headers = {
|
||
"Content-Type": "application/json",
|
||
"Authorization": OCG_ROUTER_AUTH,
|
||
"User-Agent": "curl/8.5.0",
|
||
}
|
||
# 2026-08-13 并发模式：concurrent=True 时加 header，ocg_router round-robin 锁 key 轮询
|
||
if concurrent:
|
||
_ocg_headers["X-OCG-Mode"] = "concurrent"
|
||
channels.append(("ocg_router", OCG_ROUTER, _ocg_headers))
|
||
# SenseNova 快通道（并发应急，不支持 pro 升级）
|
||
if SN_HEADERS:
|
||
channels.append(("sensenova", SN_URL, SN_HEADERS))
|
||
pool = _get_key_pool()
|
||
for kid, key in pool:
|
||
if key:
|
||
channels.append((f"ocg:{kid}", OCG_URL, {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {key}",
|
||
"User-Agent": "curl/8.5.0",
|
||
}))
|
||
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:
|
||
# ── 空输出视为失败（2026-08-14 老莫：不换pro，OCG router 内部空输出自动换 key）──
|
||
# ocg_router 已内置空输出检测（_forward_request 检测 content 为空 → 自动换下一个 key）
|
||
# llm_client 不再升级 pro，空输出直接失败（依赖 ocg_router 换 key）
|
||
if not result.strip():
|
||
# SenseNova 不托管 deepseek-v4-pro（404），空输出直接切下一通道
|
||
if ch_name == "sensenova":
|
||
print(f" [LLM] {ch_name} 空输出视为失败，切换下一通道...", flush=True)
|
||
last_err = "empty content"
|
||
continue
|
||
# ocg_router 通道：空输出视为失败（ocg_router 内部已自动换 key 重试，到这说明所有 key 都空）
|
||
last_err = "empty content"
|
||
print(f" [LLM] {ch_name} 空输出视为失败，切换下一通道...", flush=True)
|
||
continue
|
||
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,
|
||
}
|