7.1. 通过 Gateway 可将 Hermes Agent 暴露为标准化 API 服务

Hermes Agent 系统设计与工程落地 /

你需要什么

环境：一台运行 Docker 的 Linux 服务器（最低 2 核 4 GB），或本地开发机（macOS/Windows/Linux 均可）
工具：Docker（20.10+）、Docker Compose（v2）、curl、jq
账号：无需第三方账号，所有组件在本机部署
基础知识：熟悉 HTTP、WebSocket、基本的 Linux 命令行操作
预计时间：60 分钟（首次搭建），了解原理后可在 15 分钟内完成二次部署

最终成果

你将得到一个生产级 API 网关，它把 Hermes Agent 的能力暴露为一组标准化的 REST/WebSocket 端点。外部调用方（如 React 前端、微服务、定时任务）只需要携带合法的 JWT 令牌，就能向 Agent 发送自然语言指令，并获得实时流式响应（SSE / WebSocket）。整个网关层内置了高可用负载均衡、健康检查、速率限制和优雅重启能力——这正是上一章所承诺的：为你的 Agent 穿上坚硬的 HTTP 外壳，让智能流淌到任何需要它的缝隙里。

为什么做这个？
Hermes 开箱提供了 hermes 终端 UI 和一个内置 Gateway，能对接 Telegram、Discord 等消息平台。但绝大多数业务系统（订单、客服、运维面板）需要的是标准的 JSON over HTTP API，而不是依赖聊天应用。本章教你如何利用 Gateway 的架构思想，在 30 分钟内自建一个符合企业安全规范的 API 出口，把“会学习的 Agent”变成“可集成的微服务”。

架构全景

在开始动手之前，先俯瞰整个数据流（图 1）。我们将在 Docker 网络中部署三组进程：

Hermes Gateway：官方提供的 Gateway 进程，负责与 LLM 后端交互、管理对话记忆、触发技能等核心逻辑。它只通过 WebSocket 对外通信，不直接暴露 HTTP。
API 适配层（FastAPI）：我们编写的轻量级服务，一侧以 WebSocket 客户端身份连接 Gateway，另一侧通过 HTTP/WebSocket 对外提供标准化接口。这一层负责 JWT 验证、限流、请求格式校验。
Nginx 反向代理：承担 TLS 终结、负载均衡（同多个 API 适配层实例）、健康检查、路由分发。

客户端 ── HTTPS ── Nginx ── HTTP/WS ── FastAPI 适配层 (多实例)
                                          │ (内部 WebSocket)
                                          └──────── Hermes Gateway

这种分层设计遵循 解耦调用方式与执行逻辑 的核心理念：Gateway 专注于智能执行，适配层专注于 API 契约与安全，彼此通过标准 WebSocket 协议松耦合。无论你想添加新的鉴权策略、升级 Gateway 版本，还是把适配层替换成 gRPC 服务，都不会影响另一侧。

步骤说明

步骤 1：启动 Hermes Gateway 后台服务

首先，确保你的机器上已经安装了 Docker。如果还没有，可参考 Docker 官方文档安装。我们使用一个预先配置好的 Docker Compose 文件来启动 Gateway。

创建项目目录并进入：

mkdir hermes-api-gateway && cd hermes-api-gateway

新建 docker-compose.yml，写入以下内容：

version: "3.9"
services:
  hermes-gateway:
    image: nousresearch/hermes-agent:latest  # 截至 2026 年 6 月的社区建议镜像
    container_name: hermes-core
    restart: unless-stopped
    environment:
      - HERMES_LLM_BACKEND=openai        # 可替换为 ollama / deepseek 等
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - HERMES_WS_HOST=0.0.0.0
      - HERMES_WS_PORT=8765               # WebSocket 监听端口
    ports:
      - "8765:8765"
    volumes:
      - ./hermes_data:/app/data           # 持久化记忆和技能
    networks:
      - agent-net

networks:
  agent-net:
    driver: bridge

在同级目录下创建 .env 文件存放敏感信息：

OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxxxx

预期结果：运行 docker compose up -d，Gateway 容器在几秒内启动，日志中应出现 WebSocket server listening on 0.0.0.0:8765 字样。用 docker logs hermes-core -f 查看输出确认。

踩坑注意：若使用非 OpenAI 兼容的 LLM 后端（如本地 Ollama），需要额外配置 HERMES_LLM_BASE_URL 环境变量并确保网络可达。Gateway 与服务在同一 Docker 网络时，可用 host.docker.internal 访问宿主机端口。

步骤 2：编写 API 适配层（FastAPI）

Gateway 的 WebSocket 协议是一个基于 JSON 帧的 RPC 风格通信，每条消息包含 { "type": "command" | "response" | "error", "payload": {...} } 等字段。我们现在创建一个 FastAPI 应用，它既握持到 Gateway 的 WebSocket 连接，又暴露 REST 和 WebSocket 端点给外部。

在项目根目录新建 api_server.py：

import json
import asyncio
import time
from contextlib import asynccontextmanager
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
import websockets

# ---------- 与 Hermes Gateway 的连接管理 ----------
GATEWAY_WS_URL = "ws://hermes-core:8765"  # Docker 内部网络地址
gateway_ws = None
ws_lock = asyncio.Lock()

async def reconnect_gateway():
    global gateway_ws
    while True:
        try:
            gateway_ws = await websockets.connect(GATEWAY_WS_URL)
            print("Connected to Hermes Gateway")
            return
        except Exception as e:
            print(f"Gateway connection failed: {e}, retrying in 5s...")
            await asyncio.sleep(5)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时连接 Gateway
    await reconnect_gateway()
    # 启动后台心跳保持连接
    asyncio.create_task(heartbeat())
    yield
    if gateway_ws:
        await gateway_ws.close()

async def heartbeat():
    while True:
        if gateway_ws and gateway_ws.open:
            try:
                # 发送心跳帧，防止中间代理断开空闲连接
                await gateway_ws.send(json.dumps({"type": "ping"}))
                await asyncio.sleep(30)
            except Exception:
                await reconnect_gateway()
        else:
            await reconnect_gateway()

app = FastAPI(title="Hermes API Gateway", version="1.0", lifespan=lifespan)

# ---------- JWT 验证依赖 ----------
security = HTTPBearer()

# 此处为了示例简化，直接硬编码一个 HS256 密钥和 token；生产环境应使用密钥管理系统
SECRET_KEY = "dev-secret-change-me"
ALGORITHM = "HS256"
import jwt

async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")

# ---------- 请求模型 ----------
class AgentRequest(BaseModel):
    text: str
    conversation_id: str | None = None

# ---------- REST 端点：发送指令并阻塞等待结果 ----------
@app.post("/agent/ask")
async def ask_agent(req: AgentRequest, user=Depends(verify_token)):
    # 构造发送给 Gateway 的命令
    command = {
        "type": "command",
        "payload": {
            "action": "chat",
            "text": req.text,
            "conversation_id": req.conversation_id or "default"
        }
    }
    async with ws_lock:               # 确保对 Gateway 连接的独占访问
        try:
            await gateway_ws.send(json.dumps(command))
            response_raw = await asyncio.wait_for(gateway_ws.recv(), timeout=60)
            response = json.loads(response_raw)
            if response.get("type") == "response":
                return {
                    "conversation_id": response["payload"].get("conversation_id"),
                    "reply": response["payload"].get("text")
                }
            else:
                raise HTTPException(status_code=502, detail=response.get("payload", {}).get("error", "Gateway error"))
        except asyncio.TimeoutError:
            raise HTTPException(status_code=504, detail="Agent timeout")
        except websockets.ConnectionClosed:
            await reconnect_gateway()
            raise HTTPException(status_code=503, detail="Gateway connection lost, retry")

文件里几个关键点：

lifespan 中启动时连接 Gateway，失败则无限重连，保证启动顺序无关。
heartbeat 每 30 秒发送一次 ping，避免 WebSocket 空闲断开。
/agent/ask 端点接受 JSON 请求，内部串行化使用 Gateway 连接（ws_lock），并设置 60 秒超时。
JWT 验证使用 python-jose（jwt 包）的内置方法，示例密钥仅用于开发。

预期结果：运行 pip install fastapi uvicorn websockets python-jose 后，用 uvicorn api_server:app --host 0.0.0.0 --port 8000 启动。此时 /agent/ask 尚不能正确响应（因为 Gateway 需要实际的 LLM 后端），但你可以在控制台看到 “Connected to Hermes Gateway” 日志。若 Gateway 未启动，则会看到重试信息。

步骤 3：添加速率限制防止滥用

在 FastAPI 中集成限流非常容易。我们使用 slowapi 包，它基于 limits 库，提供内存或 Redis 后端。

安装依赖：pip install slowapi

修改 api_server.py，在文件顶部添加：

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

然后在 /agent/ask 路由上添加装饰器：

@app.post("/agent/ask")
@limiter.limit("5/minute")  # 每个 IP 每分钟最多 5 次，防止暴力调用
async def ask_agent(...):
    ...

预期结果：重启 API 服务后，用 curl 快速连续发送 6 次请求，第 6 次将返回 429 Too Many Requests 状态码，并携带 Retry-After 头。

生产建议：多实例部署时，需要将限流状态存储在共享的 Redis 中，而非进程内存。slowapi 支持 Limiter(storage_uri="redis://...")，只需做简单修改。

步骤 4：实现 WebSocket 流式输出，提升前端体验

许多 Agent 交互（如生成报告、遍历步骤）需要较长时间，客户端如果等待一个完整的 REST 响应，会长时间白屏。流式输出让 Agent 每生成一个 token 或每输出一行日志，就立刻推送给用户。我们借助 FastAPI 的 WebSocket 端点实现。

在 api_server.py 中新增端点：

from fastapi import WebSocket

@app.websocket("/agent/stream")
async def stream_agent(websocket: WebSocket):
    await websocket.accept()
    # 这个端点同样需要鉴权；在生产中应当在连接时验证 token，这里为清晰展示省略
    try:
        while True:
            user_msg = await websocket.receive_text()
            data = json.loads(user_msg)
            command = {
                "type": "command",
                "payload": {
                    "action": "chat",
                    "text": data["text"],
                    "conversation_id": data.get("conversation_id", "default")
                }
            }
            async with ws_lock:
                await gateway_ws.send(json.dumps(command))
                # 读取 Gateway 返回的多个 token 块（假设 Gateway 会逐块发送）
                while True:
                    raw = await asyncio.wait_for(gateway_ws.recv(), timeout=60)
                    msg = json.loads(raw)
                    if msg["type"] == "response":
                        chunk = msg["payload"].get("chunk")  # 假设 payload 中有 chunk 字段
                        if chunk:
                            await websocket.send_text(chunk)
                        else:
                            # 没有更多 chunk，结束本次流
                            await websocket.send_text("[DONE]")
                            break
                    elif msg["type"] == "error":
                        await websocket.send_text(f"Error: {msg['payload']['error']}")
                        break
    except (WebSocketDisconnect, websockets.ConnectionClosed):
        pass

预期结果：启动后，前端通过 WebSocket 连接 ws://<host>:8000/agent/stream，发送一个 {"text": "用 Python 写一个快排"} 消息，就能看到一帧一帧的代码生成过程。这不仅交互感更强，还能让前端显示实时进度条或中断按钮。

步骤 5：通过 Nginx 实现负载均衡与健康检查

单一 API 适配层实例难以支撑高并发，且无法做零宕机重启。我们启动多个 api_server 实例（例如 2 个 8000，1 个 8001），用 Nginx 反向代理到它们，并配置健康检查。

在 docker-compose.yml 中增加 API 服务定义：

  api-1:
    build: .
    command: uvicorn api_server:app --host 0.0.0.0 --port 8000
    environment:
      - GATEWAY_WS_URL=ws://hermes-core:8765
    depends_on:
      - hermes-gateway
    networks:
      - agent-net

  api-2:
    build: .
    command: uvicorn api_server:app --host 0.0.0.0 --port 8000
    environment:
      - GATEWAY_WS_URL=ws://hermes-core:8765
    depends_on:
      - hermes-gateway
    networks:
      - agent-net

  nginx:
    image: nginx:alpine
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
    ports:
      - "443:443"           # HTTPS
    depends_on:
      - api-1
      - api-2
    networks:
      - agent-net

新建 nginx.conf 配置：

events { worker_connections 1024; }

http {
    upstream api_backend {
        server api-1:8000 weight=1 max_fails=2 fail_timeout=30s;
        server api-2:8000 weight=1 max_fails=2 fail_timeout=30s;
    }

    server {
        listen 443 ssl;
        # 生成自签名证书用于本地测试
        ssl_certificate /etc/nginx/certs/selfsigned.crt;
        ssl_certificate_key /etc/nginx/certs/selfsigned.key;

        location / {
            proxy_pass http://api_backend;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;

            # WebSocket 升级支持（关键！）
            proxy_http_version 1.1;
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection "upgrade";
        }
    }
}

自行生成自签名证书（供开发使用）：

mkdir certs
openssl req -x509 -nodes -days 365 -newkey rsa:2048 \
  -keyout certs/selfsigned.key -out certs/selfsigned.crt \
  -subj "/CN=localhost"

预期结果：docker compose up -d 后，Nginx 监听 443 端口，将请求轮流分发给两个 API 实例。手动停掉一个 API 实例，Nginx 在 30 秒内将其标记为失败，请求路由到健康实例。这确保了 Agent API 的高可用。

踩坑经验：如果 WebSocket 连接总是断开，检查 Nginx 配置中是否包含了 Upgrade 和 Connection upgrade 两个头部；缺少任何一个，WebSocket 握手会失败，客户端立即断开。

步骤 6：前端集成示例

前端可以使用标准 fetch 调用 REST 端点，或用 WebSocket 对象连接流式端点。下面是一个简化的 React Hook 示例，演示鉴权与流式消费：

import { useState } from 'react';

export function useAgentStream() {
  const [token, setToken] = useState<string>('');

  const connect = (userInput: string) => {
    const ws = new WebSocket(`wss://api.example.com/agent/stream`);

    ws.onopen = () => {
      ws.send(JSON.stringify({ text: userInput, conversation_id: 'abc123' }));
    };
    ws.onmessage = (e) => {
      if (e.data === '[DONE]') {
        ws.close();
      } else {
        // 逐步更新界面
        appendToAssistantMessage(e.data);
      }
    };
    ws.onerror = (err) => console.error('WS error', err);
    ws.onclose = () => console.log('Stream closed');
    return () => ws.close(); // 清理函数
  };

  return { connect };
}

前端需要先通过登录接口获取 JWT token，并将其附加到 WebSocket 地址的查询参数中（例如 token=eyJ...），由 API 服务在握手时验证。具体实现不再展开，但核心逻辑就这几行。

回顾

做了什么：我们从零构建了一个可投入生产的 API 网关层，把 Hermes Agent 的内部 WebSocket 连接转化为标准 HTTP/WebSocket 接口，并附加了 JWT 鉴权、速率限制、流式输出、负载均衡和健康检查。
花了多久：首次搭建可能需要 60 分钟（包括环境准备），熟悉流程后 15 分钟可完成全新部署。
成果验证：使用 curl -H "Authorization: Bearer <token>" 向 /agent/ask 发送请求，得到 Agent 的文本回复；或者用 Postman/浏览器 WebSocket 客户端体验流式响应。

行动清单

用 Docker Compose 启动 Hermes Gateway，确认 WebSocket 监听正常。
编写 FastAPI 适配层，实现 Gateway WebSocket 客户端和 /agent/ask 端点。
集成 JWT 验证与 slowapi 限流，重启服务测试 429 拦截。
加入 /agent/stream WebSocket 端点，实现逐字流式返回。
配置 Nginx 反向代理，启用 TLS 和负载均衡，然后停机一个实例验证自动 failover。

此刻，你的 Agent 已经穿上了生产级的 HTTP 铠甲。它不再是躲在聊天应用里自言自语的黑箱，而是能够接受任何标准化客户端调用的智能微服务。但一个 Agent 的强大，最终取决于它手中工具的丰富程度。仅靠内置技能远远不够，你需要接入海量的已有工具——下一章，LangChain 集成让 Hermes 可以使用庞大的工具生态，将教你如何将数百个现成的 LangChain 工具包装为 Hermes Skill，让你的 Agent 瞬间长出飞驰的翅膀。

本文章首发在 LearnKu.com 网站上。

7.1. 通过 Gateway 可将 Hermes Agent 暴露为标准化 API 服务

你需要什么

最终成果

架构全景

步骤说明

步骤 1：启动 Hermes Gateway 后台服务

步骤 2：编写 API 适配层（FastAPI）

步骤 3：添加速率限制防止滥用

步骤 4：实现 WebSocket 流式输出，提升前端体验

步骤 5：通过 Nginx 实现负载均衡与健康检查

步骤 6：前端集成示例

回顾

行动清单

社区赞助商

关于 LearnKu

资源推荐

服务提供商

其他信息

7.1. 通过 Gateway 可将 Hermes Agent 暴露为标准化 API 服务

你需要什么

最终成果

架构全景

步骤说明

步骤 1：启动 Hermes Gateway 后台服务

步骤 2：编写 API 适配层（FastAPI）

步骤 3：添加速率限制防止滥用

步骤 4：实现 WebSocket 流式输出，提升前端体验

步骤 5：通过 Nginx 实现负载均衡与健康检查

步骤 6：前端集成示例

回顾

行动清单

社区赞助商

关于 LearnKu

资源推荐

服务提供商

其他信息

请登录

步骤 1：启动 Hermes Gateway 后台服务

步骤 2：编写 API 适配层（FastAPI）

步骤 3：添加速率限制防止滥用

步骤 4：实现 WebSocket 流式输出，提升前端体验

步骤 5：通过 Nginx 实现负载均衡与健康检查

步骤 6：前端集成示例