ag2-observers-and-alerts

オブザーバー、監視、アラート

使用する場面

• 可観測性 — モデルの応答、ツール呼び出し、トークン使用量をログに記録します。
• ランタイムの安全性 — 危険なツール引数をブロックし、エージェントを停止します。
• リアクティブなメトリクス — N 回目の応答ごと、または M 秒ごとに処理を実行します。
• ループ／繰り返しの検出 — 無限のツール呼び出しループを検出します。
• 状態を保持する監視 — 次に何をするか判断するために、過去のイベントを記憶する必要がある処理全般に使います。

2 種類のオブザーバー

形式	使用場面	使用方法
ステートレス関数	単発のイベントフック（ログ記録、メトリクス）	@observer(EventType)
ステートフルクラス	カウンター／ウィンドウ／しきい値／複合トリガー	BaseObserver をサブクラス化

どちらも内部ではストリームのサブスクライバーとして動作し、ストリームに直接ではなくエージェントに登録されます。

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 にわたる累積トークン数を追跡します。しきい値を超えると 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)
遅延後に 1 回	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 はストリーム上にあり、履歴にも保存されますが、デフォルトのプロバイダーマッパーは LLM に再表示しません。エージェントにアラートを認識させるには、assembly=[...] に AlertPolicy() を追加します。

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

FATAL アラート → HaltEvent → 短絡

AlertPolicy は Severity.FATAL に対して 2 つの処理を行います。

1. ストリーム上に HaltEvent を発行します。
2. システムプロンプトに停止通知を追加します。

assembly=[...] が空でない場合、ハーネスは _HaltCheckMiddleware を自動的に設定します。このミドルウェアは HaltEvent を検知し、次の LLM 呼び出しを合成された HALTED: ... 応答で短絡します。

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 が発生して停止し、エージェントの次の 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)

オブザーバー、ミドルウェア、ストリームサブスクライバーの比較

機能	オブザーバー	ミドルウェア	ストリームサブスクライバー
登録先	エージェント	エージェント	ストリーム
ライフサイクル	実行単位	実行単位	手動
定型コード	関数（または BaseObserver）	BaseMiddleware クラス	関数
イベントを変更できるか	interrupt=True	はい（実行をラップ）	interrupt=True
DI のサポート	はい	はい	はい
ユースケース	監視、メトリクス、アラート	横断的な処理（再試行、認証、レート制限）	低レベルのイベント配線

さらに詳しく

• assets/token_watchdog.py — 1 つのエージェントで 3 つのオブザーバー（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 イベントはストリーム上にありますが、デフォルトでは LLM から見えません。assembly=[...] に AlertPolicy() を追加してください。
• FATAL でも停止しない — HaltEvent を作成するのは AlertPolicy です。assembly=[..., AlertPolicy(), ...]（または _HaltCheckMiddleware を有効にする空でないアセンブリチェーン）がなければ、何も停止しません。
• 複数のエージェントで同じ AlertPolicy() を共有する — 重複排除の状態はインスタンスに保持されます。エージェントごとに独自のインスタンスを用意してください。
• Watch コールバックが events は空でないと想定する — 時間駆動の Watch（DelayWatch、IntervalWatch、CronWatch）では、events は常に [] です。
• process() が非同期であることを忘れる — BaseObserver.process は async def でなければなりません。
• subscribe() デコレーターを使いたいのに subscribe(fn) で購読する — どちらも機能します。関数を直接渡す形式は stream.subscribe(fn)、デコレーター形式は @stream.subscribe()（括弧付き）です。
• n も max_wait も指定せずに CadenceWatch を使う — ValueError が発生します。少なくとも一方が必要です。