关键词：Hermes Agent、AIAgent、源码分析、tool calling、对话循环、Python

一、背景：run_agent的介绍

Hermes Agent 是 Nous Research 开源的模型无关 AI Agent 框架。它的核心运行时入口是一个叫 run_agent.py 的文件——单文件 。我第一次打开它的时候以为核心逻辑都在里面,读完才发现被"骗"了:这是一个典型的门面(Facade)+ 兼容层,真正干活的逻辑几乎全被拆进了 agent/ 包

先看几个真实的规模数据(本机实测,Hermes v0.20.0):

文件	行数	职责
run_agent.py	8206	门面 + 兼容层 + CLI 入口
agent/conversation_loop.py	7524	对话主循环(真正实现)
agent/agent_init.py	2823	初始化(模型路由/凭据/工具装配)
agent/tool_executor.py	2403	工具执行(并发/串行/分段)
agent/turn_context.py	1281	每轮前置处理
agent/tool_dispatch_helpers.py	732	段式工具调度规划
agent/turn_finalizer.py	772	回合收尾与结果组装

这个文件要解决的工程难点,恰恰是它行数膨胀的原因:

1. 门面与实现分离:上层有 CLI、gateway(Telegram/Discord/飞书)、desktop 多个入口,它们要共享同一个对话引擎。run_agent.py 提供统一 API,实现放 agent/ 包,改行为不动接口。
2. 向后兼容的符号重导出:历史上有约 28 个测试文件用 mock.patch("run_agent.OpenAI") 这种写法,大量生产代码 from run_agent import X。重构时实现挪走了,这些符号必须留在 run_agent 模块命名空间里,于是满屏 # noqa: F401 # re-exported for tests。
3. 既是库又是 CLI:作为库被 import 时不能拖慢启动、不能因缺依赖崩掉;作为 CLI 直接跑时又要能列工具、选工具集。这两个诉求在同一文件里被小心翼翼地平衡。

下面我沿着"用户输入一条消息 → 最终输出"这条链路,把它的实现逻辑和关键代码讲清楚。

二、入口执行链：从命令到对话循环

一条消息进入 Hermes 的完整链路是这样的(以 CLI 为例):

关键点在这一步:run_agent.py 里的 AIAgent.run_conversation 只是一个 forwarder,真正的实现被转发到了 agent/conversation_loop.py:

# run_agent.py:7798
def run_conversation(
    self,
    user_message: Any,
    system_message: str = None,
    conversation_history: List[Dict[str, Any]] = None,
    task_id: str = None,
    stream_callback: Optional[callable] = None,
    ...
) -> Dict[str, Any]:
    """Forwarder — see ``agent.conversation_loop.run_conversation``."""
    from agent import relay_runtime
    from agent.conversation_loop import run_conversation
    ...
    # 1. 会话协调器加锁,防止同一 session 并发跑
    relay_lease = relay_runtime.SESSION_COORDINATOR.acquire_conversation(...)
    relay_turn = relay_runtime.SESSION_COORDINATOR.begin_turn(...)
    # 2. 发布 portal 标签 + 记账上下文
    token = set_conversation_context(self._conversation_root_id())
    acct_token = set_accounting_context(...)
    # 3. 真正调对话引擎
    with bind_subagent_parent(self), scoped_runtime_main({}):
        result = run_conversation(self, user_message, ...)
    return result

__init__ 同理,60 多个参数原样透传给 agent/agent_init.py 的 init_agent。这就是"门面模式"在大型 Agent 项目里的真实用法:接口定义和契约留在门面层,行为实现放到可独立测试的模块里,所有入口(CLI/gateway/desktop)复用同一套对话引擎。

三、分层架构：五层职责拆解

按职责可以把整条链路切成五层:

第 1 层:门面层(run_agent.py)

对外暴露 AIAgent 类,三个身份:

• 库入口:from run_agent import AIAgent
• 符号重导出锚点:让 mock.patch("run_agent.X") 的测试不崩
• CLI 入口:文件末尾 fire.Fire(main) 支持 python run_agent.py --query=...

有一个很讲究的设计:fire 只在 __main__ 块里 import,不在模块顶部:

# run_agent.py 顶部注释
# NOTE: `fire` is ONLY used in the `__main__` block below ...
# It is imported there, not here, so that importing run_agent from a
# daemon thread (e.g. curator's forked review agent) never fails with
# ModuleNotFoundError on broken/partial installs where `fire` isn't present.

这是"库/CLI 双身份"的经典取舍:import 成本高的、可能缺失的依赖,延迟到真正需要时才加载。

第 2 层:初始化层(agent/agent_init.py)

init_agent(第 459 行)负责把 60+ 个参数变成一份可运行的 agent 状态:模型路由、provider 识别、凭据池、工具集装配、客户端构建。注意它签名里 max_iterations: int = 90——这是工具调用迭代的硬上限,后面会看到它和软预算的配合。

第 3 层:每轮前置(agent/turn_context.py)

build_turn_context(第 343 行)是"每轮一次"的 prologue,全部集中在一个函数里,避免污染主循环:

# agent/turn_context.py:343
def build_turn_context(agent, user_message, system_message, ...) -> TurnContext:
    # 1. 防 broken pipe 的 stdio 守卫(daemon/headless 场景)
    install_safe_stdio()
    # 2. 恢复因压缩轮换的会话
    recovered_history = recover_rotated_compression_session(agent)
    # 3. 给本线程日志打 session id 标签
    set_session_context(agent.session_id)
    # 4. 恢复主运行时(上一轮可能激活了 fallback)
    agent._restore_primary_runtime()
    # 5. 通知 auxiliary_client 当前生效的 provider/model
    set_runtime_main(...)

把"每轮要做的杂事"收拢成一个 TurnContext 返回,主循环只读结果,这是把前置逻辑和循环逻辑解耦的关键手法。

第 4 层:对话主循环(agent/conversation_loop.py)

核心,下面第四节单独展开。

第 5 层:工具执行 + 收尾(agent/tool_executor.py + agent/turn_finalizer.py)

工具执行有并发、串行、分段三种执行器;finalize_turn(第 7500 行调用)负责把循环结果组装成标准 dict:

return finalize_turn(
    agent,
    final_response=final_response,
    api_call_count=api_call_count,
    interrupted=interrupted,
    failed=failed,
    messages=messages,
    ...
)

四、核心机制:四个值得抄的工程实践

机制 1:双层预算的主循环

这是整个对话引擎的心脏,在 conversation_loop.py 第 1540 行:

while (api_call_count < agent.max_iterations and agent.iteration_budget.remaining > 0) or agent._budget_grace_call:
    # 1. drain redirect:用户 /redirect 纠正方向
    # 2. 检查 interrupt_requested:用户发新消息打断
    # 3. 消费 iteration budget
    # 4. 构建请求 → API 调用(内层还有 retry 循环)
    # 5. 按 finish_reason 分叉:stop / tool_calls / length / content_filter
    # 6. tool_calls → 工具执行 → 追加结果 → continue
    # 7. stop → 最终回复 → break

两层循环要理解清楚:

• 外层 while 管 tool-calling 迭代次数,由 max_iterations(硬上限)和 iteration_budget(软预算)双重兜底;
• 内层 while retry_count < max_retries(第 2305 行)管单次 API 调用的重试,处理限流、fallback、凭据刷新、指数退避。

还有一个我很喜欢的细节:预算可以退。如果这一轮模型只调了 execute_code 这种 RPC 式的便宜调用,就把预算退回去:

# conversation_loop.py:6590 附近
_tc_names = {tc.function.name for tc in assistant_message.tool_calls}
if _tc_names == {"execute_code"}:
    agent.iteration_budget.refund()

机制 2:段式工具调度(并行安全分段)

run_agent.py 的 _execute_tool_calls(第 7632 行)是段式调度的入口,核心规划逻辑在 tool_dispatch_helpers.py 的 _plan_tool_batch_segments(第 116 行):

def _plan_tool_batch_segments(tool_calls, *, execution_cwd=None):
    """把一批工具调用切成有序的 (kind, calls) 段。"""
    segments = []
    reserved_paths = []  # (路径, 是否写) 的占位表
    for tool_call in tool_calls:
        tool_name = tool_call.function.name
        # 交互式工具 → 顺序屏障
        if tool_name in _NEVER_PARALLEL_TOOLS:
            _add_sequential(tool_call); continue
        # 参数解析失败 → 顺序屏障
        try:
            function_args = json.loads(tool_call.function.arguments)
        except Exception:
            _add_sequential(tool_call); continue
        # 路径域工具:读读可并行,读写/写写冲突则关闭当前并行段
        if tool_name in _PATH_SCOPED_TOOLS:
            ...
            if any((is_writer or existing_is_writer) and _paths_overlap(...)):
                _close_parallel()  # 冲突,当前段结束
            reserved_paths.extend(...)
            current.append(tool_call); continue
        # 其余并行安全工具或 opt-in 的 MCP 工具 → 并行
        if tool_name in _PARALLEL_SAFE_TOOLS or _is_mcp_tool_parallel_safe(tool_name):
            current.append(tool_call); continue
        _add_sequential(tool_call)

这个设计的精妙之处在于用"路径占位表 + 读写角色"来判定并行安全:

• read_file 读同一个文件、两个 search_files 读同一子树,是 reader↔reader,可以并行(读操作可交换);
• 只要涉及 writer(写文件、patch),并且目标路径和已占位的路径重叠,就关闭当前并行段,让冲突的调用排到前一段执行完之后。

这样既保住了"模型原始调用顺序"和"副作用边界",又能在安全子集里并发,把延迟打下来。对比那种 all-or-nothing 的"要么全并行、要么全串行"的粗粒度方案,这是一个明显的工程升级。

机制 3:合成消息标记(synthetic scaffolding)

这是整个代码库里最容易踩坑、也最见功底的地方。主循环里为了驱动内部重试,会往 messages 里注入一些"假消息"——空响应恢复、验证 nudge、kanban 收尾 nudge、dropped tool-call nudge。这些消息只用于驱动下一轮 API 调用,绝不能写进持久化 transcript,否则 resume 会话时会把这些内部指令当作用户上下文重放,污染会话。

解决办法是给每条假消息打标记,持久化层见到标记就剥掉:

# run_agent.py:234
_EPHEMERAL_SCAFFOLDING_FLAGS = (
    "_empty_recovery_synthetic",
    "_empty_terminal_sentinel",
    "_thinking_prefill",
    "_verification_stop_synthetic",
    "_pre_verify_synthetic",
    "_kanban_stop_synthetic",
    "_dropped_toolcall_nudge",
)

def _is_ephemeral_scaffolding(msg: Any) -> bool:
    return isinstance(msg, dict) and any(
        msg.get(flag) for flag in _EPHEMERAL_SCAFFOLDING_FLAGS
    )

标记用 _ 前缀不是随手写的,注释里讲得很清楚:wire sanitizer 会在请求离开进程前剥掉所有顶层 _ 前缀 key,所以这些内部字段永远不会泄漏到严格的 OpenAI 兼容网关。

这是开发 AIAgent 最有价值的一条实践:凡是"驱动内部状态机的消息"和"真实的对话历史",必须用显式标记区分,并且持久化层要能识别并剔除前者。 不做这一步,你的 agent 一旦支持 resume,就会出诡异的"重放污染"bug。

机制 4:错误分类,避免白烧重试预算

主循环外层有个巨大的 except,但它不是无脑重试,而是先分类(第 7405 行):

except Exception as e:
    # 通过 traceback 模块名判断:本地处理错误 vs API 错误
    tb_module_names = set()
    _tb = e.__traceback__
    while _tb is not None:
        tb_module_names.add(os.path.splitext(
            os.path.basename(_tb.tb_frame.f_code.co_filename))[0])
        _tb = _tb.tb_next
    _hit_local = bool(tb_module_names & _LOCAL_PROCESSING_MODULES)
    _hit_api = bool(tb_module_names & _API_CALL_MODULES)
    _is_local_processing_error = _hit_local and not _hit_api
    # 本地 bug 是确定性的,重试必失败 → 立即停,不烧预算
    if _is_local_processing_error or api_call_count >= agent.max_iterations - 1:
        ...
        break

判断逻辑是:看异常 traceback 有没有穿过已知的"本地后处理模块"(比如把多模态 content 塞进正则导致的 bug),如果没穿过任何 API 调用模块,那几乎可以断定是本地 bug——重试必败,直接停。这个区分把"上游 API 抖动(值得重试)"和"自己的确定性 bug(不值得重试)"分开,省下宝贵的迭代预算。

五、真实旅程复盘:一条消息的完整生命周期

光读代码不够,我用 sqlite3 查了本机真实的状态库,拿一条带工具调用的会话验证主循环的轮次。先看会话统计:

$ sqlite3 ~/.hermes/state.db "SELECT id, source, message_count, tool_call_count, input_tokens, output_tokens FROM sessions WHERE tool_call_count > 0 ORDER BY id DESC LIMIT 3;"

20260823_103542_f20b5f|cli|50|29|69465|12615
20260823_103334_372047|cli|13|5|27428|1262
20260823_094705_523adb|cli|4|1|140|70

取最后一条(最干净,只有 1 次工具调用、4 条消息)看 role 序列:

$ sqlite3 ~/.hermes/state.db "SELECT role, tool_name, substr(replace(content, char(10),' '), 1, 60) FROM messages WHERE session_id='20260823_094705_523adb' ORDER BY id;"

user||用 terminal 工具执行 echo tool-ok 并把输出原样告诉我
assistant||
tool|terminal|{"output": "tool-ok", "exit_code": 0, "error": null}
assistant||tool-ok

这个 role 序列 user → assistant(空) → tool → assistant 正好对应主循环的完整轮次:

1. user 消息进入,build_turn_context 完成前置;
2. 第一次 API 调用,模型返回 finish_reason=tool_calls,assistant 消息内容为空、带 tool_calls(这就是为什么第一条 assistant 内容为空);
3. 主循环走工具执行分支,调 terminal 工具,结果以 role=tool 追加进 messages;
4. 循环回到顶部,continue 发起第二次 API 调用,这次模型拿到工具结果,返回 finish_reason=stop,输出最终回复 tool-ok,循环 break。

持久化证据和源码读到的循环轮次一一对应,这就是"真实输出铁律"在源码分析里的落地:不只贴 stdout,还贴数据库里的 ground truth。

六、总结:设计取舍与"最值得抄的三件事"

设计取舍对照表

设计点	Hermes 的做法	通用启示
多入口共享引擎	门面 + 实现分离,forwarder 转发到 agent/ 包	接口定义与行为实现分层,改行为不动契约
死循环防护	max_iterations 硬上限 + iteration_budget 软预算	双层预算,软预算还能按调用成本退款
工具并行安全	路径占位表 + 读写角色判定分段	用"资源占用 + 读写语义"判定并发安全,优于粗粒度开关
内部状态机消息	_ 前缀 synthetic 标记,持久化层剔除	显式区分"驱动重试的假消息"和"真实历史"
重试策略	traceback 模块名分类本地 bug vs 上游错误	确定性错误不重试,只重试值得重试的
库/CLI 双身份	重依赖延迟到 __main__ 才 import	延迟加载,守护线程 import 不崩

边界与代价(诚实说明)

• 门面 + 重导出层是有代价的:8206 行里大量是的兼容代码,新人容易误以为逻辑在这里,实际改这里多半不生效或只影响一个调用路径。要改行为,必须定位到 agent/ 包的实现。
• 段式调度只优化了"工具执行"这一段的并发,API 调用本身(单请求)仍然是串行的,模型生成的延迟没有并行化的空间。
• 本文行号基于 Hermes v0.20.0(2026-08 实测),版本更新后行号会漂移,引用前先重新定位。

最值得抄的三件事

1. 门面 + forwarder 的架构:让 CLI/gateway/desktop 共享同一对话引擎,同时保住历史测试的 patch 写法。
2. synthetic 标记机制:任何"驱动内部重试的假消息"都必须显式标记并在持久化前剔除,否则 resume 会话会中毒——这是 Agent 项目里最容易忽视、后果最隐蔽的坑。
3. 段式工具调度:用路径占位表和读写角色判定并行安全,而不是无脑全并行或全串行,这是工具调用延迟优化的正确姿势。