给 Hermes Agent 装上 Web 控制台：后台运行的 AI 助手，终于看得见了｜TaoToken 配置实战

1. 后台跑的 Hermes Agent，为什么需要一块“仪表盘”

Hermes Agent 是一个能理解复杂指令、调用工具、执行多步任务的 AI 智能体框架，它的核心运行模式通常是异步、后台化的。也就是说，你给它一个任务，它会在后台默默拆解、调用工具、生成中间结果，最后把答案交回来。问题就出在这个“默默”上——你根本不知道它现在在干什么。

我试过在终端里挂一个 Hermes Agent 跑长任务，日志刷得飞快，但想确认“它到底卡在哪一步”“刚才那个工具调用返回了什么”“任务是不是已经失败了”，只能靠肉眼在几千行输出里翻。更麻烦的是，如果 Agent 是作为常驻服务跑在服务器上，你连终端都看不到，只能靠猜。

这种“黑盒”体验对开发者调试和日常监控来说，确实挺折磨的。具体痛点可以归成四类：

• 状态不直观：无法一眼看到所有运行中、已完成、失败的任务。
• 交互能力弱：想临时给 Agent 补一条指令，往往要重启进程或走复杂的 IPC。
• 历史追溯难：翻日志找某次特定对话或工具调用记录，效率极低。
• 缺乏实时性：看不到 Agent 的“思考过程”，比如它下一步打算调用哪个工具、为什么这么选。

一个轻量级的 Web 控制台，正好能解决这些问题。它不需要多复杂，只要能实时展示任务状态、历史消息，并且允许你从浏览器里直接发新指令，就已经比纯 CLI 好用太多了。下面我就从零开始，把这块控制台搭起来，并且把模型调用通道统一接到 TaoToken 上，让整个 AI 助手的运行环境变得可观测、可交互。

2. 前置准备：用 TaoToken 统一模型调用通道

在动手写控制台之前，先把模型调用这条链路理顺。Hermes Agent 本身不绑定特定模型供应商，它需要一个兼容 OpenAI 接口的 API 通道。如果你同时用多个模型（比如 Claude 做推理、GPT 做工具调用），每个都单独配 Key、单独改 base_url，维护起来很烦。

TaoToken 在这里的作用就是提供一个统一的 Key 和 API 通道。你只需要在 TaoToken 控制台创建一个 API Key，然后把 Hermes Agent 的模型请求指向 TaoToken 的 API 地址，后面换模型、加模型都只改配置，不用动代码。

具体操作分三步：

第一步，打开 TaoToken 官网注册并登录，进入控制台。官网地址是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，注册流程很标准，邮箱验证后就能进控制台。

第二步，在控制台左侧找到 API Keys 页面，创建一个新的 Key。建议给这个 Key 起个能识别的名字，比如hermes-agent-console，方便后面排查问题时区分。创建完成后把 Key 复制出来，它只会完整显示一次。

第三步，确认你要用的模型。TaoToken 的模型对话页面可以直观地测试各个模型是否可用，地址是 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。你可以先在页面上发一条测试消息，确认模型能正常返回，再把它写进 Hermes Agent 的配置里。

这里有个细节要注意：TaoToken 的 API 基础地址是https://taotoken.net/api，这个地址不带任何 UTM 参数，直接用于代码里的base_url。而官网和控制台页面链接才带 UTM，两者不要混用。

配置写进 Hermes Agent 的config.toml时，大概长这样：

[model] provider = "openai-compatible" base_url = "https://taotoken.net/api" api_key = "sk-你的TaoTokenKey" model_name = "claude-3-5-sonnet" max_tokens = 4096 temperature = 0.7 [agent] name = "hermes-console-demo" max_iterations = 15 tool_timeout = 30

如果你更习惯用 JSON 配置，对应的settings.json骨架如下：

{ "model": { "provider": "openai-compatible", "base_url": "https://taotoken.net/api", "api_key": "sk-你的TaoTokenKey", "model_name": "claude-3-5-sonnet", "max_tokens": 4096, "temperature": 0.7 }, "agent": { "name": "hermes-console-demo", "max_iterations": 15, "tool_timeout": 30 }, "console": { "host": "0.0.0.0", "port": 8000, "enable_websocket": true } }

把 Key 写进配置文件后，建议先跑一个最小验证，确认通道是通的。可以用 curl 直接打 TaoToken 的 API：

curl -X POST https://taotoken.net/api/v1/chat/completions \ -H "Authorization: Bearer sk-你的TaoTokenKey" \ -H "Content-Type: application/json" \ -d '{ "model": "claude-3-5-sonnet", "messages": [{"role": "user", "content": "回复 OK 两个字母即可"}], "max_tokens": 10 }'

如果返回里能看到choices字段和模型输出，说明 Key 和通道都没问题。这一步别跳过，后面控制台调不通时，你能快速判断是模型通道的问题还是控制台代码的问题。

3. 控制台后端：FastAPI + WebSocket 实时推送

控制台的核心思路是“旁路监听”，不侵入 Hermes Agent 原有的业务逻辑，只在关键节点向外发射事件。后端用 FastAPI 提供 REST 接口和 WebSocket 推送，前端订阅这些事件来更新界面。

先建项目目录并装依赖：

mkdir hermes-agent-console && cd hermes-agent-console python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install fastapi uvicorn websockets sqlalchemy pydantic

然后定义数据模型。任务和消息是两个核心实体，任务有状态流转，消息挂在任务下面：

# models.py from pydantic import BaseModel from datetime import datetime from typing import Optional, List, Dict, Any class TaskBase(BaseModel): task_id: str status: str # pending / running / success / failed user_input: str created_at: datetime class Task(TaskBase): messages: List[Dict[str, Any]] = [] result: Optional[Dict[str, Any]] = None finished_at: Optional[datetime] = None class MessageBase(BaseModel): role: str # user / assistant / tool content: str timestamp: datetime class Message(MessageBase): task_id: str

接着写 FastAPI 主应用。这里用内存字典存任务，生产环境换成数据库即可。关键点是 WebSocket 广播函数，任何任务状态变化都通过它推给所有已连接的前端：

# main.py from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware from datetime import datetime import asyncio import uuid from typing import Dict app = FastAPI(title="Hermes Agent Console API") app.add_middleware( CORSMiddleware, allow_origins=["*"], # 生产环境请限制为具体域名 allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) active_tasks: Dict[str, Dict] = {} connected_websockets = [] @app.get("/tasks") async def list_tasks(): return list(active_tasks.values()) @app.post("/tasks") async def create_task(user_input: str): task_id = str(uuid.uuid4())[:8] new_task = { "task_id": task_id, "status": "pending", "user_input": user_input, "created_at": datetime.now().isoformat(), "messages": [] } active_tasks[task_id] = new_task asyncio.create_task(_simulate_agent_execution(task_id, user_input)) await _broadcast_task_update(new_task) return new_task @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() connected_websockets.append(websocket) try: while True: data = await websocket.receive_text() # 这里可以处理前端发来的控制指令 except WebSocketDisconnect: connected_websockets.remove(websocket) async def _broadcast_task_update(task_data: Dict): for ws in connected_websockets: try: await ws.send_json({"type": "TASK_UPDATE", "data": task_data}) except Exception: pass async def _simulate_agent_execution(task_id: str, input_text: str): task = active_tasks.get(task_id) if not task: return task["status"] = "running" await _broadcast_task_update(task) await asyncio.sleep(1) task["messages"].append({ "role": "user", "content": input_text, "timestamp": datetime.now().isoformat() }) task["messages"].append({ "role": "assistant", "content": f"正在处理：{input_text}，先分析一下...", "timestamp": datetime.now().isoformat() }) await _broadcast_task_update(task) await asyncio.sleep(2) task["messages"].append({ "role": "assistant", "content": f"关于「{input_text}」的分析已完成，模拟执行成功。", "timestamp": datetime.now().isoformat() }) task["status"] = "success" await _broadcast_task_update(task) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

启动后端：

uvicorn main:app --host 0.0.0.0 --port 8000 --reload

看到Uvicorn running on http://0.0.0.0:8000就说明后端起来了。此时访问http://localhost:8000/tasks应该返回空列表[]。

4. 前端界面：任务列表 + 对话详情 + 实时刷新

前端用一个单 HTML 文件搞定，引入 Vue 3 和 Tailwind CDN，不需要构建工具。核心是三块：左侧任务列表、右侧对话详情、底部输入框。WebSocket 连接建立后，任何任务更新都会自动刷新界面。

<!DOCTYPE html> <html lang="zh-CN"> <head> <meta charset="UTF-8"> <title>Hermes Agent 控制台</title> <script src="https://unpkg.com/vue@3/dist/vue.global.js"></script> <script src="https://cdn.tailwindcss.com"></script> </head> <body class="bg-gray-100"> <div id="app" class="container mx-auto p-6"> <h1 class="text-3xl font-bold mb-6">Hermes Agent 控制台</h1> <div class="grid grid-cols-3 gap-6"> <div class="col-span-1 bg-white rounded-lg shadow p-4"> <h2 class="text-xl font-semibold mb-4">任务列表</h2> <div class="space-y-2"> <div v-for="task in tasks" :key="task.task_id" @click="selectTask(task)" :class="['p-3 rounded cursor-pointer', selectedTaskId === task.task_id ? 'bg-blue-100 border-l-4 border-blue-500' : 'hover:bg-gray-50']"> <div class="flex justify-between"> <span class="font-medium truncate">{{ task.user_input }}</span> <span :class="statusColor(task.status)" class="px-2 py-1 rounded-full text-xs"> {{ task.status }} </span> </div> <div class="text-sm text-gray-500 mt-1">{{ task.created_at }}</div> </div> </div> <button @click="fetchTasks" class="mt-4 w-full bg-gray-200 hover:bg-gray-300 py-2 rounded"> 刷新列表 </button> </div> <div class="col-span-2 bg-white rounded-lg shadow flex flex-col"> <div class="p-4 border-b"> <h2 class="text-xl font-semibold">对话详情</h2> <div v-if="selectedTask" class="text-sm text-gray-600 mt-1"> 任务ID: {{ selectedTask.task_id }} | 状态: <span :class="statusColor(selectedTask.status)"> {{ selectedTask.status }} </span> </div> </div> <div class="flex-1 overflow-y-auto p-4 space-y-4" style="max-height: 500px;"> <div v-for="msg in messages" :key="msg.timestamp" :class="['p-3 rounded-lg max-w-3/4', msg.role === 'user' ? 'bg-blue-100 ml-auto' : 'bg-gray-100']"> <div class="font-semibold capitalize">{{ msg.role }}:</div> <div class="mt-1">{{ msg.content }}</div> <div class="text-xs text-gray-500 mt-2">{{ formatTime(msg.timestamp) }}</div> </div> </div> <div class="p-4 border-t"> <div class="flex space-x-2"> <input v-model="newInput" @keyup.enter="sendMessage" placeholder="输入指令发送给 Hermes Agent..." class="flex-1 border rounded-lg px-4 py-2 focus:outline-none focus:ring-2 focus:ring-blue-500"> <button @click="sendMessage" class="bg-blue-600 hover:bg-blue-700 text-white px-6 py-2 rounded-lg font-semibold"> 发送 </button> </div> </div> </div> </div> </div> <script> const { createApp, ref, onMounted, computed } = Vue; createApp({ setup() { const tasks = ref([]); const selectedTaskId = ref(null); const newInput = ref(''); const socket = ref(null); const selectedTask = computed(() => tasks.value.find(t => t.task_id === selectedTaskId.value) ); const messages = computed(() => selectedTask.value ? selectedTask.value.messages : [] ); const statusColor = (status) => { const map = { pending: 'bg-yellow-100 text-yellow-800', running: 'bg-blue-100 text-blue-800', success: 'bg-green-100 text-green-800', failed: 'bg-red-100 text-red-800' }; return map[status] || 'bg-gray-100 text-gray-800'; }; const formatTime = (iso) => new Date(iso).toLocaleTimeString(); const fetchTasks = async () => { const resp = await fetch('http://localhost:8000/tasks'); tasks.value = await resp.json(); }; const selectTask = (task) => { selectedTaskId.value = task.task_id; }; const sendMessage = async () => { if (!newInput.value.trim()) return; const resp = await fetch('http://localhost:8000/tasks', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ user_input: newInput.value }) }); const newTask = await resp.json(); tasks.value.unshift(newTask); selectTask(newTask); newInput.value = ''; }; const connectWebSocket = () => { const ws = new WebSocket(`ws://${window.location.hostname}:8000/ws`); ws.onmessage = (event) => { const data = JSON.parse(event.data); if (data.type === 'TASK_UPDATE') { const updated = data.data; const idx = tasks.value.findIndex(t => t.task_id === updated.task_id); if (idx > -1) { tasks.value[idx] = updated; } else { tasks.value.unshift(updated); } tasks.value = [...tasks.value]; } }; socket.value = ws; }; onMounted(() => { fetchTasks(); connectWebSocket(); }); return { tasks, selectedTaskId, newInput, selectedTask, messages, statusColor, formatTime, fetchTasks, selectTask, sendMessage }; } }).mount('#app'); </script> </body> </html>

把这段保存成index.html，用浏览器直接打开（或者用python -m http.server 3000起个静态服务）。你会看到左侧任务列表、右侧对话区域和底部输入框。在输入框里发一条消息，后端会创建任务并通过 WebSocket 把状态推回来，界面实时更新。

5. 把控制台接到真实 Hermes Agent 上

上面的_simulate_agent_execution只是模拟。真正要做的，是在 Hermes Agent 执行的关键节点发射事件。假设你的 Hermes Agent 有一个核心的process方法，我们用一个包装类来注入事件发射逻辑：

# hermes_integration.py import asyncio from hermes_agent import HermesAgent # 假设的 Hermes Agent 类 from console_client import emit_event # 向控制台后端发事件的客户端 class InstrumentedHermesAgent(HermesAgent): async def run_with_console(self, task_id: str, user_input: str): try: await emit_event('task_started', task_id, {'input': user_input}) context = {"task_id": task_id} response = await self.process(user_input, context=context) await emit_event('task_completed', task_id, {'result': response}) return response except Exception as e: await emit_event('task_failed', task_id, {'error': str(e)}) raise

然后在main.py的create_task里，把模拟执行换成真实调用：

from hermes_integration import InstrumentedHermesAgent agent = InstrumentedHermesAgent() @app.post("/tasks") async def create_task(user_input: str): task_id = str(uuid.uuid4())[:8] new_task = { "task_id": task_id, "status": "pending", "user_input": user_input, "created_at": datetime.now().isoformat(), "messages": [] } active_tasks[task_id] = new_task asyncio.create_task(agent.run_with_console(task_id, user_input)) await _broadcast_task_update(new_task) return new_task

emit_event的实现很简单，就是往控制台后端的/tasks/{task_id}/events发一个 POST 请求，后端收到后更新任务状态并广播。这样 Hermes Agent 原有的执行逻辑完全不用改，控制台只是“旁路”观察。

如果你希望 Agent 在调用工具时也能实时上报，可以在 Hermes Agent 内部添加消息的地方插入emit_event('new_message', task_id, message_dict)。这样前端就能看到 Agent 的“思考过程”，比如它下一步打算调用哪个工具、为什么这么选。

6. 验证请求与常见报错排查

配置完成后，跑一个端到端验证。先确认后端在跑：

curl http://localhost:8000/tasks

返回[]说明后端正常。然后发一个任务：

curl -X POST "http://localhost:8000/tasks?user_input=帮我总结一下今天的天气"

返回的 JSON 里应该有task_id和status: pending。接着再查一次任务列表，应该能看到状态变成running或success，messages数组里有多条消息。

如果前端页面打开后任务列表一直空，按下面顺序排查：

模型通道相关的报错，优先用 curl 直接打 TaoToken API 验证。如果 curl 能通但 Hermes Agent 不通，问题就在 Agent 的配置读取上，检查config.toml或settings.json里的base_url和api_key是否被正确加载。如果 curl 也不通，就去 TaoToken 控制台确认 Key 状态和余额，或者换一个模型试试。

还有一个容易踩的坑：FastAPI 的create_task接口用了查询参数user_input，前端发的是 JSON body。如果你前端用fetch发 JSON，后端要改成async def create_task(payload: dict)然后从payload["user_input"]取。上面代码里前端发的是 JSON，所以后端需要对应调整，否则会报 422 错误。这个细节在联调时经常卡住人。

7. 下一步：让控制台真正融入日常开发

控制台跑起来之后，你可以按自己的习惯继续扩展。比如在任务列表上加筛选，只看failed状态的任务；或者在对话详情里加一个“重试”按钮，把失败的任务重新提交给 Agent。这些都不需要改后端核心逻辑，前端加个按钮、后端加个接口就行。

如果你打算长期跑 Hermes Agent 做编码或 Agent 类任务，建议把模型调用通道固定到 TaoToken 的 Coding Plan 上，地址是 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。它针对长时间、高频次的编码场景做了优化，配合控制台的任务追踪，能清楚看到每个任务的模型消耗和执行耗时。

日常调试时，我习惯把控制台和后端分开跑：后端用uvicorn main:app --reload常驻，前端用python -m http.server 3000起静态服务。这样改前端不用重启后端，改后端 uvicorn 会自动重载。唯一要注意的是，前端里的localhost:8000要换成后端实际地址，如果后端跑在另一台机器上，记得把localhost改成那台机器的 IP。

最后留一个实用技巧：在_broadcast_task_update里加一行日志，把每次广播的任务 ID 和状态打到终端。这样当前端没刷新时，你能立刻判断是后端没广播，还是前端 WebSocket 断了。排查实时推送问题，这行日志比什么都管用。