205 lines
6.7 KiB
Python
205 lines
6.7 KiB
Python
"""Client tới Hermes Agent — kênh Zalo riêng (không phải support KB).
|
|
|
|
Backend mặc định: Hermes API Server (OpenAI-compatible) trên gateway :8642.
|
|
Đọc env lúc gọi (không snapshot lúc import) để load_dotenv() luôn có hiệu lực.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
from collections import defaultdict, deque
|
|
from typing import Any, Deque, Dict, List
|
|
|
|
import requests
|
|
|
|
MISS = (
|
|
"Hermes chưa trả lời được lúc này. Bạn thử lại sau hoặc diễn đạt lại câu hỏi nhé."
|
|
)
|
|
|
|
# chat_id -> deque of {role, content} — maxlen set on first use / refresh
|
|
_HISTORIES: Dict[str, Deque[dict[str, str]]] = {}
|
|
|
|
|
|
def _cfg() -> dict[str, Any]:
|
|
turns = int(os.environ.get("HERMES_HISTORY_TURNS", "12"))
|
|
return {
|
|
"base": os.environ.get("HERMES_API_BASE", "http://127.0.0.1:8642/v1").rstrip("/"),
|
|
"key": (os.environ.get("HERMES_API_KEY") or "").strip(),
|
|
"model": os.environ.get("HERMES_MODEL", "hermes-agent"),
|
|
"timeout": float(os.environ.get("HERMES_TIMEOUT", "180")),
|
|
"turns": turns,
|
|
"backend": (os.environ.get("HERMES_BACKEND") or "api").strip().lower(),
|
|
"system": (
|
|
os.environ.get("HERMES_SYSTEM_PROMPT")
|
|
or (
|
|
"Bạn đang chat với người dùng qua Zalo Bot. "
|
|
"Trả lời ngắn gọn, tiếng Việt trừ khi họ hỏi tiếng khác. "
|
|
"Không dùng markdown nặng; ưu tiên plain text dễ đọc trên Zalo."
|
|
)
|
|
).strip(),
|
|
}
|
|
|
|
|
|
# Compat exports used by bot_app logging
|
|
def _refresh_exports() -> None:
|
|
global HERMES_API_BASE, HERMES_BACKEND
|
|
c = _cfg()
|
|
HERMES_API_BASE = c["base"]
|
|
HERMES_BACKEND = c["backend"]
|
|
|
|
|
|
_refresh_exports()
|
|
|
|
|
|
def _auth_headers(key: str) -> dict[str, str]:
|
|
h = {"Content-Type": "application/json", "User-Agent": "zalo-hermes-bot/0.1"}
|
|
if key:
|
|
h["Authorization"] = f"Bearer {key}"
|
|
return h
|
|
|
|
|
|
def _history_deque(chat_id: str, turns: int) -> Deque[dict[str, str]]:
|
|
key = str(chat_id)
|
|
maxlen = max(2, turns * 2)
|
|
d = _HISTORIES.get(key)
|
|
if d is None or d.maxlen != maxlen:
|
|
old = list(d) if d is not None else []
|
|
d = deque(old, maxlen=maxlen)
|
|
_HISTORIES[key] = d
|
|
return d
|
|
|
|
|
|
def hermes_health() -> dict[str, Any]:
|
|
c = _cfg()
|
|
_refresh_exports()
|
|
if c["backend"] == "cli":
|
|
try:
|
|
p = subprocess.run(
|
|
["hermes", "--version"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
return {
|
|
"ok": p.returncode == 0,
|
|
"backend": "cli",
|
|
"body": (p.stdout or p.stderr or "")[:200],
|
|
}
|
|
except Exception as e:
|
|
return {"ok": False, "backend": "cli", "error": str(e)}
|
|
|
|
candidates = [
|
|
f"{c['base']}/models",
|
|
c["base"].replace("/v1", "") + "/health",
|
|
"http://127.0.0.1:8642/health",
|
|
]
|
|
last: dict[str, Any] = {"ok": False, "backend": "api"}
|
|
for url in candidates:
|
|
try:
|
|
r = requests.get(url, timeout=5, headers=_auth_headers(c["key"]))
|
|
ctype = r.headers.get("content-type", "")
|
|
body: Any = r.json() if "json" in ctype else r.text[:200]
|
|
last = {
|
|
"ok": r.status_code < 400,
|
|
"backend": "api",
|
|
"url": url,
|
|
"status_code": r.status_code,
|
|
"body": body,
|
|
"api_key_len": len(c["key"]),
|
|
}
|
|
if last["ok"]:
|
|
return last
|
|
except Exception as e:
|
|
last = {"ok": False, "backend": "api", "url": url, "error": str(e)}
|
|
return last
|
|
|
|
|
|
def _history_for(chat_id: str, turns: int) -> List[dict[str, str]]:
|
|
return list(_history_deque(chat_id, turns))
|
|
|
|
|
|
def _remember(chat_id: str, role: str, content: str, turns: int) -> None:
|
|
_history_deque(chat_id, turns).append({"role": role, "content": content})
|
|
|
|
|
|
def clear_history(chat_id: str) -> None:
|
|
_HISTORIES.pop(str(chat_id), None)
|
|
|
|
|
|
def ask_hermes_api(chat_id: str, message: str) -> dict[str, Any]:
|
|
c = _cfg()
|
|
msg = (message or "").strip()
|
|
if not msg:
|
|
return {"answer": MISS, "meta": "empty"}
|
|
if not c["key"]:
|
|
print("HERMES CLIENT: HERMES_API_KEY empty — load_dotenv before import, or export env")
|
|
return {"answer": MISS, "meta": "missing api key"}
|
|
|
|
messages: list[dict[str, str]] = []
|
|
if c["system"]:
|
|
messages.append({"role": "system", "content": c["system"]})
|
|
messages.extend(_history_for(chat_id, c["turns"]))
|
|
messages.append({"role": "user", "content": msg})
|
|
|
|
url = f"{c['base']}/chat/completions"
|
|
try:
|
|
r = requests.post(
|
|
url,
|
|
json={
|
|
"model": c["model"],
|
|
"messages": messages,
|
|
"stream": False,
|
|
},
|
|
timeout=c["timeout"],
|
|
headers=_auth_headers(c["key"]),
|
|
)
|
|
if r.status_code == 401:
|
|
print(
|
|
"HERMES CLIENT: unauthorized — "
|
|
f"api_key_len={len(c['key'])} (must match API_SERVER_KEY)"
|
|
)
|
|
return {"answer": MISS, "meta": "unauthorized"}
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
choices = data.get("choices") or []
|
|
content = ""
|
|
if choices:
|
|
content = ((choices[0].get("message") or {}).get("content") or "").strip()
|
|
if not content:
|
|
return {"answer": MISS, "meta": "empty response"}
|
|
_remember(chat_id, "user", msg, c["turns"])
|
|
_remember(chat_id, "assistant", content, c["turns"])
|
|
return {"answer": content, "meta": "api"}
|
|
except Exception as e:
|
|
print(f"HERMES CLIENT ERROR: {e}")
|
|
return {"answer": MISS, "meta": f"error: {e}"}
|
|
|
|
|
|
def ask_hermes_cli(message: str) -> dict[str, Any]:
|
|
c = _cfg()
|
|
msg = (message or "").strip()
|
|
if not msg:
|
|
return {"answer": MISS, "meta": "empty"}
|
|
try:
|
|
p = subprocess.run(
|
|
["hermes", "chat", "-q", msg],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=c["timeout"],
|
|
)
|
|
out = (p.stdout or "").strip() or (p.stderr or "").strip()
|
|
if p.returncode != 0 or not out:
|
|
print(f"HERMES CLI fail rc={p.returncode}: {out[:300]}")
|
|
return {"answer": MISS, "meta": f"cli rc={p.returncode}"}
|
|
return {"answer": out, "meta": "cli"}
|
|
except Exception as e:
|
|
print(f"HERMES CLI ERROR: {e}")
|
|
return {"answer": MISS, "meta": f"cli error: {e}"}
|
|
|
|
|
|
def ask_hermes(chat_id: str, message: str) -> dict[str, Any]:
|
|
c = _cfg()
|
|
if c["backend"] == "cli":
|
|
return ask_hermes_cli(message)
|
|
return ask_hermes_api(chat_id, message)
|