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",
|
||
]
|