关键词：Hermes Agent、AIAgent、源码分析、tool calling、对话循环、Python
文章目录
- 一、背景：run_agent的介绍
- 二、入口执行链：从命令到对话循环
- 三、分层架构：五层职责拆解
- 第 1 层:门面层(`run_agent.py`)
- 第 2 层:初始化层(`agent/agent_init.py`)
- 第 3 层:每轮前置(`agent/turn_context.py`)
- 第 4 层:对话主循环(`agent/conversation_loop.py`)
- 第 5 层:工具执行 + 收尾(`agent/tool_executor.py` + `agent/turn_finalizer.py`)
- 四、核心机制:四个值得抄的工程实践
- 机制 1:双层预算的主循环
- 机制 2:段式工具调度(并行安全分段)
- 机制 3:合成消息标记(synthetic scaffolding)
- 机制 4:错误分类,避免白烧重试预算
- 五、真实旅程复盘:一条消息的完整生命周期
- 六、总结:设计取舍与"最值得抄的三件事"
- 设计取舍对照表
- 边界与代价(诚实说明)
- 最值得抄的三件事
一、背景：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 | 回合收尾与结果组装 |
这个文件要解决的工程难点,恰恰是它行数膨胀的原因:
- 门面与实现分离:上层有 CLI、gateway(Telegram/Discord/飞书)、desktop 多个入口,它们要共享同一个对话引擎。
run_agent.py提供统一 API,实现放agent/包,改行为不动接口。 - 向后兼容的符号重导出:历史上有约 28 个测试文件用
mock.patch("run_agent.OpenAI")这种写法,大量生产代码from run_agent import X。重构时实现挪走了,这些符号必须留在run_agent模块命名空间里,于是满屏# noqa: F401 # re-exported for tests。 - 既是库又是 CLI:作为库被 import 时不能拖慢启动、不能因缺依赖崩掉;作为 CLI 直接跑时又要能列工具、选工具集。这两个诉求在同一文件里被小心翼翼地平衡。
下面我沿着"用户输入一条消息 → 最终输出"这条链路,把它的实现逻辑和关键代码讲清楚。
二、入口执行链：从命令到对话循环
一条消息进入 Hermes 的完整链路是这样的(以 CLI 为例):
关键点在这一步:run_agent.py里的AIAgent.run_conversation只是一个forwarder,真正的实现被转发到了agent/conversation_loop.py:
# run_agent.py:7798defrun_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``."""fromagentimportrelay_runtimefromagent.conversation_loopimportrun_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. 真正调对话引擎withbind_subagent_parent(self),scoped_runtime_main({}):result=run_conversation(self,user_message,...)returnresult__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:343defbuild_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/modelset_runtime_main(...)把"每轮要做的杂事"收拢成一个TurnContext返回,主循环只读结果,这是把前置逻辑和循环逻辑解耦的关键手法。
第 4 层:对话主循环(agent/conversation_loop.py)
核心,下面第四节单独展开。
第 5 层:工具执行 + 收尾(agent/tool_executor.py+agent/turn_finalizer.py)
工具执行有并发、串行、分段三种执行器;finalize_turn(第 7500 行调用)负责把循环结果组装成标准 dict:
returnfinalize_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_iterationsandagent.iteration_budget.remaining>0)oragent._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.namefortcinassistant_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=[]# (路径, 是否写) 的占位表fortool_callintool_calls:tool_name=tool_call.function.name# 交互式工具 → 顺序屏障iftool_namein_NEVER_PARALLEL_TOOLS:_add_sequential(tool_call);continue# 参数解析失败 → 顺序屏障try:function_args=json.loads(tool_call.function.arguments)exceptException:_add_sequential(tool_call);continue# 路径域工具:读读可并行,读写/写写冲突则关闭当前并行段iftool_namein_PATH_SCOPED_TOOLS:...ifany((is_writerorexisting_is_writer)and_paths_overlap(...)):_close_parallel()# 冲突,当前段结束reserved_paths.extend(...)current.append(tool_call);continue# 其余并行安全工具或 opt-in 的 MCP 工具 → 并行iftool_namein_PARALLEL_SAFE_TOOLSor_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:returnisinstance(msg,dict)andany(msg.get(flag)forflagin_EPHEMERAL_SCAFFOLDING_FLAGS)标记用_前缀不是随手写的,注释里讲得很清楚:wire sanitizer 会在请求离开进程前剥掉所有顶层_前缀 key,所以这些内部字段永远不会泄漏到严格的 OpenAI 兼容网关。
这是开发 AIAgent 最有价值的一条实践:凡是"驱动内部状态机的消息"和"真实的对话历史",必须用显式标记区分,并且持久化层要能识别并剔除前者。不做这一步,你的 agent 一旦支持 resume,就会出诡异的"重放污染"bug。
机制 4:错误分类,避免白烧重试预算
主循环外层有个巨大的except,但它不是无脑重试,而是先分类(第 7405 行):
exceptExceptionase:# 通过 traceback 模块名判断:本地处理错误 vs API 错误tb_module_names=set()_tb=e.__traceback__while_tbisnotNone: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_localandnot_hit_api# 本地 bug 是确定性的,重试必失败 → 立即停,不烧预算if_is_local_processing_errororapi_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|1261520260823_103334_372047|cli|13|5|27428|126220260823_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 工具执行echotool-ok 并把输出原样告诉我 assistant||tool|terminal|{"output":"tool-ok","exit_code":0,"error":null}assistant||tool-ok这个 role 序列user → assistant(空) → tool → assistant正好对应主循环的完整轮次:
user消息进入,build_turn_context完成前置;- 第一次 API 调用,模型返回
finish_reason=tool_calls,assistant 消息内容为空、带tool_calls(这就是为什么第一条 assistant 内容为空); - 主循环走工具执行分支,调
terminal工具,结果以role=tool追加进 messages; - 循环回到顶部,
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 实测),版本更新后行号会漂移,引用前先重新定位。
最值得抄的三件事
- 门面 + forwarder 的架构:让 CLI/gateway/desktop 共享同一对话引擎,同时保住历史测试的 patch 写法。
- synthetic 标记机制:任何"驱动内部重试的假消息"都必须显式标记并在持久化前剔除,否则 resume 会话会中毒——这是 Agent 项目里最容易忽视、后果最隐蔽的坑。
- 段式工具调度:用路径占位表和读写角色判定并行安全,而不是无脑全并行或全串行,这是工具调用延迟优化的正确姿势。
