166 lines
6.2 KiB
Python
166 lines
6.2 KiB
Python
|
"""Hermes Agent 核心机制移植（以雅文工作台为主体的融合模块）。
|
|||
|
|
|
|||
|
|
来源与适配说明（参考 Hermes Agent 2026.8.27 源码）：
|
|||
|
|
- IterationBudget ← agent/iteration_budget.py：线程安全的轮次预算计数器，
|
|||
|
|
扩展为「运行级预算」，多个 Agent 会话共享同一个 run 的预算池。
|
|||
|
|
- jittered_backoff ← agent/retry_utils.py：抖动退避替代固定指数退避，
|
|||
|
|
防止多个会话同时撞限流时形成重试尖峰；支持解析 Retry-After。
|
|||
|
|
- 工具停滞护栏 ← agent/tool_guardrails.py：对「同名工具 + 相同参数」的
|
|||
|
|
重复调用做指纹跟踪，重复时注入提示，超限时熔断。
|
|||
|
|
|
|||
|
|
该模块只包含纯逻辑，不依赖数据库，便于单元测试。
|
|||
|
|
"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import hashlib
|
|||
|
|
import json
|
|||
|
|
import random
|
|||
|
|
import threading
|
|||
|
|
import time
|
|||
|
|
from dataclasses import dataclass, field
|
|||
|
|
from email.utils import parsedate_to_datetime
|
|||
|
|
from typing import Any
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------- 运行级轮次预算（源自 IterationBudget） ----------------
|
|||
|
|
|
|||
|
|
class IterationBudget:
|
|||
|
|
"""线程安全的运行级轮次预算。
|
|||
|
|
|
|||
|
|
一次任务运行（CampaignRun）创建一个预算池；其中的每个 Agent 会话
|
|||
|
|
（拆解/生产/质检/组合）消耗同一个池子，防止某个失控会话无限消耗
|
|||
|
|
模型调用和点数。execute_code 类可退回的调用可用 refund() 归还。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
def __init__(self, max_total: int):
|
|||
|
|
self.max_total = max(1, int(max_total))
|
|||
|
|
self._used = 0
|
|||
|
|
self._lock = threading.Lock()
|
|||
|
|
|
|||
|
|
def consume(self) -> bool:
|
|||
|
|
"""尝试消耗一轮。返回 False 表示预算耗尽，应停止调用模型。"""
|
|||
|
|
with self._lock:
|
|||
|
|
if self._used >= self.max_total:
|
|||
|
|
return False
|
|||
|
|
self._used += 1
|
|||
|
|
return True
|
|||
|
|
|
|||
|
|
def refund(self) -> None:
|
|||
|
|
with self._lock:
|
|||
|
|
if self._used > 0:
|
|||
|
|
self._used -= 1
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def used(self) -> int:
|
|||
|
|
with self._lock:
|
|||
|
|
return self._used
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def remaining(self) -> int:
|
|||
|
|
with self._lock:
|
|||
|
|
return max(0, self.max_total - self._used)
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------- 抖动退避（源自 retry_utils） ----------------
|
|||
|
|
|
|||
|
|
def parse_retry_after_seconds(value_or_headers: Any) -> float | None:
|
|||
|
|
"""把 Retry-After（数字/HTTP 日期/headers 映射）解析成非负秒数。"""
|
|||
|
|
raw = value_or_headers
|
|||
|
|
if raw is not None and not isinstance(raw, (str, int, float)):
|
|||
|
|
getter = getattr(raw, "get", None)
|
|||
|
|
if callable(getter):
|
|||
|
|
try:
|
|||
|
|
raw = getter("Retry-After")
|
|||
|
|
if raw is None:
|
|||
|
|
raw = getter("retry-after")
|
|||
|
|
except Exception:
|
|||
|
|
raw = None
|
|||
|
|
if raw is None:
|
|||
|
|
return None
|
|||
|
|
try:
|
|||
|
|
seconds = float(raw)
|
|||
|
|
except (TypeError, ValueError):
|
|||
|
|
try:
|
|||
|
|
from datetime import datetime, timezone
|
|||
|
|
parsed = parsedate_to_datetime(str(raw))
|
|||
|
|
if parsed is None:
|
|||
|
|
return None
|
|||
|
|
if parsed.tzinfo is None:
|
|||
|
|
parsed = parsed.replace(tzinfo=timezone.utc)
|
|||
|
|
seconds = (parsed - datetime.now(timezone.utc)).total_seconds()
|
|||
|
|
except Exception:
|
|||
|
|
return None
|
|||
|
|
return max(0.0, seconds)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def jittered_backoff(attempt: int, base: float = 1.6, cap: float = 24.0) -> float:
|
|||
|
|
"""全抖动（full jitter）退避：在 [0, min(cap, base*2^n)] 区间随机取值。
|
|||
|
|
|
|||
|
|
与固定 `2 ** n` 相比，多个并发会话不会在同一时刻集体重试，
|
|||
|
|
避免限流恢复瞬间被再次打挂（thundering herd）。
|
|||
|
|
"""
|
|||
|
|
ceiling = min(cap, base * (2 ** max(0, attempt)))
|
|||
|
|
return random.uniform(0, ceiling)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def retry_sleep(attempt: int, retry_after: Any = None,
|
|||
|
|
base: float = 1.6, cap: float = 24.0) -> float:
|
|||
|
|
"""决定一次重试应等待的秒数：优先尊重服务端 Retry-After，否则抖动退避。"""
|
|||
|
|
if retry_after is not None:
|
|||
|
|
seconds = parse_retry_after_seconds(retry_after)
|
|||
|
|
if seconds is not None:
|
|||
|
|
return min(max(seconds, 0.5), cap)
|
|||
|
|
return jittered_backoff(attempt, base=base, cap=cap)
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------- 工具停滞护栏（源自 tool_guardrails） ----------------
|
|||
|
|
|
|||
|
|
def tool_call_fingerprint(tool_name: str, arguments: dict[str, Any] | None) -> str:
|
|||
|
|
"""工具名 + 归一化参数生成指纹，用于识别「原样重跑」的停滞调用。"""
|
|||
|
|
try:
|
|||
|
|
canonical = json.dumps(arguments or {}, ensure_ascii=False,
|
|||
|
|
sort_keys=True, default=str)
|
|||
|
|
except (TypeError, ValueError):
|
|||
|
|
canonical = str(arguments)
|
|||
|
|
raw = f"{tool_name}\0{canonical}".encode("utf-8", errors="ignore")
|
|||
|
|
return hashlib.sha256(raw).hexdigest()
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class StallGuard:
|
|||
|
|
"""跟踪一个 Agent 会话内的工具调用指纹，识别重复停滞。
|
|||
|
|
|
|||
|
|
- 同一指纹连续出现达到 ``notice_at`` 次：返回 "notice"，运行时把提示注入上下文；
|
|||
|
|
- 达到 ``halt_at`` 次：返回 "halt"，熔断本轮循环并标记失败原因。
|
|||
|
|
只做纯决策，不产生副作用（与 Hermes 的无副作用护栏一致）。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
notice_at: int = 3
|
|||
|
|
halt_at: int = 5
|
|||
|
|
_counts: dict[str, int] = field(default_factory=dict)
|
|||
|
|
|
|||
|
|
def observe(self, fingerprint: str) -> str:
|
|||
|
|
"""记录一次调用，返回决策：ok | notice | halt。"""
|
|||
|
|
count = self._counts.get(fingerprint, 0) + 1
|
|||
|
|
self._counts[fingerprint] = count
|
|||
|
|
if count >= max(2, self.halt_at):
|
|||
|
|
return "halt"
|
|||
|
|
if count >= max(2, self.notice_at):
|
|||
|
|
return "notice"
|
|||
|
|
return "ok"
|
|||
|
|
|
|||
|
|
def halt_reason(self, tool_name: str) -> str:
|
|||
|
|
return (f"工具「{tool_name}」以相同参数连续调用 {self.halt_at} 次无进展，"
|
|||
|
|
f"已熔断本次会话以保护点数余额")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def stall_notice_text(tool_name: str, count: int) -> str:
|
|||
|
|
return (f"系统提示：工具「{tool_name}」已用完全相同的参数调用 {count} 次，"
|
|||
|
|
f"结果不会变化。请改用不同参数，或直接基于已有结果完成任务。")
|
|||
|
|
|
|||
|
|
|
|||
|
|
__all__ = [
|
|||
|
|
"IterationBudget", "parse_retry_after_seconds", "jittered_backoff", "retry_sleep",
|
|||
|
|
"tool_call_fingerprint", "StallGuard", "stall_notice_text",
|
|||
|
|
]
|