上周团队在讨论 Agent 产品的交互方案。前端同学主张用 Web UI + WebSocket，理由是现代、可扩展、支持多端。后端同学说那不如直接嵌到现有产品里，用 HTTP REST API。
我打开了我们自己的 Agent 代码仓库，给他们看了两套实现——一套是纯 Python 的同进程架构，另一套是 React + Python 子进程通过 JSON-RPC 通信。
“我们同一个产品里，两套通信协议并存。”
他们沉默了几秒。“为什么？”
因为这不是架构设计的失误，而是交互范式的必然分化。CLI 要的是零延迟的本地体验，TUI 要的是跨进程的事件驱动架构。用一套协议去套两种场景，只会两头都不讨好。
今天我们就来解剖一下 Hermes Agent 的两种交互架构——看看 CLI 和 TUI 分别是怎么和 Agent Loop 通信的，协议是什么，为什么这么设计。
整体架构：三层模型
在深入细节之前，先建立全局视角。Hermes Agent 的交互层可以抽象为三层：
┌──────────────────────────────────────────────────────────────┐
│ Layer 1: Frontend (UI) │
│ CLI: prompt_toolkit (Python) │
│ TUI: Ink/React (Node.js/TypeScript) │
├──────────────────────────────────────────────────────────────┤
│ Layer 2: Transport │
│ CLI: 同进程函数引用 + queue.Queue │
│ TUI: JSON-RPC 2.0 over stdio (stdin/stdout) │
├──────────────────────────────────────────────────────────────┤
│ Layer 3: AIAgent (run_agent.py) │
│ 核心对话循环：chat() / run_conversation() │
│ 回调注入：callbacks dict → 直接调用或事件推送 │
└──────────────────────────────────────────────────────────────┘
Agent Loop 是同一个——run_agent.py 里的 AIAgent.run_conversation()。它不关心谁在调用它，只关心回调函数怎么注册。两种架构的差异完全在 Layer 1 到 Layer 2。
CLI 架构：同进程直连
CLI 的核心理念是零中间层。用户启动 hermes 命令后，整个过程都在一个 Python 进程内完成。
关键组件
| 组件 | 文件 | 职责 |
|---|---|---|
HermesCLI | cli.py (~10,570 行) | REPL 循环、命令分发、状态管理 |
AIAgent | run_agent.py | 对话循环、工具调用、模型推理 |
| prompt_toolkit | 第三方库 | 终端 UI（输入框、补全、布局） |
Agent 初始化：回调函数直接注入
# cli.py: _init_agent()
self.agent = AIAgent(
model=effective_model,
platform="cli",
session_id=self.session_id,
session_db=self._session_db,
# 直接传递函数引用——不是序列化，不是网络调用
clarify_callback=self._clarify_callback,
reasoning_callback=self._current_reasoning_callback(),
thinking_callback=self._on_thinking,
status_callback=self._on_status,
tool_start_callback=self._on_tool_start,
tool_complete_callback=self._on_tool_complete,
stream_delta_callback=self._stream_delta,
tool_progress_callback=self._on_tool_progress,
)
# generated by hugo AI
这些是普通的 Python 函数对象。Agent Loop 在执行过程中直接调用它们，就像调用 print() 一样自然。没有任何序列化开销，没有任何网络延迟。
主循环：prompt_toolkit 事件循环
CLI 的 run() 方法构建了一个 prompt_toolkit Application，核心是 KeyBindings：
Enter 键 → 路由到正确的队列
├─ sudo/approval/clarify 状态中 → 回答放进 response_queue
├─ Agent 正在运行 → 放入 _interrupt_queue（中断）
└─ Agent 空闲 → 放入 _pending_input（正常输入）
Ctrl+C → 优先级链
├─ 取消语音录制
├─ 取消当前模态提示（sudo/clarify/approval）
├─ Agent 运行中 → interrupt()
└─ Agent 空闲 → 退出
chat() 的线程模型
def chat(self, message, images=None):
# Agent 在后台线程运行
agent_thread = Thread(target=run_agent, daemon=True)
agent_thread.start()
# 主线程监控中断队列
while agent_thread.is_alive():
try:
interrupt_msg = self._interrupt_queue.get(timeout=0.1)
if interrupt_msg:
self.agent.interrupt(interrupt_msg)
break
except queue.Empty:
self._invalidate() # 刷新 prompt_toolkit 渲染
# generated by hugo AI
关键设计：Agent 线程和 UI 主线程是分离的。Agent 线程阻塞在 API 调用或工具执行上，UI 线程继续处理键盘输入。这就是为什么你可以在 Agent 思考时随时打断它。
阻塞式交互：clarify 回调
当 Agent 需要用户澄清问题时，_clarify_callback 被调用：
def _clarify_callback(self, question, choices):
response_queue = queue.Queue()
# 1. 设置 UI 状态 → prompt_toolkit 切换为选择模式
self._clarify_state = {
"question": question,
"choices": choices,
"selected": 0,
"response_queue": response_queue,
}
self._invalidate() # 触发 UI 重绘
# 2. 阻塞等待（在 Agent 线程中）
while True:
try:
result = response_queue.get(timeout=1) # ← 阻塞
return result
except queue.Empty:
# 刷新倒计时显示
if should_refresh():
self._invalidate()
# 3. 超时兜底
return "The user did not respond. Use your best judgement."
# generated by hugo AI
精妙之处：response_queue.get() 阻塞的是 Agent 线程，不是 UI 线程。prompt_toolkit 主事件循环继续运行，用户的 ↑↓ 选择、Enter 确认都正常响应。当用户按 Enter 时，KeyBinding 把答案 put 进 queue，Agent 线程被唤醒返回。
Agent 线程: _clarify_callback() ──→ response_queue.get() [阻塞]
↑
UI 主线程: prompt_toolkit 事件循环 ──→ Enter → response_queue.put(answer)
CLI 回调类型汇总
| 回调 | 触发时机 | 阻塞方式 | 返回类型 |
|---|---|---|---|
clarify_callback | Agent 需要澄清 | queue.get() | str |
thinking_callback | 思考过程流式输出 | 非阻塞 | None |
reasoning_callback | 推理过程输出 | 非阻塞 | None |
tool_progress_callback | 工具执行进度 | 非阻塞 | None |
stream_delta_callback | 模型 token 流式输出 | 非阻塞 | None |
status_callback | 状态变更 | 非阻塞 | None |
TUI 架构：跨进程 JSON-RPC
TUI 走的是完全不同的路线。它是一个 Node.js 进程（React/Ink） spawn 一个 Python 子进程，两者通过 stdio 传递 JSON 行。
关键组件
| 组件 | 文件 | 语言 | 职责 |
|---|---|---|---|
| GatewayClient | ui-tui/src/gatewayClient.ts | TypeScript | 子进程管理、RPC 请求、事件分发 |
| TUI Gateway | tui_gateway/server.py (~2,931 行) | Python | RPC 方法处理、会话管理、Agent 工厂 |
| SlashWorker | tui_gateway/slash_worker.py | Python | 斜杠命令的子进程 |
| AIAgent | run_agent.py | Python | 对话循环（同 CLI） |
通信协议：JSON-RPC 2.0 over stdio
Node.js (stdin) ──JSON──→ Python
Node.js (stdout) ←─JSON── Python
Node.js (stderr) ←─文本── Python (日志)
请求格式（TUI → Python）：
{"jsonrpc": "2.0", "id": "r1", "method": "prompt.submit", "params": {"session_id": "abc", "text": "hello"}}
响应格式（Python → TUI）：
{"jsonrpc": "2.0", "id": "r1", "result": {"status": "streaming"}}
事件推送（Python → TUI，无 id）：
{"jsonrpc": "2.0", "method": "event", "params": {"type": "message.delta", "session_id": "abc", "payload": {"text": "Hel"}}}
GatewayClient 的请求/响应机制
// gatewayClient.ts
request<T>(method: string, params: {}): Promise<T> {
const id = `r${++this.reqId}`;
// 写入 stdin
this.proc.stdin.write(
JSON.stringify({ jsonrpc: '2.0', id, method, params }) + '\n'
);
// 返回 Promise，等待响应
return new Promise((resolve, reject) => {
const timeout = setTimeout(() => reject(new Error('timeout')), 120000);
this.pending.set(id, { resolve: v => { clearTimeout(timeout); resolve(v); }, reject });
});
}
// 收到 stdout 行时分发
private dispatch(msg: Record<string, unknown>) {
const id = msg.id as string | undefined;
const p = id ? this.pending.get(id) : undefined;
if (p) {
// 有 id → RPC 响应 → resolve/reject Promise
this.pending.delete(id);
msg.error ? p.reject(new Error(msg.error.message)) : p.resolve(msg.result);
return;
}
if (msg.method === 'event') {
// method == "event" → 事件推送 → emit 给 React
this.publish(msg.params);
}
}
一句话：TUI 用 Map<id, Promise> 来匹配请求和响应，事件推送则通过 EventEmitter 广播给 UI 组件。
Python 端的防污染设计
Python 进程有一个关键设计——把 stdout 重定向到 stderr，只用 _real_stdout 写 JSON-RPC：
# tui_gateway/server.py
_real_stdout = sys.stdout
sys.stdout = sys.stderr # 防止第三方库 print() 污染协议
def write_json(obj: dict) -> bool:
line = json.dumps(obj, ensure_ascii=False) + "\n"
_real_stdout.write(line)
_real_stdout.flush()
# generated by hugo AI
这意味着所有 print() 调用都会跑到 stderr（日志），只有 write_json() 写的内容才是协议数据。这是一个简单但极其有效的防协议污染策略。
RPC 方法体系
Session 生命周期 (14 个方法):
session.create session.list session.resume
session.close session.title session.undo
session.compress session.branch session.interrupt
session.steer session.usage session.history
session.save terminal.resize
Prompt 提交 (3 个方法):
prompt.submit prompt.background prompt.btw
交互响应 (4 个方法):
clarify.respond approval.respond sudo.respond
secret.respond
配置 (2 个方法):
config.get config.set
其他:
clipboard.paste image.attach input.detect_drop
shell.exec slash.exec slash.complete
skin.get tools.configure mcp.reload
model.options commands.catalog
事件推送体系
消息流:
message.start → message.delta (流式) → message.complete
工具:
tool.start → tool.progress → tool.complete
tool.generating
思考/推理:
thinking.delta → reasoning.delta → reasoning.available
交互:
clarify.request → clarify.respond (用户回答)
approval.request → approval.respond
sudo.request → sudo.respond
secret.request → secret.respond
子代理:
subagent.start → subagent.thinking → subagent.tool → subagent.complete
状态:
session.info status.update error
background.complete btw.complete
gateway.ready gateway.stderr skin.changed
回调桥接：_agent_cbs()
TUI 中 Agent 的回调不是直接函数——它们被包装成 _emit() 调用，把事件推送给前端：
def _agent_cbs(sid: str) -> dict:
return dict(
tool_start_callback=lambda tc_id, name, args:
_emit("tool.start", sid, {"tool_id": tc_id, "name": name}),
tool_complete_callback=lambda tc_id, name, args, result:
_emit("tool.complete", sid, {"tool_id": tc_id, "name": name}),
thinking_callback=lambda text:
_emit("thinking.delta", sid, {"text": text}),
reasoning_callback=lambda text:
_emit("reasoning.delta", sid, {"text": text}),
# 阻塞式交互：发事件 + threading.Event 等待
clarify_callback=lambda q, c:
_block("clarify.request", sid, {"question": q, "choices": c}),
)
# generated by hugo AI
阻塞式交互：_block()
TUI 的 clarify/sudo/secret 回调用 _block() 实现跨进程阻塞：
def _block(event: str, sid: str, payload: dict, timeout: int = 300) -> str:
rid = uuid.uuid4().hex[:8]
ev = threading.Event()
_pending[rid] = ev # 注册 pending
payload["request_id"] = rid
_emit(event, sid, payload) # 推送事件给 TUI
ev.wait(timeout=timeout) # 阻塞等待
_pending.pop(rid, None)
return _answers.pop(rid, "") # TUI 通过 clarify.respond 填入答案
# generated by hugo AI
当 TUI 收到 clarify.request 事件时，它渲染选择界面，用户确认后发送 clarify.respond RPC 请求：
@method("clarify.respond")
def _(rid, params):
r = params.get("request_id", "")
ev = _pending.get(r)
if ev:
_answers[r] = params.get("answer", "")
ev.set() # 唤醒 _block() 中的等待线程
# generated by hugo AI
Python Agent 线程: _block() ──→ _emit("clarify.request") ──→ threading.Event.wait() [阻塞]
↑
Python Gateway: _methods["clarify.respond"] ──→ ev.set() ────┘
↑
Node.js TUI: 收到 event → 渲染 UI → 用户选择 → request("clarify.respond")
两种架构的对比
| 维度 | CLI | TUI |
|---|---|---|
| 进程模型 | 单 Python 进程 | Node.js 进程 + Python 子进程 |
| 通信协议 | 无——直接函数调用 | JSON-RPC 2.0 over stdio |
| UI 框架 | prompt_toolkit (Python) | Ink/React (TypeScript) |
| 回调类型 | 同步函数引用 | 异步事件推送 |
| 阻塞交互 | queue.Queue.get() | threading.Event.wait() + RPC 响应 |
| 流式输出 | sys.stdout.write(delta) | _emit("message.delta", ...) JSON 行 |
| 中断机制 | _interrupt_queue.put() + poll | session.interrupt RPC 方法 |
| 序列化 | 无 | JSON 序列化/反序列化 |
| 容错 | 进程崩溃 = 全部崩溃 | 子进程崩溃不影响 TUI 外壳 |
| 性能 | 零开销 | 每帧 ~0.1ms JSON 序列化 |
为什么两套并存
这不是历史包袱，而是工程取舍：
CLI 选择同进程直连的原因：
- 零延迟——没有序列化、没有 IPC 开销，工具执行和 UI 更新在同一进程内
- 简单——不需要处理进程生命周期、协议版本兼容、事件丢失
- 线程安全——prompt_toolkit 和 queue.Queue 的组合是经过充分验证的模式
TUI 选择跨进程 RPC 的原因：
- 技术栈自由——React 生态的组件、hooks、状态管理远比 Python TUI 丰富
- 事件驱动——JSON-RPC 的事件推送天然适配 React 的状态更新模型
- 进程隔离——Python 进程崩溃不影响 TUI 外壳，可以自动重启
- 可扩展性——同样的协议可以被其他前端复用（未来可能有 Electron、Web 前端）
协议设计的三个关键原则
1. 单一写入者原则
# TUI Gateway: 只有 write_json() 写 _real_stdout
sys.stdout = sys.stderr # 所有 print() 被重定向
# generated by hugo AI
任何多路复用的通信通道都必须只有一个写入者。TUI 把 Python 的默认 stdout 重定向到 stderr，确保 print() 永远不会污染 JSON-RPC 协议。这是很多初学者踩坑的地方——一个第三方库的 debug print 就能让整条通信链路崩溃。
2. 请求-事件分离原则
有 id 的 JSON-RPC → 请求/响应（同步语义）
无 id 的 JSON-RPC → 事件推送（异步语义）
JSON-RPC 2.0 规范中，Notification（无 id）和 Request（有 id）是不同的消息类型。TUI 利用这个区分来分离两种通信语义：用户操作（发 prompt、切换模型）用 Request/Response，Agent 主动推送（流式输出、工具执行）用 Event。
3. 回调注入原则
# CLI
agent = AIAgent(..., clarify_callback=self._clarify_callback, ...)
# TUI
agent = AIAgent(..., **_agent_cbs(sid)) # _agent_cbs 返回回调 dict
# generated by hugo AI
Agent Loop 不依赖任何具体的 UI 框架。它只接受一组回调函数，调用方决定这些回调是 queue.Queue.put() 还是 _emit("event", ...)。这就是为什么同一个 AIAgent 能同时服务于 CLI、TUI、Telegram、Discord——回调注入是解耦的关键。
架构图
CLI 数据流:
用户输入 (Enter)
│
▼
prompt_toolkit KeyBindings ──→ _pending_input / _interrupt_queue
│ │
│ chat() 监控 queue │
│ │ │
│ ▼ │
│ AIAgent.run_conversation() │
│ │ │
│ 回调函数直接调用 ←─────────────────┘
│ ├─ _stream_delta() → sys.stdout.write()
│ ├─ _clarify_callback() → queue.get() ← 用户 Enter
│ ├─ _on_tool_progress() → _cprint()
│ └─ _on_thinking() → Rich 渲染
│
▼
终端显示
TUI 数据流:
用户输入 (Enter)
│
▼
React Ink 组件 ──→ GatewayClient.request("prompt.submit", {text})
│ │
│ JSON 写入 stdin
│ │
│ tui_gateway/server.py
│ handle_request()
│ │
│ AIAgent.run_conversation()
│ │
│ 回调函数 → _emit() → JSON 写入 stdout
│ ├─ "message.delta" → TUI 流式渲染
│ ├─ "tool.start" → TUI 工具状态更新
│ ├─ "clarify.request" → TUI 渲染选择界面
│ └─ "status.update" → TUI 状态栏更新
│
▼
React 状态更新 → UI 重新渲染
总结
CLI 和 TUI 代表了两种不同的工程哲学：
- CLI 是"嵌入式"思维——UI 和 Agent 在同一进程内，通信成本为零，代价是 UI 技术栈受限于 Python 生态
- TUI 是"微服务"思维——UI 和 Agent 是独立进程，通过标准协议通信，代价是序列化开销和进程管理复杂度
两者并存不是冗余，而是场景适配。CLI 适合 SSH 远程、低资源、极客场景；TUI 适合本地开发、富 UI 交互、现代终端体验。
你在实际项目中会选哪种架构？或者两者都用？欢迎留言讨论。