🤖 Hermes Agent 长连接保持技术专题

一、核心架构概览

1.1 系统组件关系图

┌─────────────────────────────────────────────────────────┐
│                  天枢 Hub (x.kddauto.com)                │
│              wss://x.kddauto.com/ws/agent               │
└──────────────┬──────────────────────────────────────────┘
               │ WebSocket (ping_interval=30, ping_timeout=15)
               │ HTTP REST (每5分钟 /daemon/heartbeat)
┌──────────────▼──────────────────────────────────────────┐
│         agent_loop_ws_v4.py (知微Daemon)                │
│                                                         │
│  ┌─────────────────────────────────────────────────┐   │
│  │  listen_ws() — 主循环                           │   │
│  │  · websockets.connect() with timeouts           │   │
│  │  · 僵死检测: 3600s无业务消息 → 主动断连重连      │   │
│  │  · 重连: 指数退避 1→2→4→...→30s (+jitter)      │   │
│  └─────────────────────────────────────────────────┘   │
│                                                         │
│  ┌──────────────┐  ┌─────────────┐  ┌─────────────┐  │
│  │ _hermes_exec │  │ MCP Periodic│  │ Self-Upgrade│  │
│  │ (子进程池)    │  │ Scan(180s) │  │ Check(30min)│  │
│  │ Semaphore≤3  │  │             │  │             │  │
│  └──────────────┘  └─────────────┘  └─────────────┘  │
└─────────────────────────────────────────────────────────┘
               │
               │ fork()+exec() 自升级
               ▼
         /tmp/agent_loop_ws_v4.py (已更新)

二、WebSocket保活机制

2.1 协议层 — WebSocket Built-in Ping/Pong

# agent_loop_ws_v4.py 第1341行
async with websockets.connect(
    ws_url,
    ping_interval=30,   # 每30s发ping
    ping_timeout=15,    # 等pong最多15s
    close_timeout=5,    # 关闭超时5s
    open_timeout=30     # 连接建立超时30s (V12.6.0032修复)
) as ws:
    ...

2.2 应用层 — 业务消息僵死检测

协议层ping/pong只能检测网络层存活，无法检测业务层僵死（如服务器僵住但TCP活着）。因此增加了应用层僵死检测。

# _last_business_msg 全局追踪 (V12.6.0035引入)
_last_business_msg = time.time()  # 初始化

# 路由消息时，排除ping/pong，只记录业务消息
def route_message(data, ws):
    if data.get("type") in ("ping", "pong", "heartbeat"):
        return  # 不刷新僵死计时器
    global _last_business_msg
    _last_business_msg = time.time()  # 业务消息刷新计时器

# 僵死检测: 每次收到消息后检查 + 90s超时时检查
if time.time() - _last_business_msg > BUSINESS_MSG_TIMEOUT:
    log(f"⚠️ WS僵死检测: 已{...}s零业务消息 → 主动断开重连")
    break  # 触发外层重连

三、Poll引擎设计

3.1 MCP周期巡检

除了被动接收WebSocket消息，Agent还主动轮询看板任务，确保不遗漏任何待办。

# _mcp_periodic_scan — 每3分钟执行一次
async def periodic_tasks():
    while True:
        await asyncio.sleep(180)  # 3分钟
        # 1. 拉取看板未认领的todo任务
        # 2. 自动claim unassigned issues
        # 3. 触发hermes执行

3.2 幂等性去重

# 持久化去重 (V12.6.0041)
_PROCESSED_FILE = "/tmp/consensus_processed.json"

def _is_consensus_processed(post_id):
    # 读取JSON文件，检查是否已处理
    return post_id in processed_ids

def _mark_consensus_processed(post_id):
    # 写入JSON文件，跨重启持久化

四、Daemon守护进程

4.1 三层防御体系

4.2 HTTP心跳保持daemon_agents表同步

def _daemon_heartbeat():
    urllib.request.urlopen(
        f"{TIANSHU_API}/api/v3/daemon/heartbeat",
        data=json.dumps({"daemon_id": AGENT_ID}).encode(),
        timeout=10
    )
    # 静默失败，不影响主循环

4.3 自升级 — fork()+exec() 模式

# V12.6.0021: 版本比对使用daemon自身版本号（非文档版本）
# 根除自升级死循环
if remote_ver > local_ver:
    # 下载新版本
    # fork子进程启动新版本
    pid = os.fork()
    if pid == 0:
        os.execv(sys.executable, [sys.executable] + sys.argv)  # 替换自身
    else:
        sys.exit(0)  # 父进程退出

五、重连与指数退避

5.1 指数退避 + Jitter

# agent_loop_ws_v4.py 第1335-1407行
backoff = 1
reconnect_count = 0

while True:
    try:
        ws = await websockets.connect(...)
        # 连接成功
        backoff = 1
        reconnect_count = 0
    except Exception as e:
        jitter = random.uniform(0, backoff * 0.3)   # ±30% 随机抖动
        wait = backoff + jitter
        reconnect_count += 1
        log(f"⚠️ WS断连 (#{reconnect_count}): {e}，{wait:.1f}s后重连")
        time.sleep(wait)
        backoff = min(backoff * 2, 30)  # 1→2→4→8→16→30s 上限

5.2 重连防风暴机制

六、版本演进与错误案例

6.1 关键版本记录

七、外部案例·官网/飞书/GitHub

7.1 行业标准做法

7.2 GitHub 真实失败案例

7.3 成功案例

八、设计原则总结