Hermes Agent Loop 中止机制全解析 #

# tools/interrupt.py
_interrupted_threads: set[int] = set()   # 被标记中断的线程 ID 集合
_lock = threading.Lock()

def set_interrupt(active: bool, thread_id: int | None = None):
    """将某个线程标记为中断/清除中断"""
    tid = thread_id or threading.current_thread().ident
    with _lock:
        if active:
            _interrupted_threads.add(tid)
        else:
            _interrupted_threads.discard(tid)

def is_interrupted() -> bool:
    """当前线程是否被标记中断"""
    return threading.current_thread().ident in _interrupted_threads

# run_agent.py:2237
def interrupt(self, message: str = None):
    """请求中断当前工具调用循环"""
    self._interrupt_requested = True     # 标志位
    self._interrupt_message = message    # 附带的新消息
    # 向当前执行线程发送中断信号
    _set_interrupt(True, self._execution_thread_id)
    # 传播给所有子 agent（delegate_task 创建的子任务）
    for child in self._running_children:
        child.interrupt(message)

# 1. 循环入口检测
while (api_call_count < agent.max_iterations and agent.iteration_budget.remaining > 0) \
        or agent._budget_grace_call:
    
    # ★ 每次循环迭代开始都检查
    if agent._interrupt_requested:
        interrupted = True
        _turn_exit_reason = "interrupted_by_user"
        break
    
    api_call_count += 1
    
    # ... 调用 LLM、执行工具调用、处理结果 ...
    
    # 2. 工具调用结果处理中也检查（约 1569 行）
    if agent._interrupt_requested:
        break
    
    # 3. 资源清理前检查（约 2706 行）
    if agent._interrupt_requested:
        break

def interruptible_api_call(agent, api_kwargs):
    """在后台线程中执行 API 调用，允许主循环检测中断"""
    result = {"response": None, "error": None}
    
    def _call():
        """实际 API 调用在子线程中运行"""
        try:
            client = _set_request_client(create_client(...))
            result["response"] = client.chat.completions.create(**api_kwargs)
        except Exception as e:
            result["error"] = e
    
    # ★ 启动后台线程
    worker = threading.Thread(target=_call)
    worker.start()
    
    # ★ 主线程轮询中断信号
    while worker.is_alive():
        if agent._interrupt_requested:
            # 关闭 HTTP 连接，让后台线程的 recv() 收到 EPIPE
            _close_request_client_once("interrupt_abort")
            break
        worker.join(timeout=0.5)  # 每 0.5 秒检查一次
    
    return result

# 任何工具内部都可以这样检测
from tools.interrupt import is_interrupted

if is_interrupted():
    return json.dumps({"output": "[interrupted]", "returncode": 130})

def steer(self, text: str):
    """注入消息到下一个 tool result 中，不打断当前执行"""
    self._steer_message = text
    # 注意：这里不调用 interrupt()