我读源码有个习惯:拿到一个不熟悉的项目,先不看 README 里吹的功能,而是先找 main,再顺着 main 把主循环画出来。Hermes 这个项目,我第一次在终端里执行启动命令,它立刻打出一句问候,然后静默等待输入。当时我脑子里跳出来的问题就是:从敲下命令到出现这句问候,中间到底发生了什么?
这篇就把这条链路完整拆开。Hermes 是一个跑在终端里的对话助手,它的进程生命周期没有 Web 服务那么多层:一个入口函数、一段装配逻辑、一个 REPL 主循环,然后退出。这种结构非常适合做源码阅读的切入点。这篇文章是源码解析系列的第 1 篇,重点放在入口与主循环,适合已经能跑起 Hermes、但想知道它内部怎么转的人,也适合想练习源码阅读思路的人。
先说清楚一件事:下面贴的代码是按主题裁剪过的,省略了平台差异、命令补全这类无关分支,但调用链和状态流转的顺序保留了源码里的真实逻辑。
1. 先看项目地图:Hermes 的入口到底藏在哪里
很多人读源码会犯一个错误:打开仓库直接点进最大的那个文件,然后开始看业务逻辑。结果看半天不知道这段代码什么时候被调用。正确顺序是先解决“这个进程从哪里启动”的问题。
Hermes 的入口实际上有两层。第一层是安装工具生成的命令行入口,第二层是包内部的__main__.py。两者最终都会落到hermes/cli.py的main()函数上。
1.1 从安装入口到__main__.py
在项目配置文件里会声明一个 console script:
[project.scripts] hermes = "hermes.cli:main"这句话的意思是:安装完项目之后,shell 里的hermes命令会直接调用hermes.cli模块的main函数。用 Python 标准库包结构来表示,__main__.py长这样:
from hermes.cli import main if __name__ == "__main__": raise SystemExit(main())这里有一个很多人会忽略的细节:为什么用raise SystemExit(main()),而不是直接main()?
main()返回的值会被SystemExit接收,解释器拿到这个值之后,会把空值或None当作 0 退出,把数字2当作错误码退出。也就是说,只要main()内部通过return 0或return 1表达执行结果,命令行层面就能正确拿到进程退出码。这是一个非常轻量、可靠的退出码传递方式。
1.2 源码目录的分工
Hermes 的源码目录不算大,第一眼需要关注的文件大概有这些:
| 文件 | 职责 |
|---|---|
cli.py | 参数解析、装配、REPL 主循环 |
config.py | 配置读取、默认值合并、配置错误 |
engine.py | Agent 对象、模型客户端、对话生成 |
session.py | 会话历史维护、消息数量裁剪 |
commands.py | /clear、/exit这类斜杠命令处理 |
从这个表能看到,Hermes 把“用户交互循环”和“对话生成逻辑”分开了。cli.py只负责进程生命周期,engine.py只负责一轮对话的模型调用。这个边界值得记住,后面读主循环的时候你就知道哪些代码该放哪一层。
2. 命令行参数解析:把用户的意图交给参数对象
Hermes 的main()第一步不是加载配置,也不是创建 Agent,而是解析命令行参数。这一步看似简单,但决定了后面所有装配逻辑的分支。
2.1 参数列表为什么只保留三件事
参数解析代码大概是这样的:
import argparse def build_parser(): parser = argparse.ArgumentParser( prog="hermes", description="Hermes 终端对话助手", ) parser.add_argument("-c", "--config", help="指定配置文件路径") parser.add_argument("-m", "--model", help="覆盖配置里的模型名") parser.add_argument("-v", "--verbose", action="store_true", help="输出调试日志") parser.add_argument("-q", "--quiet", action="store_true", help="只输出对话内容") return parser你可能会问:一个对话助手为什么参数这么少?会话文件、温度、系统提示词不都应该暴露成参数吗?
从设计上看,这些属于“高频使用”还是“低频配置”的问题。用户在终端里每次启动都想去调整的东西只有模型、配置路径和日志级别。像系统提示词、温度这类内容,写死在配置里比塞进命令行参数更合适。参数一旦变多,用户每次敲命令的成本会急剧上升,而且字符串参数很容易被 shell 转义规则坑到。所以 Hermes 选择把高频操作留给命令行,把低频设置留给配置文件。
2.2 配置路径的三级优先顺序
接下来要解决“读取哪个配置文件”。Hermes 的逻辑是三级优先:
import os from pathlib import Path def resolve_config_path(explicit: str | None) -> Path: if explicit: return Path(explicit).expanduser() env_path = os.getenv("HERMES_CONFIG") if env_path: return Path(env_path).expanduser() return Path.home() / ".config" / "hermes" / "config.json"这个顺序是命令行参数优先,其次环境变量,最后才是当前用户目录下的默认配置。为什么要这样?
命令行参数是临时的、只对本次启动生效,优先级必须最高;环境变量适合在 CI、容器、不同项目脚本里批量指定,不需要每个脚本都加-c;默认路径则保证用户什么都不传时也能跑起来。三级顺序也符合大多数命令行工具的用户预期。
实际读配置的时候,还需要处理“文件不存在”和“JSON 解析失败”两种情况:
import json class ConfigError(Exception): pass def load_config(path: Path) -> dict: if not path.exists(): write_default_config(path) try: raw = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: raise ConfigError(f"配置文件不是合法 JSON: {exc}") from exc defaults = { "model": "hermes-local", "temperature": 0.7, "verbose": False, "history_size": 20, } return {**defaults, **raw}这里有个很实际的处理:配置文件不存在时直接生成一个默认文件。第一次跑命令的人不会看到一个 Red Error,而是会自动得到一个可修改的默认配置。我见过不少工具在配置缺失时直接抛异常,这对新手特别不友好。Hermes 选择“先落一份默认配置再继续”,显然是考虑到了终端用户的体验。
2.3 参数解析失败时应该发生什么
argparse解析失败时会自己打印 usage,并且调用sys.exit(2),所以main()里通常不需要针对“参数不合法”再写分支。但这带来一个测试上的小问题:如果你在测试里直接调用main(["--not-exist"]),进程会直接退出,测试也没法继续。
解决方法是把argv作为main()的参数暴露出来:
def main(argv=None): parser = build_parser() args = parser.parse_args(argv) ...argparse的parse_args在参数为None时默认读取sys.argv[1:],所以命令行启动不受影响;但测试时你可以显式传入一个字符串列表,从而模拟不同启动参数。这就是为什么很多可测试的 CLI 项目都会写成main(argv=None)而不是直接访问sys.argv。
3. main 的装配顺序:config、日志与 Agent 谁先谁后
命令行参数解析完之后,真正的装配才开始。main()里那几行代码的顺序很关键:
def main(argv=None): args = build_parser().parse_args(argv) config_path = resolve_config_path(args.config) config = load_config(config_path) config["verbose"] = args.verbose or config["verbose"] setup_logging(verbose=config["verbose"]) agent = Agent.from_config( config, model=args.model or config["model"], ) return run_chat_loop(agent, config)顺序是:配置加载、日志初始化、Agent 创建、进入主循环。这个顺序不是随手写的,每一步都有依赖关系。
3.1 日志必须在 Agent 创建前准备好
有人会在项目里先创建 Agent 再配置日志,结果 Agent 初始化时想打的日志全部丢失。Hermes 把setup_logging()放在 Agent 之前,就是因为 Agent 的构造函数里会记录当前加载的模型名、配置路径这类启动信息。
日志配置本身也要从config["verbose"]读取,所以日志必须在 config 加载之后。完整的依赖链是:命令行参数 → 配置 → 日志 → Agent。顺序反了,要么读不到日志,要么日志级别不对。
这里补充一个实战经验:日志级别不要只做成布尔开关。更好的方式是支持--log-level,但 Hermes 只保留--verbose和--quiet两个布尔参数,底层映射到 Python 的logging.DEBUG和logging.WARNING。对小项目来说,布尔开关已经够用,没必要为“高度可配置”付出额外复杂度。
3.2 Agent 工厂方法负责创建模型客户端
Agent.from_config这个工厂方法承担了“把配置变成可用对象”的职责:
class Agent: def __init__(self, client, system_prompt, max_history): self.client = client self.system_prompt = system_prompt self.max_history = max_history @classmethod def from_config(cls, config, model=None): client = create_model_client( model=model or config["model"], temperature=config["temperature"], ) return cls( client=client, system_prompt=config["system_prompt"], max_history=config["history_size"], )注意一点:create_model_client()只是创建客户端对象,并不会立刻建立网络连接。真正连接模型服务的时机是第一次发起对话时。这个“懒连接”设计很实用,因为它保证了用户只是运行hermes --help时,不会傻乎乎地去请求一次模型接口。
另外,把“根据配置创建 Agent”放在工厂方法里,而不是在__init__里读全局 config,是为了依赖注入。测试时可以传入一个假的 client 对象,不需要真连模型服务。
3.3 使用argv=None带来的测试空间
前面已经提到main(argv=None)。这个设计在装配阶段体现得尤其明显:你可以在测试里构造一个临时配置目录、传入["-c", "/tmp/test.json", "--verbose"],然后直接调用main(),不需要真的在终端里跑命令。
如果你想把main()的测试覆盖做到位,至少要有三类用例:
--help能正常退出且不触发网络请求;- 配置文件缺失时能生成默认配置;
- 配置 JSON 非法时能抛出
ConfigError,而不是在加载途中崩成 Traceback。
这三类用例都是针对入口函数的,不需要 mock 模型,跑起来很快。很多项目的入口测试缺失,根源往往是入口函数和业务逻辑耦合太深,argv传不进去。
4. 主循环拆解:一次对话从输入到输出的完整生命周期
装配结束之后,程序进入run_chat_loop()。这是整个 Hermes 最核心的部分之一,也是“从命令行到一次对话”的关键转折点。
4.1 REPL 的骨架
Hermes 的主循环是一个典型的 REPL:读入一行,处理一行,输出结果,再回到读入状态。骨架如下:
def run_chat_loop(agent, config, input_func=input, output_func=print): session = Session.load( config.get("history_path"), max_messages=config["history_size"], ) output_func("Hermes 已启动,输入 /help 查看命令") while True: try: raw = input_func("你> ") except EOFError: break except KeyboardInterrupt: output_func("") continue text = raw.strip() if not text: continue if text.startswith("/"): should_quit = handle_command(text, agent, session, output_func=output_func) if should_quit: break continue reply = agent.turn(text, session, output_func=output_func) if reply: output_func("")这个循环表面上只有几行,但它其实划分了三条完全不同的处理路径:空输入直接忽略、斜杠命令不走模型、普通文本才进入对话生成。
有一点很值得学:主循环把“输入获取”这件事抽象成了input_func和output_func参数。这样测试时可以把真实的input()换成预设字符串序列,把print()换成收集器,完全不需要 mock 标准输入输出。
4.2 Agent.turn 内部发生了什么
主循环本身不做对话生成,它把普通文本交给agent.turn()。一次“对话回合”的完整过程是这样的:
class Agent: def turn(self, user_text, session, output_func=print): session.add("user", user_text) messages = self._build_messages(session) stream = self.client.stream_chat(messages) chunks = [] for chunk in stream: chunks.append(chunk) output_func(chunk, end="", flush=True) reply = "".join(chunks) output_func() session.add("assistant", reply) session.prune(self.max_history) return reply def _build_messages(self, session): messages = [{"role": "system", "content": self.system_prompt}] messages.extend(session.messages) return messages这轮操作可以拆成四步:
- 把用户输入追加到会话历史;
- 在历史前面加上系统提示词,构造完整的模型请求;
- 调用模型接口的流式生成,边生成边往终端打印;
- 把完整回复追加到会话历史,并裁剪超长历史。
这里最值得思考的是“为什么流式输出时还要累积完整回复”。因为屏幕上的流式输出只是给用户看的,历史记录需要的是完整回复。如果只打印不累积,那历史里就只剩用户消息,下一轮对话模型就会丢失上一轮的助手回复。因此chunks列表的作用是既兼顾实时体验,又不破坏上下文完整性。
还有一个小细节:session.add("user", user_text)发生在模型调用之前。如果模型请求失败,历史里会留下一条用户消息,但不会有助手回复。这个状态对用户来说其实没问题,因为下一轮再提问时,模型能知道用户刚才问过什么,用户也可以选择重新发送同一句话。
4.3 命令路由与对话状态的关系
斜杠命令在主循环里的判断很简单,就是text.startswith("/")。进来之后全部交给handle_command(),由命令模块自己决定要不要退出循环。
/clear这种命令会直接改动 session 状态,/exit会返回True让主循环 break。这里有一个容易踩的坑:不要把斜杠命令写进对话历史。Hermes 的处理方式是在handle_command()内部直接操作 session,而不是把它伪装成用户消息塞给模型。否则模型会看到一堆和对话无关的控制指令,反而影响回复质量。
命令路由之所以放在主循环里而不是 Agent 里,是因为命令本质上是“进程控制”和“会话控制”,跟模型生成能力没有关系。如果你把/exit交给 Agent,那 Agent 就得关心用户界面层的退出逻辑,职责就乱了。
5. 中断、异常与退出:主循环的兜底设计
一个 REPL 程序只写完主流程是不够的。终端环境下用户会按 Ctrl+C,会重定向输入导致 EOF,模型接口也会偶发超时。这些情况在主循环里都得有明确的兜底策略。
5.1 KeyboardInterrupt 在不同阶段的不同处理
KeyboardInterrupt在 Python 里就是普通异常,但它在不同代码位置出现时,处理方式完全不同。
在主循环等待输入的位置,用户按 Ctrl+C 通常意味着“我想重新输入”,而不是“我想退出程序”。所以 Hermes 捕获后只是输出一个空行,然后continue,回到下一次input_func。这符合大多数 REPL 工具的习惯,也和 shell 的 Ctrl+C 语义一致。
但在模型流式输出过程中按 Ctrl+C,情况就不一样了。用户看到一条回答生成到一半,按了中断,这时程序应该停止继续消费流,并且确保历史里没有半截回复。Hermes 的做法是Agent.turn()内部只在完整回复生成后才追加 assistant 消息,所以在中断发生时,session.add("assistant", reply)还没来得及执行,历史不会被污染。你只需要在主循环外层捕获异常,打印提示,然后 continue:
try: reply = agent.turn(text, session, output_func=output_func) except KeyboardInterrupt: output_func("\n[已中断]") continue这个设计里最关键的一点是“先完整累积,后写入历史”。如果边生成边写历史,中断就必然留下脏数据;累积完成后统一写入,中断只会影响当前回合,不会破坏会话一致性。
5.2 异常边界:哪些错误必须让程序死掉
很多人在写主循环时会犯另一个错误:把所有异常全部捕获,然后继续跑。这样程序确实不会崩,但错误会被静默吞掉,用户完全不知道发生了什么。
Hermes 的划分方式是:
- 启动阶段的配置错误、模型客户端创建失败,属于致命错误,直接向上抛,进程退出;
- 主循环里单次模型调用失败,属于可恢复错误,捕获后打印友好信息,继续等待下一轮输入;
- 本地代码的
KeyError、TypeError这类 bug,不捕获,让 Traceback 暴露出来。
这个边界很重要。你可以在主循环里加上except ModelAPIError来告诉用户“模型暂时不可用”,但绝对不要吞掉你自己代码里的逻辑错误。否则一个 KeyError 可能会让程序进入半死状态,用户还不清楚哪里出了问题。
5.3 退出前最后要做的事
主循环 break 之后,并不是直接return就完事。会话历史需要持久化,日志缓冲需要 flush。Hermes 的做法是在run_chat_loop()外层用try/finally保证退出时一定执行清理:
def main(argv=None): ... session = None try: return run_chat_loop(agent, config) finally: if session is not None: session.close() logging.shutdown()session.close()会把当前历史写回本地文件,这样下次启动hermes时还能继续上一次的上下文。很多人会忽略这一步,结果是每次会话都从零开始。对对话工具来说,历史持久化是体验的一部分。
如果你在命令行里用Ctrl+D触发 EOF,input()会抛出EOFError,主循环 break,finally 里的清理逻辑同样会执行。这也是为什么不要在run_chat_loop()里直接os._exit(0),那样会跳过 Python 的清理机制,可能丢失历史。
6. 复现与调试:把启动流程变成可测试的接口
源码读到这里,你会发现 Hermes 的整个启动链路其实很适合测试。它不像很多脚本那样把main()写得又长又不可调用,而是通过argv、input_func、output_func三个参数把外部依赖隔离掉了。
6.1 为循环注入 input/output
前面看到run_chat_loop(agent, config, input_func=input, output_func=print),这意味着测试里可以替换掉这两个函数。一个简单的假输出收集器长这样:
class FakeOutput: def __init__(self): self.chunks = [] def __call__(self, *args, **kwargs): self.chunks.append("".join(args))如果 Agent 使用了output_func(chunk, end="", flush=True),那FakeOutput.__call__会忽略关键字参数,只收集文本内容。测试完成后,把chunks拼起来看是否符合预期即可。
6.2 用 pytest 覆盖一次对话
你可以用假客户端把一次完整对话跑通:
class StubClient: def stream_chat(self, messages): yield "你好," yield "我是 Hermes。" def test_one_turn_then_exit(tmp_path): config = { "history_size": 20, "history_path": tmp_path / "history.json", } session = Session.load(config["history_path"], max_messages=20) agent = Agent(client=StubClient(), system_prompt="", max_history=20) inputs = iter(["你好", "/exit"]) out = FakeOutput() run_chat_loop(agent, config, input_func=lambda: next(inputs), output_func=out) text = "".join(out.chunks) assert "你好," in text assert "我是 Hermes。" in text这个测试不需要真实网络,也不需要等模型返回,跑起来几乎不耗时。它能验证的事却很关键:主循环会不会正确消费两条输入、/exit会不会中断循环、流式输出会不会被收集到。
6.3 实际调试入口时最常用的小技巧
如果你不是在写测试,而是想手动断点调试 Hermes 的启动流程,我有几个实际常用的手段:
用python -m hermes --verbose启动,可以看到日志里打出配置文件路径和模型名,确认装配阶段是否按预期执行。
如果怀疑配置加载有问题,先跑python -m hermes --help。这个命令不应该触发任何模型请求,如果它卡住或者开始读配置,说明入口函数在参数解析阶段做了不该做的事。
想快速验证主循环时,可以把标准输入重定向成一个文本文件:
printf "你好\n/exit\n" | python -m hermes --config /tmp/test.json这样不用手动敲字,就能看到主循环对两行输入的处理结果。遇到异常想追调用链,直接在run_chat_loop()的while True那一行打上断点,观察text值的变化,比到处加 print 高效得多。
我自己的经验是,源码阅读最怕一口气读太多模块。入口和主循环像是整栋房子的门厅和走廊,先把这两块的路径走通,后面看 Engine、Session、Commands 时心里就有了一张地图。Hermes 这个项目把这层关系做得特别清晰,读完之后你甚至可以把这套“入口 + 装配 + REPL”的结构直接搬到你自己的 CLI 工具里——它的价值不只在对话本身,更在进程生命周期的组织方式。