文档 03

运行时 · 状态机 · 核心流程

Hermes 三种运行模式的进程模型、Agent 对话循环完整状态机,以及工具调用并发、Gateway 消息路由、上下文压缩等核心流程图。

运行时进程模型
💻 CLI 模式
同步单进程,主线程运行
prompt_toolkit 占据主线程输入循环
AIAgent.run_conversation() 同步调用
工具调用:ThreadPoolExecutor 并发
LLM 流式输出:同步 SSE 读取
会话存储:SQLite WAL 模式
🌐 Gateway 模式
asyncio事件循环驱动
每个平台适配器异步连接
消息 → asyncio.to_thread → AIAgent
每会话独立 AIAgent 实例
threading.Lock 保护 agent 缓存
asyncio + threading 混合并发
🔌 ACP Server
HTTP
HTTP REST 接口暴露 Agent 能力
POST /task → 创建 Agent 任务
GET /task/{id}/stream → SSE 结果流
支持并发多任务(独立 Agent 实例)
可被外部系统程序化调用
CLI 模式并发模型(时间轴示意)
线程
时间轴 →
主线程
用户输入
LLM 请求/流式
解析响应
等待工具完成
LLM 请求/流式
输出结果
工具线程 1
terminal
工具线程 2
file_read × 2
工具线程 3
🔒
ask_human
主线程工作 LLM 调用 工具并发执行 互斥锁等待(交互类工具)
🔄Agent 对话循环状态机
● 开始 / 新消息到达
用户输入 / 平台消息 / API 请求
INIT — 初始化会话
加载记忆 · 构建 system prompt · 注入 skills
检查:iterations < max_iter
AND budget.remaining > 0 ?
✗ 否(超限)
✓ 是
✗ 否(超限)
LLM_CALL — 调用语言模型
transports.send_message(messages, tools) → stream
响应包含 tool_calls ?
✓ 是
TOOL_DISPATCH — 分发工具
互斥检查 → ThreadPoolExecutor 并发执行
TOOL_EXEC — 执行工具函数
tool_fn(args) → result / error
RESULT_APPEND — 追加结果
messages.append(tool_results) · iterations++
↑ 回到 budget 检查
✗ 否(纯文本响应)
COMPRESS_CHECK — 压缩检查
token_count > threshold ?
MEMORY_WRITE — 写入记忆
提取关键信息 → memory_manager.save()
PERSIST — 持久化会话
hermes_state.save_session() → SQLite
● 结束 — 返回响应
● 终止 — 超出预算/迭代上限
返回截断结果 + 警告信息
核心循环流程(run_conversation)
🐍 run_agent.py — AIAgent.run_conversation() 伪代码对应流程
def run_conversation(self, user_message):
  # INIT
  messages = self.prompt_builder.build(user_message) # 注入 skills/memory
  iterations = 0

  # LOOP
  while iterations < self.max_iterations and budget.remaining > 0:
    # LLM CALL
    response = self.transport.send_message(messages, tools) # 流式

    if response.tool_calls:
      # TOOL DISPATCH — 并发执行
      results = execute_tool_calls_concurrent(response.tool_calls)
      messages.append(results)
      iterations += 1
    else:
      # DONE — 最终回复
      self._post_conversation() # 压缩检查 + 记忆写入 + 持久化
      return response.content
🔧工具调用并发执行流程
execute_tool_calls_concurrent() 内部流程
接收 LLM 返回的 tool_calls 列表
[{name: "terminal", args: {...}}, {name: "file_read", ...}, ...]
检查互斥约束
(交互类工具需串行?)
⚠ 含互斥工具(ask_human等)
acquire_lock() 串行执行
等待前一个工具完成再执行
✓ 可并行工具
ThreadPoolExecutor 并发
submit all → wait(all)
汇聚所有结果
tool_fn(args) 实际执行
registry 查找 → 调用工具函数 → 返回 result 或 error
返回 tool_results 列表
格式化为 LLM 可理解的 tool_result messages
工具互斥规则
✓ 可并发执行
file_read · file_write · web_fetch · terminal(独立进程)· git_tool · code_analysis
⚠ 需串行(互斥锁)
ask_human · send_message · browser_tool(共享 CDP 会话)· 写同一文件的操作
🌐Gateway 消息路由流程
gateway/run.py — 多平台消息路由架构
Telegram
TelegramAdapter
Discord
DiscordAdapter
Slack
SlackAdapter
WhatsApp
WhatsAppAdapter
Matrix
MatrixAdapter
+17 platforms
...Adapter
asyncio.to_thread
每消息一个任务
Gateway Router
platform_id + user_id → session_key
Agent Cache
session_key → AIAgent 实例
threading.Lock 保护
AIAgent.run_conversation()
同步执行 · asyncio.to_thread 桥接
platform.send_reply()
通过对应 adapter 回复
📦上下文压缩流程
context_compressor.py — 触发条件与压缩策略
每轮 LLM 调用后
统计 token 使用量
token_count > threshold?
默认 80% context window
触发压缩
LLM 对历史消息做摘要
替换历史段
保留最近 N 轮完整
压缩前(超出阈值)
U第1轮问题~800 tok
A第1轮回答~1200 tok
T工具结果~600 tok
U第2轮问题~400 tok
A第2轮回答~900 tok
U第3轮问题~500 tok
A第3轮回答~1100 tok
U最新问题(保留)~400 tok
LLM
摘要
压缩后(token 释放)
S历史摘要:讨论了X问题,解决方案是Y...~400 tok
U(已压缩)
A(已压缩)
U(已压缩)
A第3轮回答(保留)~1100 tok
U最新问题(保留)~400 tok
📝System Prompt 构建流程
prompt_builder.py — 每次 LLM 调用前动态构建
Base Prompt
角色定义
能力说明
输出格式
Skills
SKILL.md 内容
动态注入
/skill 命令加载
Memory
跨会话记忆摘要
相关历史片段
用户偏好
Context
当前工作目录
环境信息
Token 预算
Jinja2 模板渲染
合并所有组件 → 计算 token 数 → 截断/优先级处理(预算不足时按 Skills > Memory > Context 优先级裁剪)
最终 System Prompt
注入到 messages[0] → 发送给 LLM