Hermes WebUI 深度实战：随时随地用手机/浏览器驱动 Hermes Agent——从 SSE 流式传输到多模型路由的完全指南（2026）

Hermes WebUI 今日飙升 1,725 星，成为 GitHub Trending 榜首。它不只是个聊天界面——而是把 Hermes Agent 的能力完整搬到了 Web 和移动端，支持 SSE 流式输出、多模型切换、会话管理、甚至是语音输入。本文将深入拆解其架构设计、实时通信机制、多模型适配层，以及如何基于它快速搭建自己的 AI 对话平台。

目录

1. 背景介绍：为什么需要 Hermes WebUI？
2. 核心概念：Hermes Agent 与 WebUI 的架构哲学
3. 架构分析：从前端 SPA 到后端 API Proxy 的完整链路
4. 代码实战：本地部署 + 二次开发完整指南
5. 性能优化：SSE 流式传输、模型路由与缓存策略
6. 安全与认证：JWT、API Key 管理与多租户隔离
7. 移动端适配：PWA + 响应式设计的工程实践
8. 生产级部署：Docker Compose 编排与 Nginx 反向代理
9. 总结与展望：WebUI 作为 AI Agent 交互层的未来

1. 背景介绍：为什么需要 Hermes WebUI？

1.1 Hermes Agent 的能力与局限

Hermes Agent 是 NousResearch 开发的旗舰级开源 AI Agent 框架，基于 Hermes 系列大模型（Hermmes-3、Hermes-2 等），具备：

• 工具调用（Function Calling）：原生支持 OpenAI Functions 格式
• 多轮对话管理：自动维护上下文窗口，支持长会话
• RAG 集成：可连接向量数据库实现知识增强
• 代码执行：安全沙箱中运行生成的代码
• 多模型支持：可切换 Hermes、Llama、Mistral 等开源模型

但 Hermes Agent 的传统使用方式是 CLI（命令行界面） 和 Python API，存在明显痛点：

1.2 Hermes WebUI 的诞生

Hermes WebUI 由 @nesquena 主导开发，今日（2026-06-02）单日暴涨 1,725 stars，总计 12,253 stars。它的核心定位是：

"The best way to use Hermes Agent from the web or from your phone!"

关键特性一览：

# Hermes WebUI 核心特性（伪代码表示）
features = {
    "real_time_streaming": "SSE (Server-Sent Events) 实现 token-by-token 输出",
    "multi_model_routing": "支持 Hermes/Llama/Mistral/Qwen 等模型一键切换",
    "mobile_first": "PWA + 响应式设计，手机浏览器原生体验",
    "session_management": "多会话隔离，支持重命名/删除/导出",
    "function_calling_ui": "工具调用过程可视化（调用了哪个函数、参数、返回值）",
    "voice_input": "Web Speech API 实现语音转文字输入",
    "dark_mode": "自适应暗黑模式，保护程序员眼睛",
    "local_storage": "会话历史本地存储，刷新页面不丢失",
    "api_key_management": "支持自定义 API Endpoint（可连本地 Ollama）"
}

1.3 为什么现在是使用 Hermes WebUI 的最佳时机？

1. 模型生态成熟：Hermes-3 系列模型已达到 GPT-4 级别性能，且可本地部署
2. Web 标准完善：SSE、WebSocket、Web Speech API 已广泛支持
3. 移动端需求爆发：程序员希望随时随地用手机调 AI（地铁上、咖啡厅里）
4. 开源替代浪潮：用户对闭源 ChatGPT UI 的隐私担忧，转向自托管方案

2. 核心概念：Hermes Agent 与 WebUI 的架构哲学

2.1 Hermes Agent 的核心抽象

要理解 WebUI，必须先理解 Hermes Agent 的后端抽象：

# Hermes Agent 核心类结构（简化版）
class HermesAgent:
    def __init__(self, model: str, temperature: float = 0.7):
        self.model = model                    # 模型名称（如 "hermes-3")
        self.temperature = temperature        # 生成温度
        self.tools = []                       # 注册的工具列表
        self.conversation = []                # 对话历史
    
    def register_tool(self, func: Callable):
        """注册一个可调用的工具（Function Calling）"""
        self.tools.append({
            "type": "function",
            "function": {
                "name": func.__name__,
                "description": func.__doc__,
                "parameters": self._infer_schema(func)
            }
        })
    
    async def chat(self, message: str) -> AsyncGenerator[str, None]:
        """流式对话：yield 每个 token"""
        # 1. 将用户消息加入对话历史
        self.conversation.append({"role": "user", "content": message})
        
        # 2. 调用模型（支持 streaming）
        response_stream = await self._call_model(
            messages=self.conversation,
            tools=self.tools,
            stream=True
        )
        
        # 3. 流式返回 token
        full_response = ""
        async for token in response_stream:
            full_response += token
            yield token
        
        # 4. 将完整回复加入对话历史
        self.conversation.append({"role": "assistant", "content": full_response})
    
    async def _call_model(self, messages, tools, stream):
        """调用底层模型 API（OpenAI 兼容格式）"""
        # 实际实现会调用 Hermes API / Ollama / 其他 OpenAI 兼容接口
        pass

2.2 WebUI 的架构分层

Hermes WebUI 采用经典的 前后端分离 架构：

┌─────────────────────────────────────────────────────────┐
│                      Browser / Mobile                   │
│  ┌─────────────────────────────────────────────────┐   │
│  │          React + TypeScript Frontend             │   │
│  │  (SSE Client, Context Management, UI Components) │   │
│  └───────────────────┬─────────────────────────────┘   │
│                      │ HTTP / SSE                       │
└──────────────────────┼──────────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────────┐
│                  FastAPI Backend Proxy                  │
│  ┌─────────────────────────────────────────────────┐   │
│  │  API Routes (chat, models, sessions)            │   │
│  │  SSE Bridge (stream Hermes response → browser)  │   │
│  │  Auth Middleware (JWT / API Key)                │   │
│  └───────────────────┬─────────────────────────────┘   │
│                      │ HTTP (OpenAI Compatible)        │
└──────────────────────┼──────────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────────┐
│              Hermes Agent / Ollama / API                │
│  (Actual LLM inference)                                │
└─────────────────────────────────────────────────────────┘

关键设计决策：

1. Backend Proxy 的必要性：

   • 浏览器无法直接调用 Hermes API（CORS 限制）
   • 需要统一处理认证、日志、限流
   • SSE 流式传输需要专门的流式代理

2. 为什么选择 SSE 而非 WebSocket？

   • SSE 是 HTTP 标准，防火墙/NAT 更友好
   • 实现更简单（无需双向通信）
   • 自动重连机制（浏览器原生支持）
   • 对于 LLM 场景（服务端 → 客户端单向流），SSE 完全够用

2.3 核心数据结构

// TypeScript 前端类型定义
interface ChatMessage {
  id: string;                      // 消息唯一 ID（UUID）
  role: 'user' | 'assistant' | 'system' | 'tool';
  content: string;                 // 消息内容（Markdown 格式）
  timestamp: number;               // 时间戳（毫秒）
  toolCalls?: ToolCall[];          // 工具调用记录（可选）
}

interface ToolCall {
  id: string;
  name: string;                   // 工具名称（如 "get_weather"）
  arguments: Record<string, any>; // 调用参数（JSON）
  result?: any;                   // 调用结果（可选，流式返回后填充）
}

interface Conversation {
  id: string;
  title: string;                  // 会话标题（自动生成或从第一条消息提取）
  messages: ChatMessage[];
  model: string;                  // 使用的模型（如 "hermes-3"）
  createdAt: number;
  updatedAt: number;
}

interface ModelInfo {
  id: string;                     // 模型 ID
  name: string;                   // 显示名称
  provider: 'hermes' | 'ollama' | 'openai' | 'anthropic';
  contextLength: number;           // 上下文窗口大小（token 数）
  supportsTools: boolean;          // 是否支持工具调用
  supportsVision: boolean;         // 是否支持视觉输入
}

3. 架构分析：从前端 SPA 到后端 API Proxy 的完整链路

3.1 前端架构：组件拆分与状态管理

Hermes WebUI 前端采用 React + TypeScript + Vite 技术栈，核心组件树：

App
├── AuthProvider (认证上下文)
├── Router
│   ├── /chat/:conversationId (ChatPage)
│   │   ├── Sidebar (会话列表 + 新建按钮)
│   │   ├── ChatWindow
│   │   │   ├── MessageList (消息列表，虚拟滚动)
│   │   │   │   └── MessageBubble (单条消息)
│   │   │   │       ├── MarkdownRenderer (Markdown 渲染)
│   │   │   │       └── ToolCallVisualizer (工具调用可视化)
│   │   │   ├── InputArea (输入区域)
│   │   │   │   ├── TextInput (文本输入框)
│   │   │   │   ├── VoiceButton (语音输入按钮)
│   │   │   │   └── ModelSelector (模型选择器)
│   │   │   └── StreamingIndicator (流式输出指示器)
│   │   └── Header (顶部栏：标题、设置、主题切换)
│   ├── /settings (SettingsPage)
│   └── /login (LoginPage)

状态管理方案：

// 使用 Zustand（轻量级状态管理库）
import { create } from 'zustand';

interface ChatStore {
  // 会话列表
  conversations: Conversation[];
  activeConversationId: string | null;
  
  // 流式输出状态
  streamingMessage: string;
  isStreaming: boolean;
  
  // 模型列表
  availableModels: ModelInfo[];
  currentModel: string;
  
  // Actions
  sendMessage: (content: string) => Promise<void>;
  switchModel: (modelId: string) => void;
  createConversation: () => void;
  deleteConversation: (id: string) => void;
}

const useChatStore = create<ChatStore>((set, get) => ({
  conversations: [],
  activeConversationId: null,
  streamingMessage: '',
  isStreaming: false,
  availableModels: [],
  currentModel: 'hermes-3',
  
  sendMessage: async (content: string) => {
    const { activeConversationId, conversations } = get();
    const conversation = conversations.find(c => c.id === activeConversationId);
    
    // 1. 添加用户消息到本地状态（乐观更新）
    const userMessage: ChatMessage = {
      id: crypto.randomUUID(),
      role: 'user',
      content,
      timestamp: Date.now()
    };
    conversation.messages.push(userMessage);
    set({ isStreaming: true, streamingMessage: '' });
    
    // 2. 调用后端 SSE 接口
    const response = await fetch(`/api/chat`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        conversation_id: activeConversationId,
        message: content,
        model: get().currentModel,
        stream: true
      })
    });
    
    // 3. 解析 SSE 流
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let assistantMessage = '';
    
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      
      const chunk = decoder.decode(value);
      const lines = chunk.split('\n');
      
      for (const line of lines) {
        if (line.startsWith('data: ')) {
          const data = JSON.parse(line.slice(6));
          if (data.token) {
            assistantMessage += data.token;
            set({ streamingMessage: assistantMessage });
          }
        }
      }
    }
    
    // 4. 流式输出完成，保存完整消息到会话
    const assistantMsg: ChatMessage = {
      id: crypto.randomUUID(),
      role: 'assistant',
      content: assistantMessage,
      timestamp: Date.now()
    };
    conversation.messages.push(assistantMsg);
    set({ isStreaming: false, streamingMessage: '' });
  },
  
  switchModel: (modelId: string) => {
    set({ currentModel: modelId });
    // 持久化到 localStorage
    localStorage.setItem('currentModel', modelId);
  },
  
  // ... 其他 actions
}));

3.2 后端架构：FastAPI + SSE Bridge

后端采用 FastAPI（Python 异步框架），核心路由：

# backend/main.py（简化版）
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import json

app = FastAPI(title="Hermes WebUI Backend")

class ChatRequest(BaseModel):
    conversation_id: str
    message: str
    model: str = "hermes-3"
    stream: bool = True
    temperature: float = 0.7

class ModelInfo(BaseModel):
    id: str
    name: str
    provider: str
    context_length: int
    supports_tools: bool

# 内存中的会话存储（生产环境用 Redis）
conversations = {}

@app.get("/api/models", response_model=list[ModelInfo])
async def list_models():
    """列出可用模型"""
    return [
        ModelInfo(
            id="hermes-3",
            name="Hermes 3 (Local)",
            provider="ollama",
            context_length=32768,
            supports_tools=True
        ),
        ModelInfo(
            id="llama-3.1",
            name="Llama 3.1 8B (Local)",
            provider="ollama",
            context_length=8192,
            supports_tools=True
        ),
        # ... 其他模型
    ]

@app.post("/api/chat")
async def chat(request: ChatRequest):
    """对话接口（支持 SSE 流式输出）"""
    if not request.stream:
        # 非流式：返回完整响应
        return {"message": "Non-streaming not yet implemented"}
    
    # 流式：返回 SSE 响应
    async def generate():
        # 1. 从 Ollama / Hermes API 获取流式响应
        async for token in call_hermes_api(
            message=request.message,
            model=request.model,
            conversation_id=request.conversation_id
        ):
            # 2. 格式化为 SSE 格式
            yield f"data: {json.dumps({'token': token})}\n\n"
        
        # 3. 流结束标记
        yield f"data: {json.dumps({'done': True})}\n\n"
    
    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"  # 禁用 Nginx 缓冲
        }
    )

async def call_hermes_api(message: str, model: str, conversation_id: str):
    """调用 Hermes / Ollama API（流式）"""
    import aiohttp
    
    # 1. 获取会话历史
    conversation = conversations.get(conversation_id, [])
    
    # 2. 构造 OpenAI 兼容格式的请求
    payload = {
        "model": model,
        "messages": conversation + [{"role": "user", "content": message}],
        "stream": True,
        "temperature": 0.7
    }
    
    # 3. 调用本地 Ollama API（示例）
    async with aiohttp.ClientSession() as session:
        async with session.post(
            "http://localhost:11434/api/chat",  # Ollama 默认地址
            json=payload
        ) as response:
            async for line in response.content:
                if line:
                    data = json.loads(line)
                    if 'message' in data and 'content' in data['message']:
                        yield data['message']['content']

@app.get("/api/conversations", response_model=list)
async def list_conversations():
    """列出所有会话"""
    return [
        {"id": cid, "title": msgs[0]["content"][:50] if msgs else "New Conversation"}
        for cid, msgs in conversations.items()
    ]

@app.post("/api/conversations/{conversation_id}/messages")
async def add_message(conversation_id: str, message: dict):
    """添加消息到会话（用于历史记录）"""
    if conversation_id not in conversations:
        conversations[conversation_id] = []
    conversations[conversation_id].append(message)
    return {"ok": True}

3.3 SSE 流式传输的底层机制

SSE（Server-Sent Events） 是 HTTP 长连接的标准化方案，格式非常简单：

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

data: {"token": "Hello"}

data: {"token": " world"}

data: {"done": true}

前端解析 SSE 的两种方式：

// 方式 1：使用 Fetch API（推荐，更灵活）
async function streamChat(message: string) {
  const response = await fetch('/api/chat', {
    method: 'POST',
    body: JSON.stringify({ message, stream: true })
  });
  
  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    
    const chunk = decoder.decode(value);
    // 解析 SSE 格式（以 "\n\n" 分隔事件）
    const events = chunk.split('\n\n');
    for (const event of events) {
      if (event.startsWith('data: ')) {
        const data = JSON.parse(event.slice(6));
        console.log('Received token:', data.token);
      }
    }
  }
}

// 方式 2：使用 EventSource API（原生支持，但仅支持 GET）
const eventSource = new EventSource('/api/chat-stream?message=' + encodeURIComponent(message));
eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log('Received:', data);
};
eventSource.onerror = () => eventSource.close();

SSE vs WebSocket vs Long Polling：

4. 代码实战：本地部署 + 二次开发完整指南

4.1 快速启动：Docker Compose 一键部署

# docker-compose.yml
version: '3.8'

services:
  # Ollama（本地 LLM 推理引擎）
  ollama:
    image: ollama/ollama:latest
    ports:
      - "11434:11434"
    volumes:
      - ollama_data:/root/.ollama
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
  
  # Hermes WebUI Backend
  backend:
    build: ./backend
    ports:
      - "8000:8000"
    environment:
      - OLLAMA_BASE_URL=http://ollama:11434
      - JWT_SECRET=your-secret-key-change-me
    depends_on:
      - ollama
  
  # Hermes WebUI Frontend（生产环境用 Nginx 托管静态文件）
  frontend:
    build: ./frontend
    ports:
      - "3000:80"
    depends_on:
      - backend

volumes:
  ollama_data:

启动步骤：

# 1. 克隆仓库
git clone https://github.com/nesquena/hermes-webui.git
cd hermes-webui

# 2. 下载 Hermes 3 模型（通过 Ollama）
docker-compose up -d ollama
docker exec -it hermes-webui-ollama-1 ollama pull hermes3

# 3. 启动所有服务
docker-compose up -d

# 4. 访问前端
open http://localhost:3000

4.2 从源码构建：前端开发环境

# 1. 安装依赖
cd frontend
npm install

# 2. 启动开发服务器（热重载）
npm run dev
# 访问 http://localhost:5173

# 3. 构建生产版本
npm run build
# 输出到 dist/ 目录

前端环境变量配置（.env.development）：

VITE_API_BASE_URL=http://localhost:8000/api
VITE_ENABLE_VOICE_INPUT=true
VITE_DEFAULT_MODEL=hermes-3
VITE_MAX_TOKENS=4096

4.3 后端开发：添加自定义工具调用支持

Hermes WebUI 的一个核心优势是 工具调用可视化。让我们扩展后端，支持自定义工具：

# backend/tools.py
from typing import Callable, Any
import inspect

class ToolRegistry:
    def __init__(self):
        self.tools = {}
    
    def register(self, func: Callable):
        """注册工具的装饰器"""
        schema = self._infer_schema(func)
        self.tools[func.__name__] = {
            "function": schema,
            "handler": func
        }
        return func
    
    def _infer_schema(self, func: Callable) -> dict:
        """从函数签名自动推断 JSON Schema"""
        sig = inspect.signature(func)
        parameters = {}
        
        for name, param in sig.parameters.items():
            param_type = param.annotation
            parameters[name] = {
                "type": self._python_type_to_json(param_type),
                "description": ""  # 从 docstring 提取（可用 ast 解析）
            }
        
        return {
            "name": func.__name__,
            "description": func.__doc__ or "",
            "parameters": {
                "type": "object",
                "properties": parameters,
                "required": [
                    name for name, p in sig.parameters.items()
                    if p.default == inspect.Parameter.empty
                ]
            }
        }
    
    def _python_type_to_json(self, py_type: Any) -> str:
        """Python 类型 → JSON Schema 类型"""
        mapping = {
            str: "string",
            int: "integer",
            float: "number",
            bool: "boolean",
            list: "array",
            dict: "object"
        }
        return mapping.get(py_type, "string")
    
    def get_openai_schema(self) -> list:
        """返回 OpenAI Functions 格式的工具列表"""
        return [
            {"type": "function", "function": tool["function"]}
            for tool in self.tools.values()
        ]
    
    async def execute(self, name: str, arguments: dict) -> Any:
        """执行工具调用"""
        if name not in self.tools:
            raise ValueError(f"Tool {name} not found")
        
        handler = self.tools[name]["handler"]
        return await handler(**arguments)

# 全局工具注册表
tool_registry = ToolRegistry()

# 示例工具：获取天气
@tool_registry.register
async def get_weather(city: str, unit: str = "celsius") -> dict:
    """
    获取指定城市的天气信息。
    
    Args:
        city: 城市名称（如 "Beijing"）
        unit: 温度单位（"celsius" 或 "fahrenheit"）
    
    Returns:
        包含温度、湿度、天气状况的字典
    """
    # 实际实现会调用天气 API（如 OpenWeatherMap）
    import random
    return {
        "city": city,
        "temperature": random.randint(10, 35),
        "unit": unit,
        "humidity": random.randint(40, 90),
        "condition": random.choice(["Sunny", "Cloudy", "Rainy"])
    }

集成到聊天接口：

# backend/main.py（扩展版）
from fastapi import FastAPI
from .tools import tool_registry

@app.post("/api/chat")
async def chat(request: ChatRequest):
    async def generate():
        # 1. 调用 Hermes API，传入工具定义
        async for chunk in call_hermes_api(
            message=request.message,
            model=request.model,
            tools=tool_registry.get_openai_schema()  # 传入工具列表
        ):
            # 2. 检查是否是工具调用
            if chunk.get("tool_calls"):
                for tool_call in chunk["tool_calls"]:
                    tool_name = tool_call["function"]["name"]
                    arguments = json.loads(tool_call["function"]["arguments"])
                    
                    # 3. 执行工具
                    result = await tool_registry.execute(tool_name, arguments)
                    
                    # 4. 将结果返回给 LLM（继续对话）
                    yield f"data: {json.dumps({
                        'tool_call': tool_name,
                        'result': result
                    })}\n\n"
            
            # 5. 正常的 token 输出
            if chunk.get("token"):
                yield f"data: {json.dumps({'token': chunk['token']})}\n\n"
    
    return StreamingResponse(generate(), media_type="text/event-stream")

4.4 前端工具调用可视化

// frontend/components/ToolCallVisualizer.tsx
import React, { useState } from 'react';

interface ToolCallProps {
  name: string;
  arguments: Record<string, any>;
  result?: any;
  status: 'calling' | 'completed' | 'error';
}

export const ToolCallVisualizer: React.FC<ToolCallProps> = ({
  name, arguments: args, result, status
}) => {
  const [expanded, setExpanded] = useState(false);
  
  return (
    <div className="tool-call border-l-4 border-blue-500 pl-4 my-2">
      {/* 工具调用头部 */}
      <div
        className="flex items-center cursor-pointer"
        onClick={() => setExpanded(!expanded)}
      >
        <span className="text-blue-500 font-mono">🔧 {name}</span>
        {status === 'calling' && <Spinner className="ml-2" />}
        {status === 'completed' && <span className="text-green-500 ml-2">✓</span>}
        {status === 'error' && <span className="text-red-500 ml-2">✗</span>}
      </div>
      
      {/* 展开详情 */}
      {expanded && (
        <div className="mt-2 text-sm">
          <div className="bg-gray-100 p-2 rounded">
            <strong>Arguments:</strong>
            <pre className="mt-1">{JSON.stringify(args, null, 2)}</pre>
          </div>
          {result && (
            <div className="bg-green-50 p-2 rounded mt-2">
              <strong>Result:</strong>
              <pre className="mt-1">{JSON.stringify(result, null, 2)}</pre>
            </div>
          )}
        </div>
      )}
    </div>
  );
};

5. 性能优化：SSE 流式传输、模型路由与缓存策略

5.1 SSE 流式传输的性能瓶颈与优化

问题 1：首 Token 延迟（Time to First Token, TTFT）过高

• 原因：LLM 推理需要预处理输入（tokenization、KV Cache 加载）
• 优化方案：
  1. Prompt Caching（vLLM / Ollama 支持）：缓存系统提示词的 KV Cache
  2. 量化模型（GPTQ / AWQ）：减少显存占用，加快推理速度
  3. 流式输出立即返回：不要在服务端缓冲完整回复再发送

# 错误的做法：缓冲完整回复
async def bad_stream():
    full_response = ""
    async for token in llm.generate(prompt):
        full_response += token
    yield f"data: {json.dumps({'token': full_response})}\n\n"  # ❌ 一次性返回

# 正确的做法：token 级流式返回
async def good_stream():
    async for token in llm.generate(prompt):
        yield f"data: {json.dumps({'token': token})}\n\n"  # ✅ 立即返回

问题 2：SSE 连接占用过多内存

• 原因：每个 SSE 连接占用一个协程（coroutine），高并发时内存暴涨
• 优化方案：
  1. 连接超时：设置 SSE 连接的最大持续时间（如 5 分钟）
  2. 心跳机制：定期发送 ping 事件，检测死连接

async def sse_with_heartbeat(generate_tokens):
    import asyncio
    
    queue = asyncio.Queue()
    
    # 启动生成任务
    async def produce():
        async for token in generate_tokens():
            await queue.put(token)
        await queue.put(None)  # 结束标记
    
    asyncio.create_task(produce())
    
    # 发送 token + 心跳
    while True:
        try:
            # 等待 token 或超时（30 秒心跳）
            token = await asyncio.wait_for(queue.get(), timeout=30.0)
            if token is None:
                yield "data: {\"done\": true}\n\n"
                break
            yield f"data: {json.dumps({'token': token})}\n\n"
        except asyncio.TimeoutError:
            # 发送心跳
            yield ": heartbeat\n\n"

5.2 模型路由：根据请求特征动态选择模型

不同任务适合不同模型，Hermes WebUI 可以实现 智能模型路由：

# backend/model_router.py
from enum import Enum

class TaskType(Enum):
    CODE_GENERATION = "code"
    CREATIVE_WRITING = "creative"
    FACTUAL_QA = "qa"
    TOOL_USE = "tool"
    LONG_CONTEXT = "long"

class ModelRouter:
    def __init__(self):
        self.rules = [
            # (任务类型, 模型 ID, 优先级)
            (TaskType.CODE_GENERATION, "codellama-7b", 10),
            (TaskType.CODE_GENERATION, "hermes-3", 5),
            (TaskType.CREATIVE_WRITING, "hermes-3", 10),
            (TaskType.FACTUAL_QA, "llama-3.1", 8),
            (TaskType.TOOL_USE, "hermes-3", 10),  # Hermes 工具调用最强
            (TaskType.LONG_CONTEXT, "hermes-3", 10),  # 32K 上下文
        ]
    
    def route(self, message: str, conversation: list) -> str:
        """根据消息内容路由到最合适的模型"""
        task_type = self._classify_task(message)
        
        # 找到匹配规则中优先级最高的模型
        candidates = [
            (model, priority) for (tt, model, priority) in self.rules
            if tt == task_type
        ]
        if not candidates:
            return "hermes-3"  # 默认模型
        
        best_model = max(candidates, key=lambda x: x[1])[0]
        return best_model
    
    def _classify_task(self, message: str) -> TaskType:
        """简单的基于关键词的任务分类"""
        message_lower = message.lower()
        
        if any(kw in message_lower for kw in ["code", "function", "bug", "debug"]):
            return TaskType.CODE_GENERATION
        elif any(kw in message_lower for kw in ["write", "story", "poem", "creative"]):
            return TaskType.CREATIVE_WRITING
        elif any(kw in message_lower for kw in ["what", "who", "when", "where", "why"]):
            return TaskType.FACTUAL_QA
        elif "{" in message or "tool" in message_lower:
            return TaskType.TOOL_USE
        elif len(message) > 2000:
            return TaskType.LONG_CONTEXT
        else:
            return TaskType.FACTUAL_QA

5.3 缓存策略：减少重复计算

1. 会话历史缓存（Redis）

# backend/cache.py
import redis
import json

class ConversationCache:
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url, decode_responses=True)
    
    def get(self, conversation_id: str) -> list:
        """获取会话历史"""
        key = f"conv:{conversation_id}"
        data = self.redis.get(key)
        return json.loads(data) if data else []
    
    def set(self, conversation_id: str, messages: list, ttl: int = 3600):
        """保存会话历史（TTL 1 小时）"""
        key = f"conv:{conversation_id}"
        self.redis.setex(key, ttl, json.dumps(messages))
    
    def append(self, conversation_id: str, message: dict):
        """追加一条消息"""
        messages = self.get(conversation_id)
        messages.append(message)
        self.set(conversation_id, messages)

2. RAG 检索缓存（向量数据库 + Redis）

# 对频繁查询的知识库问题，缓存检索结果
async def rag_with_cache(query: str, cache: ConversationCache):
    cache_key = f"rag:{hash(query)}"
    
    # 1. 检查缓存
    cached = cache.get(cache_key)
    if cached:
        return cached
    
    # 2. 执行向量检索
    results = await vector_db.search(query, top_k=5)
    
    # 3. 缓存结果（TTL 1 小时）
    cache.set(cache_key, results, ttl=3600)
    return results

6. 安全与认证：JWT、API Key 管理与多租户隔离

6.1 认证方案：JWT + API Key 双模式

Hermes WebUI 支持两种认证方式：

# backend/auth.py
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, APIKeyHeader
import jwt
from functools import lru_cache

# JWT 认证
jwt_scheme = HTTPBearer()
api_key_scheme = APIKeyHeader(name="X-API-Key")

def get_current_user(
    jwt_token: str = Depends(jwt_scheme),
    api_key: str = Depends(api_key_scheme)
):
    """支持 JWT 或 API Key 认证"""
    if jwt_token:
        try:
            payload = jwt.decode(
                jwt_token.credentials,
                key=get_secret_key(),
                algorithms=["HS256"]
            )
            return payload["sub"]  # 返回用户 ID
        except jwt.ExpiredSignatureError:
            raise HTTPException(401, "Token expired")
        except jwt.InvalidTokenError:
            raise HTTPException(401, "Invalid token")
    
    elif api_key:
        # 验证 API Key（从数据库查询）
        user_id = validate_api_key(api_key)
        if not user_id:
            raise HTTPException(401, "Invalid API key")
        return user_id
    
    else:
        raise HTTPException(401, "Authentication required")

# 保护路由示例
@app.post("/api/chat")
async def chat(
    request: ChatRequest,
    user_id: str = Depends(get_current_user)  # ✅ 需要认证
):
    # 只有认证用户才能访问
    pass

6.2 API Key 管理：多租户隔离

# backend/models.py
from pydantic import BaseModel
import hashlib

class APIKey(BaseModel):
    id: str
    user_id: str
    key_hash: str  # API Key 的 SHA-256 哈希（不存储明文）
    name: str       # 用户给 Key 起的名字（如 "My iPhone"）
    created_at: int
    last_used_at: int
    rate_limit: int = 100  # 每小时请求数限制

def generate_api_key(user_id: str) -> tuple[str, str]:
    """生成 API Key（返回明文 key 和哈希值）"""
    import secrets
    
    # 格式：sk-<32 位随机字符>
    raw_key = f"sk-{secrets.token_urlsafe(32)}"
    key_hash = hashlib.sha256(raw_key.encode()).hexdigest()
    
    return raw_key, key_hash  # 明文只显示一次，提醒用户保存

6.3 多租户隔离：会话数据分片

# backend/multi_tenant.py
from typing import Dict
import json

class TenantManager:
    def __init__(self):
        self.tenant_data: Dict[str, dict] = {}  # user_id -> {conversations, settings}
    
    def get_conversations(self, user_id: str) -> list:
        """获取租户的会话列表（隔离）"""
        if user_id not in self.tenant_data:
            self.tenant_data[user_id] = {"conversations": [], "settings": {}}
        return self.tenant_data[user_id]["conversations"]
    
    def save_conversation(self, user_id: str, conversation: dict):
        """保存会话（仅限该租户）"""
        conversations = self.get_conversations(user_id)
        # 更新或插入
        for i, conv in enumerate(conversations):
            if conv["id"] == conversation["id"]:
                conversations[i] = conversation
                return
        conversations.append(conversation)

7. 移动端适配：PWA + 响应式设计的工程实践

7.1 PWA（Progressive Web App）配置

Hermes WebUI 可以作为 PWA 安装到手机主屏幕，获得原生 App 体验：

// frontend/public/manifest.json
{
  "name": "Hermes WebUI",
  "short_name": "Hermes",
  "description": "The best way to use Hermes Agent from your phone!",
  "icons": [
    {
      "src": "/icon-192x192.png",
      "sizes": "192x192",
      "type": "image/png"
    },
    {
      "src": "/icon-512x512.png",
      "sizes": "512x512",
      "type": "image/png"
    }
  ],
  "start_url": "/",
  "display": "standalone",
  "theme_color": "#000000",
  "background_color": "#ffffff"
}

注册 Service Worker（实现离线缓存）：

// frontend/src/service-worker.js
const CACHE_NAME = 'hermes-webui-v1';
const urlsToCache = [
  '/',
  '/index.html',
  '/assets/index.js',
  '/assets/index.css'
];

self.addEventListener('install', (event) => {
  event.waitUntil(
    caches.open(CACHE_NAME).then((cache) => cache.addAll(urlsToCache))
  );
});

self.addEventListener('fetch', (event) => {
  event.respondWith(
    caches.match(event.request).then((response) => {
      // 缓存命中，直接返回
      if (response) return response;
      
      // 否则发起网络请求
      return fetch(event.request).then((response) => {
        // 克隆响应（response 是流，只能读一次）
        const responseClone = response.clone();
        
        caches.open(CACHE_NAME).then((cache) => {
          cache.put(event.request, responseClone);
        });
        
        return response;
      });
    })
  );
});

7.2 响应式设计：移动端输入框优化

// frontend/components/InputArea.tsx
import React, { useState, useEffect } from 'react';

export const InputArea: React.FC = () => {
  const [message, setMessage] = useState('');
  const [isMobile, setIsMobile] = useState(false);
  
  // 检测移动端
  useEffect(() => {
    const checkMobile = () => {
      setIsMobile(window.innerWidth < 768);
    };
    checkMobile();
    window.addEventListener('resize', checkMobile);
    return () => window.removeEventListener('resize', checkMobile);
  }, []);
  
  // 移动端虚拟键盘弹出时，滚动到输入框
  useEffect(() => {
    if (isMobile) {
      const input = document.querySelector('textarea');
      input?.addEventListener('focus', () => {
        setTimeout(() => {
          input.scrollIntoView({ behavior: 'smooth', block: 'center' });
        }, 300);  // 等待虚拟键盘弹出
      });
    }
  }, [isMobile]);
  
  return (
    <div className={`input-area ${isMobile ? 'mobile' : 'desktop'}`}>
      <textarea
        value={message}
        onChange={(e) => setMessage(e.target.value)}
        placeholder="Type your message..."
        rows={isMobile ? 3 : 5}  // 移动端减少行数
        className="w-full p-2 border rounded"
      />
      <button onClick={() => sendMessage(message)}>
        Send
      </button>
    </div>
  );
};

8. 生产级部署：Docker Compose 编排与 Nginx 反向代理

8.1 生产环境 Docker Compose 配置

# docker-compose.prod.yml
version: '3.8'

services:
  # Nginx 反向代理（统一入口）
  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/nginx.conf
      - ./ssl:/etc/nginx/ssl
    depends_on:
      - frontend
      - backend
  
  # 前端（静态文件由 Nginx 托管）
  frontend:
    build:
      context: ./frontend
      dockerfile: Dockerfile.prod
    # 不需要暴露端口（由 Nginx 代理）
  
  # 后端（FastAPI）
  backend:
    build: ./backend
    environment:
      - ENV=production
      - OLLAMA_BASE_URL=http://ollama:11434
      - REDIS_URL=redis://redis:6379
      - JWT_SECRET=${JWT_SECRET}  # 从 .env 文件读取
    depends_on:
      - ollama
      - redis
  
  # Ollama（本地 LLM）
  ollama:
    image: ollama/ollama:latest
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    volumes:
      - ollama_data:/root/.ollama
  
  # Redis（缓存 + 会话存储）
  redis:
    image: redis:alpine
    command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru
    volumes:
      - redis_data:/data

volumes:
  ollama_data:
  redis_data:

8.2 Nginx 配置：反向代理 + SSE 支持

# nginx/nginx.conf
events {
    worker_connections 1024;
}

http {
    upstream frontend {
        server frontend:80;
    }
    
    upstream backend {
        server backend:8000;
    }
    
    server {
        listen 80;
        server_name your-domain.com;
        
        # 重定向到 HTTPS
        return 301 https://$server_name$request_uri;
    }
    
    server {
        listen 443 ssl http2;
        server_name your-domain.com;
        
        ssl_certificate /etc/nginx/ssl/fullchain.pem;
        ssl_certificate_key /etc/nginx/ssl/privkey.pem;
        
        # 前端静态文件
        location / {
            proxy_pass http://frontend;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
        }
        
        # 后端 API（支持 SSE）
        location /api/ {
            proxy_pass http://backend;
            proxy_http_version 1.1;
            proxy_set_header Host $host;
            proxy_set_header Connection '';
            
            # SSE 关键配置
            proxy_buffering off;          # ✅ 禁用缓冲
            proxy_cache off;              # ✅ 禁用缓存
            chunked_transfer_encoding on; # ✅ 启用分块传输
            
            # 超时设置（SSE 连接可能持续很久）
            proxy_connect_timeout 60s;
            proxy_send_timeout 300s;      # 5 分钟
            proxy_read_timeout 300s;
        }
    }
}

9. 总结与展望：WebUI 作为 AI Agent 交互层的未来

9.1 本文回顾

本文深度拆解了 Hermes WebUI 的架构设计与工程实现，涵盖：

1. 背景与动机：为什么 CLI 不够用，WebUI 如何填补空白
2. 架构分析：前后端分离、SSE 流式传输、模型路由
3. 代码实战：Docker 部署、工具调用扩展、前端可视化
4. 性能优化：SSE 优化、模型路由、缓存策略
5. 安全认证：JWT + API Key、多租户隔离
6. 移动端适配：PWA、响应式设计
7. 生产部署：Docker Compose、Nginx 反向代理

9.2 Hermes WebUI 的技术亮点

9.3 未来展望：WebUI 作为 AI Agent 的标准交互层

趋势 1：从"聊天界面"到"Agent 工作台"

未来的 WebUI 不只是聊天，而是：

• 多步骤任务可视化：显示 Agent 的规划、执行、反思过程
• 工具市场：用户可以浏览、安装、配置新工具
• 知识库管理：上传文档、构建 RAG 知识库

趋势 2：多模态交互

• 语音输入/输出：Web Speech API + TTS
• 图片输入：支持上传图片，调用视觉模型（如 LLaVA）
• 代码执行结果可视化：图表、表格、交互式组件

趋势 3：本地优先（Local First）

• IndexedDB 存储：会话历史完全本地化
• 离线推理：Service Worker 缓存模型权重（未来 WebGPU 可行）
• 端到端加密：即使服务器端也无法读取用户对话

9.4 立即行动

想要快速上手 Hermes WebUI？

# 1. 克隆仓库
git clone https://github.com/nesquena/hermes-webui.git
cd hermes-webui

# 2. 一键启动（Docker Compose）
docker-compose up -d

# 3. 打开浏览器
open http://localhost:3000

# 4. 开始对话！

想要深度定制？

• 前端：修改 frontend/src/components/MessageBubble.tsx 自定义消息样式
• 后端：在 backend/tools.py 中添加自己的工具
• 模型：编辑 docker-compose.yml，切换到其他 Ollama 模型

参考资源

• Hermes WebUI GitHub：https://github.com/nesquena/hermes-webui
• Hermes Agent 文档：https://github.com/NousResearch/hermes-agent
• Ollama 官方文档：https://ollama.com/docs
• SSE 规范：https://html.spec.whatwg.org/multipage/server-sent-events.html
• FastAPI 流式响应：https://fastapi.tiangolo.com/advanced/response-directly/?h=streaming#streaming-response

作者注：本文基于 Hermes WebUI 2026-06-02 版本的架构分析，具体实现可能随版本更新而变化。建议结合官方文档和源码阅读本文。

字数统计：约 12,500 字

适合读者：有一定 Web 开发经验的程序员，对 AI Agent、LLM 推理、Web 全栈开发感兴趣。

免责声明：本文所述技术方案仅供参考，生产环境部署请根据实际需求调整安全配置。