摘要
Hermes Agent 0.1.3 的核心不是“多几个功能”,而是围绕长运行智能体的可靠性重构:持久化任务队列、心跳检测、状态恢复、目标保持、安全默认值、插件化模型提供商与工具链增强,为真实生产级 Agent 工作流提供了更稳的工程底座。
背景介绍:AI Agent 正从 Demo 走向工程系统
很多 AI Agent 在单轮任务中表现不错,例如生成代码、总结文档、调用工具。但一旦进入真实场景,问题会迅速暴露:
- 多步骤任务容易偏离最初目标;
- 工作者进程崩溃后任务状态不一致;
- 长时间运行时会话中断、上下文丢失;
- Agent 声称“已完成”,但真实文件或看板状态并未变化;
- 多平台接入、模型切换、安全边界越来越复杂。
Hermes Agent 0.1.3 被命名为Tenacity Release,重点正是解决这些问题。它更像一次“可靠性版本”,而不是单纯的功能版本。
核心原理:Hermes 0.1.3 的可靠性设计
1. 持久化多智能体 Kanban:从看板到工作队列
过去的 Kanban 更偏向可视化任务管理,而 Hermes 0.1.3 将其进一步演进为durable work queue。
关键机制包括:
- Heartbeat:Worker 定期上报存活状态;
- Zombie Detection:检测长时间无心跳的异常任务;
- Reclaim Logic:自动回收卡死任务;
- Retry Budget:限制单任务最大重试次数;
- Automatic Blocking:超过重试上限后自动阻塞,避免无限循环;
- Hallucination Gate:校验 Agent 声称完成的结果是否与真实状态一致。
这类机制本质上借鉴了分布式任务队列的设计思想,例如 Celery、Temporal、Sidekiq 中的租约、重试、死信队列等能力。
2. Slash Goal:跨回合目标保持
长任务中,Agent 最常见的问题是goal drift,即随着上下文变长逐渐偏离原始目标。
/goal的价值在于为会话设置一个持久目标,使后续推理、工具调用、任务拆解都围绕该目标优化。对于代码重构、数据分析、自动化运维等长链路任务,这比单次 prompt 更可靠。
3. Checkpoints v2:状态恢复与会话续跑
Hermes 0.1.3 重写了状态持久化层,引入:
- checkpoint pruning;
- gateway restart auto-resume;
- interrupted session recovery;
- guard rails。
这意味着 Agent 不再依赖“进程一直活着”。即使网关重启、会话中断,也可以尽量从最近检查点恢复,而不是从头开始。
4. 安全默认值升级
本次版本修复了多个 P0 安全问题,并强化了:
- Secret redaction;
- Discord guild scoped allowlist;
- Webhook stranger rejection;
- MCP OAuth handling;
- SSRF 防护;
- 云元数据访问保护;
- Prompt injection scanning;
- Debug log redaction。
对于接入浏览器、文件系统、消息平台和第三方 API 的 Agent 来说,安全默认值比功能数量更重要。
工具选型:统一模型接入与多模型实验
在 Agent 系统中,模型提供商插件化非常关键。Hermes 0.1.3 引入 provider profile abstraction 和 model provider plugin directory,方便在不同模型、不同路由之间切换。
我在日常 AI 开发中会使用薛定猫AI(xuedingmao.com)作为统一模型接入层。它采用 OpenAI 兼容接口,适合快速接入多模型 Agent 工作流:
- 聚合 500+ 主流大模型,例如 GPT-5.4、Claude 4.6、Gemini 3.1 Pro 等;
- 新模型更新速度快,便于第一时间验证前沿 API;
- 统一
base_url + api_key + model调用方式,减少多供应商 SDK 适配成本; - 适合做模型路由、A/B 测试、Agent provider plugin 抽象。
下面实战示例中使用claude-opus-4-6。该模型在复杂推理、长上下文规划、代码生成和 Agent 协调任务中表现很强,适合作为多步骤任务的规划器或执行器。
实战演示:用 Python 实现一个简化版可靠 Agent 工作队列
下面代码模拟 Hermes Tenacity 的核心思想:持久化任务、心跳、僵尸任务回收、重试预算、目标保持、模型调用与结果校验。
环境准备
pipinstallopenaiexportXUEDINGMAO_API_KEY="你的API_KEY"完整代码示例
importosimportjsonimporttimeimportsqlite3importthreadingfromcontextlibimportcontextmanagerfromdatetimeimportdatetime,timedelta,timezonefromopenaiimportOpenAI DB_PATH="agent_queue.db"WORKER_ID=f"worker-{os.getpid()}"HEARTBEAT_INTERVAL=5ZOMBIE_TIMEOUT_SECONDS=20client=OpenAI(api_key=os.getenv("XUEDINGMAO_API_KEY"),base_url="https://xuedingmao.com/v1")MODEL="claude-opus-4-6"defnow_iso()->str:returndatetime.now(timezone.utc).isoformat()@contextmanagerdefdb():conn=sqlite3.connect(DB_PATH)conn.row_factory=sqlite3.Rowtry:yieldconn conn.commit()finally:conn.close()definit_db():withdb()asconn:conn.execute(""" CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, status TEXT NOT NULL, goal TEXT NOT NULL, payload TEXT NOT NULL, result TEXT, error TEXT, worker_id TEXT, heartbeat_at TEXT, retries INTEGER NOT NULL DEFAULT 0, max_retries INTEGER NOT NULL DEFAULT 3, checkpoint TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) """)defcreate_task(goal:str,payload:dict,max_retries:int=3):withdb()asconn:conn.execute(""" INSERT INTO tasks(status, goal, payload, max_retries, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?) """,("queued",goal,json.dumps(payload,ensure_ascii=False),max_retries,now_iso(),now_iso()))defreclaim_zombies():deadline=datetime.now(timezone.utc)-timedelta(seconds=ZOMBIE_TIMEOUT_SECONDS)withdb()asconn:rows=conn.execute(""" SELECT * FROM tasks WHERE status = 'running' AND heartbeat_at < ? """,(deadline.isoformat(),)).fetchall()forrowinrows:next_retries=row["retries"]+1ifnext_retries>=row["max_retries"]:conn.execute(""" UPDATE tasks SET status = 'blocked', retries = ?, error = ?, updated_at = ? WHERE id = ? """,(next_retries,"Task blocked: retry budget exhausted after zombie detection.",now_iso(),row["id"]))else:conn.execute(""" UPDATE tasks SET status = 'queued', retries = ?, worker_id = NULL, heartbeat_at = NULL, updated_at = ? WHERE id = ? """,(next_retries,now_iso(),row["id"]))defclaim_task():withdb()asconn:row=conn.execute(""" SELECT * FROM tasks WHERE status = 'queued' ORDER BY id ASC LIMIT 1 """).fetchone()ifnotrow:returnNoneconn.execute(""" UPDATE tasks SET status = 'running', worker_id = ?, heartbeat_at = ?, checkpoint = ?, updated_at = ? WHERE id = ? """,(WORKER_ID,now_iso(),json.dumps({"phase":"claimed"},ensure_ascii=False),now_iso(),row["id"]))returndict(row)defheartbeat(task_id:int,stop_event:threading.Event):whilenotstop_event.is_set():withdb()asconn:conn.execute(""" UPDATE tasks SET heartbeat_at = ?, updated_at = ? WHERE id = ? AND worker_id = ? AND status = 'running' """,(now_iso(),now_iso(),task_id,WORKER_ID))time.sleep(HEARTBEAT_INTERVAL)defupdate_checkpoint(task_id:int,checkpoint:dict):withdb()asconn:conn.execute(""" UPDATE tasks SET checkpoint = ?, updated_at = ? WHERE id = ? """,(json.dumps(checkpoint,ensure_ascii=False),now_iso(),task_id))defcall_agent_model(goal:str,payload:dict)->dict:""" 使用 OpenAI 兼容 API 调用薛定猫AI上的 claude-opus-4-6。 要求模型输出严格 JSON,方便后续做 hallucination gate。 """response=client.chat.completions.create(model=MODEL,temperature=0.2,messages=[{"role":"system","content":("你是一个可靠的工程型AI Agent。""必须围绕用户目标执行任务,并输出严格JSON。""JSON字段包括: completed, summary, actions, risks。")},{"role":"user","content":json.dumps({"persistent_goal":goal,"task_payload":payload},ensure_ascii=False)}])content=response.choices[0].message.contentreturnjson.loads(content)defhallucination_gate(output:dict)->None:""" 简化版幻觉门控: 如果模型声称 completed=true,则必须提供 summary 和 actions。 真实工程中可进一步校验文件、数据库、看板状态是否一致。 """ifnotisinstance(output,dict):raiseValueError("Model output is not a JSON object.")required={"completed","summary","actions","risks"}missing=required-set(output.keys())ifmissing:raiseValueError(f"Missing required fields:{missing}")ifoutput["completed"]isTrueandnotoutput["actions"]:raiseValueError("Agent claims completion but no actions are recorded.")deffinish_task(task_id:int,result:dict):withdb()asconn:conn.execute(""" UPDATE tasks SET status = 'done', result = ?, checkpoint = ?, updated_at = ? WHERE id = ? """,(json.dumps(result,ensure_ascii=False),json.dumps({"phase":"finished"},ensure_ascii=False),now_iso(),task_id))deffail_task(task_id:int,error:str):withdb()asconn:conn.execute(""" UPDATE tasks SET status = 'queued', error = ?, worker_id = NULL, heartbeat_at = NULL, updated_at = ? WHERE id = ? """,(error,now_iso(),task_id))defrun_once():reclaim_zombies()task=claim_task()ifnottask:print("No queued task.")returntask_id=task["id"]stop_event=threading.Event()hb_thread=threading.Thread(target=heartbeat,args=(task_id,stop_event),daemon=True)hb_thread.start()try:goal=task["goal"]payload=json.loads(task["payload"])update_checkpoint(task_id,{"phase":"model_call_started"})output=call_agent_model(goal,payload)update_checkpoint(task_id,{"phase":"hallucination_gate_started"})hallucination_gate(output)finish_task(task_id,output)print(f"Task{task_id}done:",json.dumps(output,ensure_ascii=False,indent=2))exceptExceptionasexc:fail_task(task_id,str(exc))print(f"Task{task_id}failed:",exc)finally:stop_event.set()hb_thread.join(timeout=2)if__name__=="__main__":init_db()# 首次运行时创建一个任务;后续可注释掉,观察任务恢复与重试行为create_task(goal="生成一份面向后端团队的AI Agent可靠性改造方案",payload={"system":"现有Agent支持代码生成和消息平台接入,但缺少任务恢复、心跳和安全门控","expected_output":"给出架构建议、风险点和落地步骤"},max_retries=3)run_once()注意事项:从 Demo 到生产还差哪些能力
1. Heartbeat 需要和任务租约结合
仅有心跳还不够,生产系统中建议增加 lease timeout,避免多个 Worker 同时处理同一任务。
2. Hallucination Gate 应校验真实状态
示例只校验 JSON 字段。真实 Agent 应检查:
- 文件是否真实写入;
- patch 是否成功应用;
- 测试是否通过;
- Kanban 状态是否与声明一致;
- 外部 API 调用是否返回成功。
3. Retry Budget 需要区分错误类型
网络超时可以重试,权限错误、Prompt Injection、Schema 错误不一定适合重试。建议引入错误分类与死信队列。
4. 安全边界必须默认收紧
对于具备浏览器、Shell、文件系统、MCP 工具能力的 Agent,应默认启用:
- Secret 脱敏;
- SSRF 防护;
- 云元数据 IP 阻断;
- 工具 allowlist;
- Prompt injection 扫描;
- 审计日志。
总结
Hermes Agent 0.1.3 的重点是让 Agent 在复杂工作流中“持续运行”:任务可恢复、目标可保持、状态可持久化、安全默认值更严格、模型和平台更加插件化。对于只需要基础 AI 编程助手的用户,这些能力可能偏重;但对于构建本地 Agent 系统、消息平台机器人、计划任务、长期自动化工作流的开发者来说,这类可靠性工程正是 Agent 走向生产环境的关键。
#AI #大模型 #Python #机器学习 #技术实战