test_hermes_api.py
5.25 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
#!/usr/bin/env python3
"""
Hermes API 调用测试 — 验证 daemon 到 hermes OpenAI-compatible API 的 HTTP 链路。
在 Ubuntu proot 或 Termux 环境内执行。
"""
import subprocess, json, sys, os, time, urllib.request, urllib.error
CONFIG_PATH = "config.json"
REPORT = {"test": "hermes_api", "steps": {}, "overall": None}
def log_step(name, ok, detail=""):
status = "[OK]" if ok else "[FAIL]"
print(f" {status} {name}")
if detail:
print(f" {detail}")
REPORT["steps"][name] = {"ok": ok, "detail": detail}
return ok
def call_hermes_api(api_url, message, session_id, timeout=60):
"""Call hermes OpenAI-compatible API and parse SSE response."""
payload = json.dumps({
"model": "hermes-agent",
"messages": [
{"role": "user", "content": message}
],
"stream": True,
"session_id": session_id
}).encode("utf-8")
req = urllib.request.Request(
api_url,
data=payload,
headers={"Content-Type": "application/json", "Accept": "text/event-stream"},
method="POST"
)
try:
full_reply = []
with urllib.request.urlopen(req, timeout=timeout) as resp:
buffer = ""
for chunk in resp:
chunk_str = chunk.decode("utf-8", errors="replace")
buffer += chunk_str
while "\n\n" in buffer:
block, buffer = buffer.split("\n\n", 1)
data_str = None
for line in block.strip().split("\n"):
if line.startswith("data:"):
data_str = line[5:].strip()
if data_str:
if data_str == "[DONE]":
break
try:
obj = json.loads(data_str)
delta = obj.get("choices", [{}])[0].get("delta", {})
content = delta.get("content", "")
if content:
full_reply.append(content)
except json.JSONDecodeError:
pass
return True, "".join(full_reply)
except urllib.error.URLError as e:
return False, f"URLError: {e}"
except Exception as e:
return False, str(e)
def main():
print("=" * 60)
print("Hermes API 调用测试 (OpenAI-compatible)")
print("=" * 60)
config = {}
if os.path.exists(CONFIG_PATH):
with open(CONFIG_PATH) as f:
config = json.load(f)
all_ok = True
hermes_cfg = config.get("hermes", {})
api_url = hermes_cfg.get("api_url", "http://localhost:8642/v1/chat/completions")
timeout = hermes_cfg.get("timeout_sec", 60)
# Step 1: Check hermes server is reachable
print("\n[Step 1] 检查 hermes 服务可达性")
try:
health_url = api_url.replace("/v1/chat/completions", "/health")
req = urllib.request.Request(health_url, method="GET")
with urllib.request.urlopen(req, timeout=5) as resp:
health_ok = resp.status == 200
body = resp.read().decode("utf-8", errors="replace")[:100]
except Exception as e:
health_ok = False
body = str(e)
all_ok &= log_step("hermes_reachable", health_ok, body)
if not health_ok:
print("\n [提示] hermes 服务未启动或端口不对。请确认：")
print(f" 1. hermes gateway 已启动 (python3 -m gateway.run)")
print(f" 2. api_server 已启用 (config.yaml platforms.api_server.enabled=true)")
print(f" 3. 监听端口与 {api_url} 一致")
save_report()
return 1
# Step 2: Send a test message
print("\n[Step 2] 发送测试消息到 hermes")
test_msg = "你好，这是一个来自 daemon 的测试消息。请回复一句简短的话。"
session_id = f"test-{int(time.time())}"
print(f" sessionId: {session_id}")
print(f" message: {test_msg[:60]}")
ok, result = call_hermes_api(api_url, test_msg, session_id, timeout)
detail = f"回复: {result[:80]}" if result else f"错误: {result}"
all_ok &= log_step("hermes_chat", ok, detail)
# Step 3: Context persistence test (same session)
if ok and result:
print("\n[Step 3] 上下文持久化测试（同一 session）")
followup = "刚才你说了什么？请重复一遍。"
ok2, result2 = call_hermes_api(api_url, followup, session_id, timeout)
has_context = ok2 and result2 and (result[:30] in result2 or "刚才" in result2 or "说了" in result2)
all_ok &= log_step("context_persist", has_context,
f"回复: {result2[:80] if result2 else 'None'}" if ok2 else f"失败: {result2}")
# Summary
REPORT["overall"] = "PASS" if all_ok else "FAIL"
REPORT["timestamp"] = subprocess.check_output(["date", "-Iseconds"]).decode().strip()
save_report()
print("\n" + "=" * 60)
print(f"结果: {REPORT['overall']}")
print("=" * 60)
print("\n详细报告: test_report_hermes_api.json")
return 0 if all_ok else 1
def save_report():
with open("test_report_hermes_api.json", "w") as f:
json.dump(REPORT, f, indent=2, ensure_ascii=False)
if __name__ == "__main__":
sys.exit(main())