【技术干货】Hermes Agent Kanban 深度解析:从聊天式 Agent 到持久化多角色工作流
摘要Hermes Agent 近期版本更新的核心不只是新增 Provider 和插件而是向“持久化 Agent 工作流”演进。本文结合 Kanban 教程解析任务编排、运行历史、失败恢复与多角色协作并用 Python 实现一个可运行的轻量级 Agent Kanban 示例。背景介绍Hermes Agent 最近连续发布了 v0.1.1 与 v0.1.12 两个重要版本。表面上看这两次更新包含大量功能新的 CLI、Provider 扩展、Dashboard 增强、后台 Agent、更多插件与多模态能力。但从工程视角看真正值得关注的是 Hermes 正在从“聊天式 Agent”转向“持久化 Agent 协作系统”。v0.1.1Interface Releasev0.1.1 被称为 Interface Release重点是交互层和 Provider 架构升级基于 React Ink 重构交互式 CLI支持 sticky composer、实时流式输出、状态栏、浅色主题子 Agent 重启时具备更高可见性Provider 通信拆分为更清晰的可插拔 Transport Layer原生支持 AWS Bedrock Converse API扩展 NVIDIA、Google Gemini CLI OAuth、Vercel Gateway 等推理路径增加 QQ Bot Adapter、Shell Hooks、Direct Delivery、Slash Steer 等能力。这说明 Hermes 开始把 Agent 系统拆解为更标准的接口层、传输层和插件层。v0.1.12Curator Releasev0.1.12 被称为 Curator Release重点是后台自治能力Autonomous Curator后台 Agent 可定期评分、裁剪、合并技能库Self-improvement Loop 增强支持基于 Rubric 的技能更新子任务可继承父级运行时的 Provider、Model 与 Credential新增 Azure AI Foundry、MiniMax、Tencent Token Hub、LM Studio 等 ProviderDashboard 增加 Models Tab改进多模态图片路由、Gateway 媒体处理集成 Spotify、Google Meet、ComfyUI、TouchDesigner MCP冷启动性能提升约 57%。这些变化共同指向一个方向Hermes 不再只是“问一句、答一句”的 Agent而是在构建一个可持续运行、可观测、可恢复的 Agent 工作流系统。核心原理KanbanAgent 工作流的状态载体Hermes 文档中新加入的 Kanban 教程是理解其演进方向的关键。传统 Agent 系统通常依赖 Chat Log 保存上下文但聊天记录并不适合作为工程任务的状态管理系统。Kanban 的价值在于将任务状态显式结构化ready任务可执行claimed任务已被某个 Worker 获取blocked任务执行失败需要人工或上游修正completed任务完成crashedWorker 中途崩溃gave_up超过失败阈值后熔断。这与真实研发流程非常接近。例如PM Agent 编写需求文档Engineer Agent 根据需求实现功能Reviewer Agent 审查实现如果缺少密码强度校验或重置链接不是一次性 Token则任务进入blockedEngineer 根据反馈重试第二次执行成功并记录变更文件、测试结果与元数据。重点是重试历史不再隐藏在聊天记录中而是作为结构化数据保存。运行历史与元数据Hermes Kanban 的一个关键设计是 Run History。每一次执行都应记录执行角色开始与结束时间输出摘要失败原因修改文件测试结果上一次失败反馈。这样 Reviewer 在审查时可以先查看实现摘要和变更文件再进入 Diff。对于长周期 Agent 工作流这比单纯依赖上下文窗口更加可靠。熔断与崩溃恢复真实 Worker 会失败常见原因包括API Key 缺失网络异常模型限流内存不足子进程崩溃。Kanban 调度器需要两类保护Circuit Breaker连续启动失败超过阈值后将任务标记为gave_up避免无限重试Crash RecoveryWorker 中途崩溃时释放任务 Claim将任务重新放回ready交给新 Worker 执行。这正是 Hermes Kanban 教程强调的工程化价值任务状态、失败记录、重试历史和人工干预点必须可见。技术资源与工具选型在多 Agent 工作流中模型能力与 API 稳定性会直接影响执行质量。我个人常用的 AI 开发平台是薛定猫 AIxuedingmao.com它采用 OpenAI 兼容接口适合快速接入多模型实验环境。其技术价值主要体现在聚合 500 主流大模型包括 GPT-5.4、Claude 4.6、Gemini 3.1 Pro 等新模型实时首发开发者可以第一时间验证前沿模型 API使用统一 OpenAI Compatible 接口降低多模型切换与集成成本对 Agent 系统开发来说统一base_url api_key model的接入方式非常适合做 Provider 抽象。下面示例默认使用claude-opus-4-6。该模型在长上下文理解、复杂任务规划、代码审查、多步骤推理方面表现很强适合承担 PM、Engineer、Reviewer 这类角色化 Agent。实战演示用 Python 实现轻量级 Agent Kanban下面代码实现一个简化版 Kanban 调度器使用 SQLite 保存任务与运行历史使用 OpenAI 兼容接口调用模型模拟 PM → Engineer → Reviewer 流程支持失败记录、阻塞状态与重试包含基础 Circuit Breaker。安装依赖pipinstallopenai完整代码importosimportjsonimportsqlite3importtimefromdatetimeimportdatetimefromtypingimportOptionalfromopenaiimportOpenAI DB_PATHagent_kanban.dbMODEL_NAMEclaude-opus-4-6MAX_FAILURES3defnow()-str:returndatetime.utcnow().isoformat()defget_client()-OpenAI: 薛定猫 AI 采用 OpenAI 兼容模式 base_url api_key model 即可完成接入。 api_keyos.getenv(XUEDINGMAO_API_KEY)ifnotapi_key:raiseRuntimeError(Missing XUEDINGMAO_API_KEY)returnOpenAI(api_keyapi_key,base_urlhttps://xuedingmao.com/v1)definit_db()-None:connsqlite3.connect(DB_PATH)curconn.cursor()cur.execute( CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, role TEXT NOT NULL, status TEXT NOT NULL DEFAULT ready, input TEXT NOT NULL, output TEXT, failure_count INTEGER NOT NULL DEFAULT 0, parent_id INTEGER, metadata TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) )cur.execute( CREATE TABLE IF NOT EXISTS runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id INTEGER NOT NULL, role TEXT NOT NULL, status TEXT NOT NULL, summary TEXT, error TEXT, metadata TEXT, started_at TEXT NOT NULL, ended_at TEXT, FOREIGN KEY(task_id) REFERENCES tasks(id) ) )conn.commit()conn.close()defcreate_task(title:str,role:str,input_text:str,parent_id:Optional[int]None)-int:connsqlite3.connect(DB_PATH)curconn.cursor()cur.execute( INSERT INTO tasks(title, role, status, input, parent_id, metadata, created_at, updated_at) VALUES (?, ?, ready, ?, ?, {}, ?, ?) ,(title,role,input_text,parent_id,now(),now()))task_idcur.lastrowid conn.commit()conn.close()returntask_iddefclaim_task()-Optional[dict]: 获取一个 ready 任务并将其状态改为 claimed。 单机 SQLite 场景下可满足本地 Agent 编排。 connsqlite3.connect(DB_PATH)conn.row_factorysqlite3.Row curconn.cursor()cur.execute( SELECT * FROM tasks WHERE status ready ORDER BY id ASC LIMIT 1 )rowcur.fetchone()ifnotrow:conn.close()returnNonecur.execute( UPDATE tasks SET status claimed, updated_at ? WHERE id ? ,(now(),row[id]))conn.commit()conn.close()returndict(row)defcall_agent(role:str,task_input:str)-str:clientget_client()system_promptf 你是一个专业的软件研发 Agent当前角色是{role}。 请输出结构化结果包含 1. summary执行摘要 2. details详细内容 3. metadataJSON 格式元数据例如 changed_files、tests、risk responseclient.chat.completions.create(modelMODEL_NAME,messages[{role:system,content:system_prompt.strip()},{role:user,content:task_input}],temperature0.2)returnresponse.choices[0].message.contentdefrecord_run(task_id:int,role:str,status:str,summary:str,error:str,metadata:Optional[dict]None)-None:connsqlite3.connect(DB_PATH)curconn.cursor()cur.execute( INSERT INTO runs(task_id, role, status, summary, error, metadata, started_at, ended_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ,(task_id,role,status,summary,error,json.dumps(metadataor{},ensure_asciiFalse),now(),now()))conn.commit()conn.close()defupdate_task(task_id:int,status:str,output:str,metadata:Optional[dict]None)-None:connsqlite3.connect(DB_PATH)curconn.cursor()cur.execute( UPDATE tasks SET status ?, output ?, metadata ?, updated_at ? WHERE id ? ,(status,output,json.dumps(metadataor{},ensure_asciiFalse),now(),task_id))conn.commit()conn.close()defincrease_failure(task_id:int,error:str)-None:connsqlite3.connect(DB_PATH)curconn.cursor()cur.execute(SELECT failure_count FROM tasks WHERE id ?,(task_id,))failure_countcur.fetchone()[0]1iffailure_countMAX_FAILURES:statusgave_upelse:statusreadycur.execute( UPDATE tasks SET failure_count ?, status ?, updated_at ? WHERE id ? ,(failure_count,status,now(),task_id))conn.commit()conn.close()record_run(task_idtask_id,roledispatcher,statusstatus,summary任务执行失败已触发失败计数,errorerror,metadata{failure_count:failure_count})defworker_loop_once()-bool:taskclaim_task()ifnottask:returnFalsetask_idtask[id]roletask[role]try:resultcall_agent(role,task[input])# 示例Reviewer 发现缺陷时将任务阻塞ifrolereviewerand缺少inresult:update_task(task_id,blocked,result,{reason:review_failed})record_run(task_id,role,blocked,result)else:update_task(task_id,completed,result,{model:MODEL_NAME})record_run(task_id,role,completed,result)exceptExceptionasexc:increase_failure(task_id,str(exc))returnTruedefprint_runs()-None:connsqlite3.connect(DB_PATH)conn.row_factorysqlite3.Row curconn.cursor()cur.execute( SELECT r.id, r.task_id, r.role, r.status, r.summary, r.error, r.metadata FROM runs r ORDER BY r.id ASC )forrowincur.fetchall():print(*80)print(fRun #{row[id]}| Task{row[task_id]}|{row[role]}|{row[status]})ifrow[summary]:print(row[summary][:500])ifrow[error]:print(ERROR:,row[error])print(META:,row[metadata])conn.close()if__name____main__:init_db()pm_taskcreate_task(title编写密码重置功能需求文档,rolepm,input_text请为一个 SaaS 系统设计密码重置功能需求包含安全约束、验收标准和边界场景。)engineer_taskcreate_task(title实现密码重置功能,roleengineer,input_text根据 PM 需求实现密码重置功能要求包含密码强度校验、一次性重置链接和测试说明。,parent_idpm_task)reviewer_taskcreate_task(title审查密码重置实现,rolereviewer,input_text请审查工程实现是否满足安全要求重点检查密码强度、Token 单次使用、过期时间和测试覆盖。,parent_idengineer_task)whileworker_loop_once():time.sleep(1)print_runs()运行前设置 API KeyexportXUEDINGMAO_API_KEY你的 API Keypython kanban_agent.py这个示例虽然简化但已经体现 Hermes Kanban 的核心思想任务不是聊天上下文的一部分而是可查询、可恢复、可审计的数据对象。注意事项1. Kanban 不适合所有 Agent 任务如果只是让另一个 Agent 快速回答一个问题直接 Delegate Task 更轻量。Kanban 更适合长周期任务多角色协作需要审查与重试需要保留历史需要人工介入周期性运营或监控任务。2. 单机边界需要明确Hermes 文档中提到Kanban 设计上更像单主机本地协调层通常基于本地 SQLite 文件和同机 Worker。它不是多服务器共享的企业级工作流引擎。如果要扩展到分布式场景需要额外引入PostgreSQL / MySQL分布式锁消息队列Worker 心跳任务租约权限模型。3. Dashboard 不应随意暴露如果 Dashboard 绑定到0.0.0.0插件路由可能从网络访问。对于普通开发环境保持 localhost 是更安全的默认选择。4. 失败历史必须结构化长任务 Agent 系统最怕“失败但不可见”。建议至少记录输入输出错误堆栈模型名称运行时间重试次数阻塞原因人工处理记录。总结Hermes Agent v0.1.1 与 v0.1.12 的更新本质上是在构建一个更工程化的 Agent Runtime。Kanban 教程尤其重要因为它把 Agent 的执行过程从“聊天记录”提升为“持久化状态机”。对于真实开发场景Agent 系统不能只关注模型回答质量还必须处理任务状态、角色协作、失败恢复、审查反馈和运行审计。Hermes Kanban 的设计方向正是 AI Agent 从 Demo 走向工程落地的关键一步。#AI #大模型 #Python #机器学习 #技术实战