ag2-observers-and-alerts

观察器、监视与告警

适用场景

• 可观测性 — 记录模型响应、工具调用和 token 用量。
• 运行时安全 — 阻止危险的工具参数，停止 Agent。
• 响应式指标 — 每 N 次响应或每 M 秒触发一次。
• 循环／重复检测 — 捕获无限工具调用循环。
• 有状态监控 — 需要记住先前事件，以决定下一步操作的任何场景。

两种观察器形式

形式	适用场景	使用方式
无状态函数	一次性事件钩子（日志记录、指标）	@observer(EventType)
有状态类	计数器／窗口／阈值／组合触发器	子类化 BaseObserver

两者在底层都是流订阅者 — 注册在 Agent 上，而非直接注册到流上。

60 秒快速上手 — @observer

from ag2 import Agent, observer
from ag2.config import OpenAIConfig
from ag2.events import ModelResponse

@observer(ModelResponse)
async def log_response(event: ModelResponse) -> None:
    print(f"Model said: {event.content}")

agent = Agent(
    "assistant",
    config=OpenAIConfig(model="gpt-4o-mini"),
    observers=[log_response],
)

也可以在构造完成后使用 @agent.observer(...) 附加观察器。也支持针对单次调用的观察器（agent.ask("...", observers=[...])）。

观察器回调支持完整的依赖注入（Context、Inject、Variable、Depends）。可按事件类型、多个类型（ModelRequest | ModelResponse）或字段值（ToolCallEvent.name == "search"）筛选。使用 interrupt=True 可在常规订阅者看到事件之前修改或抑制事件。

内置有状态观察器

from ag2 import Agent
from ag2.observers import LoopDetector, TokenMonitor

agent = Agent(
    "assistant",
    config=config,
    observers=[
        TokenMonitor(warn_threshold=50_000, alert_threshold=100_000),
        LoopDetector(window_size=10, repeat_threshold=3),
    ],
)

• TokenMonitor — 跟踪 ModelResponse 和 TaskCompleted 中的累计 token 数。达到阈值时发出 WARNING／CRITICAL 级别的 ObserverAlert。通过 monitor.total_tokens 读取状态。
• LoopDetector — 对近期工具调用使用滑动窗口。检测到连续 repeat_threshold 次相同调用时发出 WARNING 告警。

自定义 BaseObserver

BaseObserver 将一个 Watch（何时触发）与一个 process() 方法（要执行什么操作）组合在一起：

from ag2 import Context
from ag2.observers import BaseObserver
from ag2.watch import CadenceWatch
from ag2.events import BaseEvent, ModelResponse
from ag2.events.alert import ObserverAlert, Severity

class AvgCompletionObserver(BaseObserver):
    """Every N responses, emit an INFO alert with avg completion-token count."""

    def __init__(self, window: int = 5) -> None:
        super().__init__("avg-completion", watch=CadenceWatch(n=window, condition=ModelResponse))
        self._window = window

    async def process(self, events: list[BaseEvent], ctx: Context) -> ObserverAlert | None:
        tokens = [e.usage.completion_tokens for e in events if isinstance(e, ModelResponse) and e.usage]
        if not tokens:
            return None
        return ObserverAlert(
            source=self.name,
            severity=Severity.INFO,
            message=f"Avg completion tokens over last {self._window}: {sum(tokens) / len(tokens):.0f}",
        )

如果 process() 返回 ObserverAlert，基类会将其发送到流上。你也可以通过 await ctx.send(...) 手动发送事件。

Watch 基元 — 选择何时触发

需求	使用方式
每次匹配的事件	EventWatch(EventType) 或直接使用 stream.subscribe(fn, condition=...)
每 N 次匹配事件	CadenceWatch(n=N, condition=EventType)
每 T 秒（缓冲事件）	CadenceWatch(max_wait=T, condition=EventType)
满足任一阈值	CadenceWatch(n=N, max_wait=T, condition=EventType)
延迟后触发一次	DelayWatch(seconds)
定时周期触发	IntervalWatch(seconds)
Cron 计划	CronWatch("0 9 * * MON")
所有子监视都触发	AllOf(w1, w2)
任一子监视触发	AnyOf(w1, w2)
按顺序触发	Sequence(w1, w2)

均可从 ag2.watch 导入。回调签名统一为：async def cb(events: list[BaseEvent], ctx: Context) -> None。由时间驱动的监视会传入 events=[]。

ObserverAlert — 告警类型

from ag2.events.alert import ObserverAlert, Severity

ObserverAlert(
    source="my-observer",
    severity=Severity.WARNING,    # INFO, WARNING, CRITICAL, FATAL
    message="What happened",
)

重要提示： ObserverAlert 会出现在流上并保存在历史记录中，但默认的提供方映射器不会将其呈现给大模型。要让 Agent 看到告警，请将 AlertPolicy() 添加到 assembly=[...] 中：

from ag2.policies import AlertPolicy
agent = Agent("assistant", config=config, assembly=[AlertPolicy()])

FATAL 告警 → HaltEvent → 短路

AlertPolicy 在 Severity.FATAL 时会执行两项操作：

1. 在流上发出 HaltEvent。
2. 将停止通知附加到系统提示词中。

当 assembly=[...] 非空时，运行框架会自动接入 _HaltCheckMiddleware，它会检测 HaltEvent，并通过合成的 HALTED: ... 响应短路下一次 LLM 调用。

from ag2 import Context
from ag2.observers import BaseObserver
from ag2.events import BaseEvent, ToolCallEvent
from ag2.events.alert import HaltEvent, ObserverAlert, Severity
from ag2.policies import AlertPolicy
from ag2.watch import EventWatch

class PathGuardian(BaseObserver):
    def __init__(self) -> None:
        super().__init__("path-guardian", watch=EventWatch(ToolCallEvent))

    async def process(self, events: list[BaseEvent], ctx: Context) -> ObserverAlert | None:
        for event in events:
            if not isinstance(event, ToolCallEvent) or event.name != "write_file":
                continue
            if "/etc/" in event.arguments or "/usr/" in event.arguments:
                return ObserverAlert(
                    source=self.name,
                    severity=Severity.FATAL,
                    message=f"blocked dangerous write: {event.arguments}",
                )
        return None

agent = Agent(
    "safe-shell",
    prompt="...",
    config=config,
    tools=[write_file],
    observers=[PathGuardian()],
    assembly=[AlertPolicy()],   # routes FATAL → HaltEvent
)

首次危险的工具调用会触发 FATAL → 停止；Agent 的下一次 ask 会被短路。完整的可运行演示：assets/safety_guard.py。

从外部订阅告警和停止事件

from ag2 import MemoryStream
from ag2.events.alert import HaltEvent, ObserverAlert

stream = MemoryStream()
stream.where(ObserverAlert).subscribe(lambda e: print(f"[{e.severity}] {e.source}: {e.message}"))
stream.where(HaltEvent).subscribe(lambda e: print(f"HALT: {e.reason}"))
await agent.ask("...", stream=stream)

观察器 vs 中间件 vs Stream 订阅者

特性	观察器	中间件	Stream 订阅者
注册对象	Agent	Agent	Stream
生命周期	限定于执行期间	限定于执行期间	手动
样板代码	函数（或 BaseObserver）	BaseMiddleware 类	函数
能否修改事件	interrupt=True	可以（包装执行过程）	interrupt=True
支持 DI	是	是	是
用例	监控、指标、告警	横切关注点（重试、身份验证、速率限制）	底层事件接线

深入了解

• assets/token_watchdog.py — 一个 Agent 上的三个观察器（TokenMonitor、LoopDetector、自定义 AlertConsole）。与 code_examples/04 对应。
• assets/safety_guard.py — PathGuardian → FATAL → AlertPolicy → HaltEvent → 短路。与 code_examples/08 对应。
• 源文档：
  • website/docs/user-guide/advanced/observers.mdx — @observer、BaseObserver、注册、内置观察器、ObserverAlert。
  • website/docs/user-guide/advanced/watches.mdx — 所有 Watch 原语及组合规则。
  • website/docs/user-guide/advanced/stream.mdx — Stream API、where、subscribe、中断器、RedisStream。
  • website/docs/user-guide/advanced/assembly.mdx — AlertPolicy 的排序和去重。

常见问题

• 告警没有传达到模型 — ObserverAlert 事件会出现在 Stream 上，但默认对 LLM 不可见。将 AlertPolicy() 添加到 assembly=[...]。
• FATAL 未能停止执行 — AlertPolicy 负责创建 HaltEvent。如果没有 assembly=[..., AlertPolicy(), ...]（或任何能启用 _HaltCheckMiddleware 的非空装配链），执行就不会停止。
• 在多个 Agent 间共享同一个 AlertPolicy() — 去重状态保存在实例上。每个 Agent 都应使用自己的实例。
• Watch 回调假定 events 非空 — 对于基于时间驱动的 Watch（DelayWatch、IntervalWatch、CronWatch），events 始终为 []。
• 忘记 process() 是异步的 — BaseObserver.process 必须是 async def。
• 想用 subscribe() 装饰器，却使用了 subscribe(fn) — 两种方式都可用；直接调用形式是 stream.subscribe(fn)，装饰器形式是 @stream.subscribe()（带括号）。
• 使用 CadenceWatch 时既未设置 n，也未设置 max_wait — 会引发 ValueError；至少需要设置其中一个。