Skip to main content 홈 크리에이터 54laowang hermes-agent supervisor-mode
supervisor-mode 监察者模式 - 主 Agent 不干活,只监控 subagent 执行过程,通过干预文件实时纠偏。适用于复杂多步骤任务、高风险操作、需要质量把关的场景。
Use when: 监察者, supervisor, 监控agent, 纠偏, 质量把关, subagent监控, 任务监督.
Do NOT use for:
- 直接执行任务(用其他执行类 skills)
- 简单单步任务(无需监控)
- 低风险操作(测试环境、本地实验)
- 实时交互任务(需要用户反馈的任务)
설치로 이동 Skills Marketplace 커뮤니티가 만든 AI 스킬을 발견하고 탐색하세요.
Codex 또는 Claude로 설치 이 Prompt를 복사해 Codex, Claude 또는 다른 어시스턴트에 붙여 넣으면 Skill 페이지를 검토하고 설치를 진행할 수 있습니다.
직접 명령은 검토 Prompt를 거치지 않습니다. 실행하기 전에 소스를 확인하세요.
npx skills add https://github.com/54laowang/hermes-agent --skill supervisor-mode명령은 한 줄로 유지됩니다. 복사하기 전에 가로로 스크롤해 전체 내용을 확인하세요.
로컬 사본을 원하시나요? SkillsMP에서 현재 제공할 수 있는 파일을 다운로드하세요.
Zip 다운로드 다운로드 중... name supervisor-mode description 监察者模式 - 主 Agent 不干活,只监控 subagent 执行过程,通过干预文件实时纠偏。适用于复杂多步骤任务、高风险操作、需要质量把关的场景。
Use when: 监察者, supervisor, 监控agent, 纠偏, 质量把关, subagent监控, 任务监督.
Do NOT use for:
- 直接执行任务(用其他执行类 skills)
- 简单单步任务(无需监控)
- 低风险操作(测试环境、本地实验)
- 实时交互任务(需要用户反馈的任务)
version 1.3.0 category agent-optimization author Hermes + GenericAgent tags ["monitoring","subagent","quality-control","intervention"] keywords ["监察者","supervisor","监控agent","纠偏","质量把关","subagent监控"] triggers ["监察者模式","supervisor mode","监控agent","纠偏","质量把关","任务监督"]
监察者模式 (Supervisor Mode)
核心理念
你是挑刺的监工,不是干活的工人。你的唯一任务:确保工作 agent 高质量完成任务。有 SOP 按 SOP 约束,无 SOP 凭常理和经验把关。
三大红线 :
❌ 禁止下场干活 :不操作浏览器、不写代码、不执行任务步骤
✅ 可以读环境 :file_read / web_scan / code_run(只读命令) 获取情报
✅ 只做判断和干预 :检查约束、发现偏差、实时纠偏
使用场景
场景 1:有 SOP 的任务(推荐)
用户任务:用 A股市场分析标准化流程 分析今日大盘
启动命令:python agentmain.py --task stock_analysis --bg --verbose
监察重点:
- 是否先获取精确时间
- 是否判断盘前/盘后状态
- 数据源是否使用财联社(P0级)
- 时间戳是否三重校验
场景 2:无 SOP 的复杂任务
用户任务:爬取某网站数据并生成报告
启动命令:python agentmain.py --task web_scrape --bg
监察重点:
- 是否遵循 robots.txt
- 是否处理反爬机制
- 错误是否优雅降级
- 输出格式是否符合预期
场景 3:高风险操作
用户任务:部署到生产环境
启动命令:python agentmain.py --task deploy_prod --bg
监察重点:
- 是否有备份
- 是否有回滚计划
- 是否通知相关方
- 是否验证结果
完整工作流程
第零步:环境预检(新增)
⚠️ 必须在启动监控前执行 ,确保环境就绪。
0.1 环境检查清单
def pre_flight_check ():
"""
监控启动前的环境检查
"""
checks = []
task_dir = f"~/.hermes/tasks/{task_id} "
if not os.path.exists(task_dir):
os.makedirs(task_dir, exist_ok=True )
checks.append(("task_dir" , "created" , "OK" ))
else :
checks.append(("task_dir" , "exists" , "OK" ))
output_file = f"{task_dir} /output.txt"
try :
with open (output_file, 'a' ) as f:
f.write( )
checks.append(( , , ))
PermissionError:
checks.append(( , , ))
EnvironmentError( )
intervention_file [ , , ]:
:
test_path =
os.path.exists(test_path):
os.remove(test_path)
Exception e:
checks.append((intervention_file, (e), ))
log_dir =
os.path.exists(log_dir):
os.makedirs(log_dir, exist_ok= )
checks.append(( , , ))
sop_name:
sop_path =
os.path.exists(sop_path):
checks.append(( , , ))
( )
:
checks.append(( , , ))
{
: (c[ ] == c checks) ,
: checks,
: datetime.now().isoformat()
}
""
"output_file"
"writable"
"OK"
except
"output_file"
"permission_denied"
"FAIL"
raise
f"无法写入输出文件: {output_file} "
for
in
"_intervene"
"_keyinfo"
"_stop"
try
f"{task_dir} /{intervention_file} "
if
except
as
str
"WARN"
"~/.hermes/logs"
if
not
True
"log_dir"
"created"
"OK"
if
f"~/.hermes/skills/{sop_name} /SKILL.md"
if
not
"sop_file"
"not_found"
"WARN"
print
f"警告:SOP 文件不存在: {sop_path} "
else
"sop_file"
"exists"
"OK"
return
"status"
"READY"
if
all
2
"OK"
for
in
else
"WARNINGS"
"checks"
"timestamp"
0.2 快速检查模板
result = pre_flight_check()
if result["status" ] != "READY" :
print (f"⚠️ 环境检查有警告:{result['checks' ]} " )
第一步:准备阶段
1.1 提取约束清单
sop_content = read_file("~/.hermes/skills/a-stock-market-time-aware-analysis/SKILL.md" )
constraints = extract_constraints(sop_content)
constraints = infer_constraints_from_task(task_description)
1.2 启动 subagent
result = delegate_task(
goal="用 A股市场分析标准化流程 分析今日大盘" ,
context="当前时间:2026-04-30 01:54,深夜,需注意数据时效性" ,
role="leaf" ,
toolsets=["web" , "terminal" , "file" ]
)
task_id = result.get("task_id" )
1.3 记录初始状态 working_memory = {
"task_id" : task_id,
"constraints" : constraints,
"current_step" : "启动" ,
"start_time" : "2026-04-30 01:54:34" ,
"output_file" : f"~/.hermes/tasks/{task_id} /output.txt" ,
"interventions" : []
}
第二步:监控循环
2.1 轮询输出文件 import time
def monitor_subagent (task_id, constraints, poll_interval=2 ):
"""
持续监控 subagent 执行过程
"""
output_file = f"~/.hermes/tasks/{task_id} /output.txt"
last_position = 0
while True :
with open (output_file, 'r' ) as f:
f.seek(last_position)
new_content = f.read()
last_position = f.tell()
if new_content:
analysis = analyze_output(new_content, constraints)
if analysis.needs_intervention:
apply_intervention(task_id, analysis.intervention_type, analysis.message)
if is_task_complete(new_content):
break
time.sleep(poll_interval)
2.2 分析输出 def analyze_output (content, constraints ):
"""
分析 subagent 输出,判断是否需要干预
"""
current_step = detect_current_step(content)
violations = check_constraints(content, constraints)
anomalies = detect_anomalies(content)
return {
"current_step" : current_step,
"needs_intervention" : len (violations) > 0 or len (anomalies) > 0 ,
"intervention_type" : determine_intervention_type(violations, anomalies),
"message" : generate_intervention_message(violations, anomalies)
}
2.3 判断逻辑表 发现 干预类型 示例消息 跳步 _intervene"你跳过了 Step 2(获取精确时间),先做这一步" 细节遗漏 _intervene"你漏了约束:必须使用财联社作为数据源,重做" 光说不做 _intervene"别分析了,直接执行搜索" 断言无据 _intervene"你怎么确认数据是最新的?验证一下时间戳" 连续失败 _intervene"停,先读错误日志 /tmp/agent_error.log 再决定" 感觉要偏 _intervene"去重读 SOP 的 Step 3 再继续" 即将进入关键步骤 _keyinfo"下一步需要调用财联社 API,注意 token 参数" 严重错误 _stop"检测到严重错误,立即停止"
第三步:干预机制
3.1 干预文件协议
def apply_intervention (task_id, intervention_type, message ):
"""
应用干预(带异常处理)
"""
task_dir = os.path.expanduser(f"~/.hermes/tasks/{task_id} " )
if not os.path.exists(task_dir):
raise FileNotFoundError(
f"任务目录不存在: {task_dir} \n"
"可能原因:subagent 未正确启动或 task_id 无效\n"
"解决方案:检查 delegate_task 返回值,确认 task_id 正确"
)
try :
if intervention_type == "_intervene" :
file_path = f"{task_dir} /_intervene"
with open (file_path, 'w' ) as f:
f.write(f"[{datetime.now().isoformat()} ] {message} " )
log_intervention(task_id, intervention_type, message, "success" )
elif intervention_type == "_keyinfo" :
file_path = f"{task_dir} /_keyinfo"
with open (file_path, 'w' ) as f:
f.write(f"[{datetime.now().isoformat()} ] {message} " )
log_intervention(task_id, intervention_type, message, "success" )
elif intervention_type == "_stop" :
file_path = f"{task_dir} /_stop"
with open (file_path, 'w' ) as f:
f.write(f"[{datetime.now().isoformat()} ] {message} " )
log_intervention(task_id, intervention_type, message, "success" )
else :
raise ValueError(f"未知的干预类型: {intervention_type} " )
except PermissionError as e:
log_intervention(task_id, intervention_type, message, f"permission_error: {e} " )
raise PermissionError(
f"无法写入干预文件: {file_path} \n"
"可能原因:权限不足\n"
"解决方案:检查目录权限或使用 sudo"
)
except IOError as e:
log_intervention(task_id, intervention_type, message, f"io_error: {e} " )
raise IOError(
f"写入干预文件失败: {file_path} \n"
f"错误详情: {e} \n"
"解决方案:检查磁盘空间和文件系统状态"
)
def log_intervention (task_id, intervention_type, message, status ):
"""记录干预行为到日志"""
log_entry = {
"timestamp" : datetime.now().isoformat(),
"task_id" : task_id,
"type" : intervention_type,
"message" : message,
"status" : status
}
log_file = os.path.expanduser(f"~/.hermes/logs/interventions_{datetime.now().strftime('%Y%m%d' )} .jsonl" )
os.makedirs(os.path.dirname(log_file), exist_ok=True )
with open (log_file, 'a' ) as f:
f.write(json.dumps(log_entry) + "\n" )
3.2 干预文件不存在时的处理 def read_intervention_file (task_id, intervention_type ):
"""
安全读取干预文件(带降级处理)
"""
task_dir = os.path.expanduser(f"~/.hermes/tasks/{task_id} " )
file_path = f"{task_dir} /{intervention_type} "
if not os.path.exists(file_path):
return None
try :
with open (file_path, 'r' ) as f:
content = f.read()
os.remove(file_path)
return content
except PermissionError:
print (f"警告:无法读取干预文件 {file_path} ,权限不足" )
return None
except IOError as e:
print (f"警告:读取干预文件失败 {file_path} : {e} " )
return None
第四步:Subagent 失败恢复机制(新增)
⚠️ 关键边界条件 :当 subagent 失败时,需要自动恢复或优雅降级。
4.1 失败检测与分类 class SubagentFailure :
"""Subagent 失败类型枚举"""
TIMEOUT = "timeout"
CRASH = "crash"
STUCK = "stuck"
CONSTRAINT_VIOLATION = "constraint_violation"
RESOURCE_ERROR = "resource_error"
def detect_failure (output_content, last_activity_time, timeout=300 ):
"""
检测 subagent 失败类型
Args:
output_content: 当前输出内容
last_activity_time: 最后活动时间
timeout: 超时阈值(秒)
Returns:
SubagentFailure | None
"""
if time.time() - last_activity_time > timeout:
return SubagentFailure.TIMEOUT
crash_patterns = [
"Traceback (most recent call last)" ,
"Segmentation fault" ,
"AssertionError" ,
"OutOfMemoryError" ,
"Process finished with exit code"
]
for pattern in crash_patterns:
if pattern in output_content:
return SubagentFailure.CRASH
if len (output_content.strip()) == 0 and time.time() - last_activity_time > 60 :
return SubagentFailure.STUCK
return None
4.2 恢复策略 def recover_from_failure (task_id, failure_type, context ):
"""
根据 failure 类型选择恢复策略
Returns:
dict: {"action": "retry|abort|escalate", "message": "..."}
"""
recovery_config = {
SubagentFailure.TIMEOUT: {
"max_retries" : 2 ,
"action" : "retry" ,
"backoff" : 5 ,
"message" : "任务超时,正在重试..."
},
SubagentFailure.CRASH: {
"max_retries" : 1 ,
"action" : "retry" ,
"backoff" : 3 ,
"message" : "检测到崩溃,尝试重启任务..."
},
SubagentFailure.STUCK: {
"max_retries" : 1 ,
"action" : "intervene" ,
"message" : "任务卡住,注入提示继续执行"
},
SubagentFailure.CONSTRAINT_VIOLATION: {
"max_retries" : 0 ,
"action" : "abort" ,
"message" : "约束违规,终止任务"
},
SubagentFailure.RESOURCE_ERROR: {
"max_retries" : 0 ,
"action" : "escalate" ,
"message" : "资源不足,需要人工介入"
}
}
config = recovery_config.get(failure_type, {"action" : "abort" , "message" : "未知错误" })
retry_count = context.get("retry_count" , 0 )
max_retries = config.get("max_retries" , 0 )
if retry_count >= max_retries:
return {
"action" : "abort" ,
"message" : f"已达到最大重试次数 ({max_retries} ),终止任务"
}
return {
"action" : config["action" ],
"message" : config["message" ],
"backoff" : config.get("backoff" , 0 ),
"retry_count" : retry_count + 1
}
4.3 自动重试流程 def execute_with_recovery (delegate_func, goal, context, max_total_retries=3 ):
"""
带自动恢复的执行流程
Args:
delegate_func: delegate_task 函数
goal: 任务目标
context: 上下文
max_total_retries: 最大总重试次数
Returns:
dict: 执行结果
"""
total_retries = 0
last_result = None
while total_retries < max_total_retries:
result = delegate_func(goal=goal, context=context)
task_id = result.get("task_id" )
monitor = SupervisorMonitor(task_id, constraints=extract_constraints(goal))
try :
monitor.start()
while monitor.is_running():
failure = detect_failure(
monitor.get_output(),
monitor.last_activity_time
)
if failure:
recovery = recover_from_failure(
task_id,
failure,
{"retry_count" : total_retries}
)
if recovery["action" ] == "retry" :
total_retries += 1
print (f"🔄 {recovery['message' ]} (重试 {total_retries} /{max_total_retries} )" )
time.sleep(recovery.get("backoff" , 5 ))
break
elif recovery["action" ] == "abort" :
return {
"status" : "failed" ,
"reason" : failure,
"message" : recovery["message" ],
"output" : monitor.get_output()
}
elif recovery["action" ] == "escalate" :
return {
"status" : "needs_attention" ,
"reason" : failure,
"message" : recovery["message" ],
"output" : monitor.get_output()
}
time.sleep(2 )
except Exception as e:
print (f"监控异常: {e} " )
total_retries += 1
continue
return {
"status" : "failed" ,
"reason" : "max_retries_exceeded" ,
"message" : f"已重试 {max_total_retries} 次仍失败" ,
"last_result" : last_result
}
4.4 失败恢复配置
recovery:
enabled: true
timeouts:
task_timeout: 3600
activity_timeout: 300
output_timeout: 60
retry_policy:
max_retries: 3
backoff_strategy: exponential
initial_backoff: 5
max_backoff: 60
escalation:
enabled: true
notify_on: [resource_error , max_retries_exceeded ]
notification_channel: "slack"
3.3 干预原则
沉默为主 :没问题不说话,减少噪音
一句话 :像用户一样直接说,禁止长篇解释
_keyinfo 只用于提前预注入 :在工作 agent 到达该步之前塞细节
已经犯错的一律用 _intervene 纠正
第五步:完成验证
4.1 检查最终结果 def verify_completion (task_id, constraints ):
"""
验证任务完成质量
"""
output_file = f"~/.hermes/tasks/{task_id} /output.txt"
result_file = f"~/.hermes/tasks/{task_id} /result.json"
with open (output_file, 'r' ) as f:
final_output = f.read()
checklist = []
for constraint in constraints:
is_satisfied = check_constraint(final_output, constraint)
checklist.append({
"constraint" : constraint,
"satisfied" : is_satisfied,
"evidence" : extract_evidence(final_output, constraint)
})
quality_score = sum (1 for item in checklist if item["satisfied" ]) / len (checklist)
return {
"quality_score" : quality_score,
"checklist" : checklist,
"recommendations" : generate_recommendations(checklist)
}
4.2 反馈给用户 def report_to_user (task_id, verification_result ):
"""
向用户报告任务完成情况
"""
score = verification_result["quality_score" ]
if score >= 0.9 :
status = "✅ 高质量完成"
elif score >= 0.7 :
status = "⚠️ 基本完成,部分约束未满足"
else :
status = "❌ 质量不达标,建议重新执行"
return f"""
{status}
质量得分:{score*100 :.1 f} %
约束检查清单:
{format_checklist(verification_result["checklist" ])}
建议:
{chr (10 ).join(f"- {r} " for r in verification_result["recommendations" ])}
"""
高级特性
特性 1:智能预判 def predict_next_step (current_step, constraints ):
"""
预测下一步可能遇到的问题,提前注入关键信息
"""
if current_step == "数据收集" and "时间戳验证" in constraints:
return "_keyinfo: 下一步需要验证数据时间戳,注意财联社返回的 publish_time 字段"
if current_step == "生成报告" and "表格格式" in constraints:
return "_keyinfo: 输出需要 Markdown 表格格式,提前准备好列标题"
return None
特性 2:异常模式库 ANOMALY_PATTERNS = {
"死循环" : {
"pattern" : r"重复执行同一操作超过 3 次" ,
"intervention" : "_intervene: 检测到可能的死循环,尝试其他方法"
},
"过度解释" : {
"pattern" : r"连续 5 轮只输出分析不执行" ,
"intervention" : "_intervene: 别分析了,直接执行"
},
"依赖缺失" : {
"pattern" : r"ModuleNotFoundError|ImportError" ,
"intervention" : "_intervene: 缺少依赖,先执行 pip install"
},
"权限不足" : {
"pattern" : r"Permission denied|Access denied" ,
"intervention" : "_intervene: 权限不足,尝试 sudo 或更换目录"
}
}
特性 3:自适应监控频率 def adjust_poll_interval (current_step, risk_level ):
"""
根据步骤风险等级调整监控频率
"""
if risk_level == "high" :
return 1
elif risk_level == "medium" :
return 2
else :
return 5
if current_step in ["数据验证" , "发布上线" , "文件写入" ]:
return 1
与 Hermes 集成方案
✅ 已完成集成(2026-04-30) 监察者模式已通过非侵入式 Wrapper 模式 集成到 Hermes delegate_task。
~/.hermes/hermes-agent/tools/supervisor_integration.py (17KB) - 集成模块
~/.hermes/config/supervisor_config.yaml (2.6KB) - 配置示例
~/.hermes/scripts/test_supervisor_integration.py (7.4KB) - 测试套件
方式 1:显式调用(推荐) from tools.supervisor_integration import delegate_task_with_supervisor
from tools.delegate_tool import delegate_task as original_delegate_task
result = delegate_task_with_supervisor(
original_delegate_task,
goal="分析今日A股大盘走势" ,
context="当前时间:2026-04-30 02:10 深夜" ,
supervisor=True ,
supervisor_config={
"sop" : "a-stock-data-acquisition-sop" ,
"risk_level" : "high" ,
"poll_interval" : 1.0
}
)
result_data = json.loads(result)
print (f"质量得分: {result_data['supervisor' ]['quality_score' ]:.1 %} " )
方式 2:自动判断 result = delegate_task_with_supervisor(
original_delegate_task,
goal="部署到生产环境" ,
context="版本 v2.0.0" ,
)
高风险:生产环境、部署、删除、A股、大盘、股票、爬取、分析
系统会自动启用监察者并调整风险等级
方式 3:Monkey Patch(全局生效)
from tools.supervisor_integration import patch_delegate_task
patch_delegate_task()
result = delegate_task(
goal="分析今日A股大盘" ,
context="..."
)
配置优先级 supervisor_config (调用时传入)
> supervisor_rules (按模式匹配)
> supervisor (全局配置)
> DEFAULT_SUPERVISOR_CONFIG
配置示例 (~/.hermes/config.yaml):
supervisor:
enabled: auto
poll_interval: 2.0
timeout: 3600
auto_intervene: true
risk_level: medium
log_interventions: true
supervisor_rules:
- pattern: "A股|大盘|股票"
config:
enabled: true
sop: a-stock-data-acquisition-sop
risk_level: high
- pattern: "生产环境|部署"
config:
enabled: true
risk_level: high
poll_interval: 1.0
集成架构 delegate_task (用户调用)
↓
delegate_task_with_supervisor (Wrapper)
↓
├─ 构建监察者配置(关键词匹配 + 配置合并)
├─ 创建 SupervisorWrapper(后台线程)
├─ 调用原始 delegate_task
└─ 合并监察者结果
↓
返回带 supervisor 字段的结果
✅ 无需修改 Hermes 核心代码
✅ 通过 Wrapper 模式包装
✅ 后台线程监控,不阻塞执行
✅ 配置驱动,灵活可扩展
方案 4:Shell Hook 自动触发(高级)
def pre_delegate_task_hook (goal, context, **kwargs ):
"""在 delegate_task 前触发"""
if should_enable_supervisor(goal, context):
kwargs["supervisor" ] = True
kwargs["supervisor_config" ] = {
"sop" : recommend_sop(goal),
"risk_level" : assess_risk(goal)
}
return kwargs
注意 :此方案需要修改 Hermes 启动流程,推荐使用方式 1-3。
最佳实践
实践 1:SOP 驱动监察 # 在 Skill 中添加监察约束
---
name: a-stock-market-analysis
constraints:
- ⚠️ 必须先获取精确时间
- ⚠️ 必须判断盘前/盘后状态
- ❌ 禁止使用非 P0 级数据源
- ✅ 必须三重校验时间戳
---
实践 2:分级监察策略 SUPERVISOR_LEVELS = {
"none" : {
"poll_interval" : None ,
"auto_intervene" : False
},
"light" : {
"poll_interval" : 10 ,
"auto_intervene" : False ,
"check_points" : ["完成" ]
},
"normal" : {
"poll_interval" : 2 ,
"auto_intervene" : True ,
"check_points" : ["启动" , "数据收集" , "分析" , "输出" ]
},
"strict" : {
"poll_interval" : 1 ,
"auto_intervene" : True ,
"check_points" : "all" ,
"pre_inject" : True
}
}
实践 3:监察日志记录 def log_supervisor_action (task_id, action_type, details ):
"""
记录监察行为,用于后续优化
"""
log_entry = {
"timestamp" : datetime.now().isoformat(),
"task_id" : task_id,
"action" : action_type,
"details" : details
}
with open (f"~/.hermes/logs/supervisor_{date} .jsonl" , 'a' ) as f:
f.write(json.dumps(log_entry) + "\n" )
故障排查
问题 1:监察过于频繁,影响性能
调高 poll_interval 到 5-10 秒
只在关键步骤检查
使用 check_points 限制检查点
问题 2:干预消息被忽略
确保 subagent 定期读取干预文件
使用 _keyinfo 立即生效而非 _intervene
检查文件路径是否正确
问题 3:约束提取不准确
在 Skill 中显式声明 constraints 字段
使用结构化格式(YAML/JSON)
添加示例和说明
参考资源
更新日志
2026-05-02 v1.2.0 : 新增边界条件处理 - 环境预检(第零步)、干预文件异常处理、Subagent 失败恢复机制(超时/崩溃/卡住/约束违规/资源错误)、自动重试流程
2026-04-30 v1.1.0 : 完成 Hermes 集成(非侵入式 Wrapper 模式),新增三种使用方式(显式调用、自动判断、Monkey Patch),测试状态 5/5 通过
2026-04-30 v1.0.0 : 初始版本,基于 GenericAgent 监察者模式移植
相关文件
核心实现 : ~/.hermes/scripts/supervisor.py (20KB)
集成模块 : ~/.hermes/hermes-agent/tools/supervisor_integration.py (17KB)
配置示例 : ~/.hermes/config/supervisor_config.yaml (2.6KB)
测试套件 : ~/.hermes/scripts/test_supervisor_integration.py (7.4KB)
使用指南 : ~/.hermes/docs/SUPERVISOR-INTEGRATION-GUIDE.md (8.2KB)
⚠️ Known Gotchas
三大红线违反
违规下场干活 : 主 Agent 直接操作
terminal("npm install package" )
write_file("config.yaml" , content)
未获取情报就干预 : 盲目纠偏
write_intervention("stop" , reason="感觉不对" )
log = file_read("~/.hermes/logs/agent.log" )
if "error" in log:
write_intervention("pause" , reason=f"发现错误: {error} " )
过度干预 : 频繁打断 Subagent
while True :
sleep(5 )
write_intervention("check" )
干预文件问题
干预文件格式错误 : Subagent 无法解析
action: stop
reason: 觉得不对
timestamp: "2024-01-01 12:00:00"
action: pause
reason: "数据源切换未按 SOP"
checkpoint: "step_3_data_fetch"
correction: "使用财联社 P0 数据源"
干预文件路径错误 : Subagent 读不到
path = "~/.hermes/intervention.txt"
path = "~/.hermes/runs/{run_id}/intervention.yaml"
干预文件冲突 : 多个 Supervisor 同时写
import fcntl
with open (intervention_file, 'w' ) as f:
fcntl.flock(f.fileno(), fcntl.LOCK_EX)
yaml.dump(intervention, f)
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
Subagent 失败恢复
超时未处理 : Subagent 卡住
subagent.run()
import signal
def timeout_handler (signum, frame ):
write_intervention("terminate" , reason="超时" )
raise TimeoutError("Subagent 超时" )
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(300 )
try :
subagent.run()
finally :
signal.alarm(0 )
崩溃无恢复 : Subagent 异常退出
try :
subagent.run()
except Exception as e:
log_error(e)
if recoverable(e):
checkpoint = get_last_checkpoint()
subagent.resume(checkpoint)
else :
notify_user(f"任务失败: {e} " )
资源泄漏 : 进程/文件未清理
import atexit
import os
def cleanup ():
if os.path.exists(temp_dir):
shutil.rmtree(temp_dir)
if subagent_process.is_alive():
subagent_process.terminate()
atexit.register(cleanup)
环境预检陷阱
依赖未检查 : 运行时才发现缺失
subagent.run()
def check_environment ():
checks = [
("python3" , "Python 3.10+" ),
("node" , "Node.js 16+" ),
("docker" , "Docker" ),
]
for cmd, desc in checks:
if not shutil.which(cmd):
return False , f"缺少 {desc} "
return True , "环境检查通过"
ok, msg = check_environment()
if not ok:
raise EnvironmentError(msg)
权限不足 : Subagent 无法执行关键操作
test -w ~/.hermes/runs
curl -I https://api.openai.com
ps aux | grep agent
SOP 遵循检查
SOP 过时 : 文档与实际不符
sop_version = get_sop_version("stock_analysis" )
if sop_version < "2.0" :
print (f"警告: SOP 版本过旧 ({sop_version} )" )
检查点遗漏 : 跳过关键步骤
if step > 3 :
pass
checkpoints = [
"step_1_time_anchor" ,
"step_2_market_status" ,
"step_3_data_fetch" ,
]
for cp in checkpoints:
if not verify_checkpoint(cp):
write_intervention("pause" , reason=f"{cp} 未完成" )
约束违规 : Subagent 违反规则
def check_constraints ():
violations = []
if uses_banned_source():
violations.append("使用禁止数据源" )
if skips_validation():
violations.append("跳过验证步骤" )
return violations
性能问题
监控频率过高 : 消耗大量资源
while True :
check_subagent()
POLL_INTERVAL = 5
while True :
check_subagent()
time.sleep(POLL_INTERVAL)
日志过大 : 占满磁盘
import logging
from logging.handlers import RotatingFileHandler
handler = RotatingFileHandler(
"supervisor.log" ,
maxBytes=10 *1024 *1024 ,
backupCount=5
)
内存泄漏 : Subagent 占用内存不释放
import psutil
def check_memory ():
process = psutil.Process(subagent_pid)
mem = process.memory_info().rss / 1024 / 1024
if mem > 1024 :
write_intervention("restart" , reason="内存泄漏" )
多 Subagent 协调
Subagent 间冲突 : 资源竞争
import threading
resource_lock = threading.Lock()
def coordinate_subagents ():
with resource_lock:
subagent.run()
依赖顺序错误 : 后置任务先执行
tasks = [
{"id" : "fetch" , "deps" : []},
{"id" : "process" , "deps" : ["fetch" ]},
{"id" : "report" , "deps" : ["process" ]},
]
for task in topological_sort(tasks):
run_subagent(task)
与 Hook 系统冲突
重复注入时间 : unified_time_awareness 已处理
hooks:
pre_llm_call:
- name: unified_time_awareness
priority: 10
- name: supervisor_precheck
priority: 20
Hook 覆盖 : 监察者模式被其他 Hook 禁用
hermes hooks list
hermes hooks enable supervisor_precheck