ag2-knowledge-and-memory

ナレッジ、メモリ、コンテキストの組み立て

このスキルでは、連携して動作する3つの関連プリミティブについて説明します。

プリミティブ	所属先	役割
KnowledgeStore	ag2.knowledge	パスベースの永続ストレージ（メモリ／sqlite／ディスク／redis）
アセンブリポリシー	ag2.policies	LLM 呼び出しの前に、各ターンの (prompts, events) の形を整えます
集約／圧縮	ag2.aggregate / .compact	構造化されたナレッジをストアに書き込む／イベント履歴を削減する

KnowledgeConfig は、knowledge= コンストラクターパラメーターを介して3つすべてを Agent に設定します。アセンブリポリシーは assembly= を介して設定します。

用途に応じた選び方

ユーザーの意図	使用するもの
会話をまたいでユーザーの好みや状態を記憶する	WorkingMemoryAggregate + WorkingMemoryPolicy（および永続ストア）
次回のために各セッションを要約する	ConversationSummaryAggregate + EpisodicMemoryPolicy
LLM に送信するイベント履歴に厳密な上限を設ける	SlidingWindowPolicy(max_events=N)
おおよそのトークン数に基づいて上限を設ける	TokenBudgetPolicy(max_tokens=N)
LLM から見えるイベントからライフサイクル／オブザーバーイベントを除外する	ConversationPolicy()
ストリーム履歴を削減する（LLM から見える範囲だけではありません）	TailWindowCompact または SummarizeCompact
オブザーバーのアラートを LLM にルーティングする	AlertPolicy()

60秒でできるレシピ — 永続的なワーキングメモリ

from ag2 import Agent, KnowledgeConfig
from ag2.aggregate import AggregateTrigger, WorkingMemoryAggregate
from ag2.config import OpenAIConfig
from ag2.knowledge import DiskKnowledgeStore
from ag2.policies import ConversationPolicy, WorkingMemoryPolicy

store = DiskKnowledgeStore("./journal-state")
config = OpenAIConfig(model="gpt-5")

agent = Agent(
    "journal",
    prompt="You are a daily journal companion.",
    config=config,
    knowledge=KnowledgeConfig(
        store=store,
        aggregate=WorkingMemoryAggregate(config=config),
        aggregate_trigger=AggregateTrigger(on_end=True),
    ),
    assembly=[
        WorkingMemoryPolicy(),  # injects /memory/working.md on every LLM call
        ConversationPolicy(),
    ],
)

各会話の後、集約処理によって /memory/working.md に書き込まれます。次回、同じストアを使って Agent を構築すると、WorkingMemoryPolicy がそのファイルを読み込み、プロンプトのコンテキストとして挿入します。チャット履歴を再生しなくても、エージェントが「記憶」します。実行可能な完全版の例：assets/journal_companion.py。

注意：knowledge=KnowledgeConfig(store=...) を設定すると、LLM に knowledge ツールが渡され、SKILL.md ファイルがストアにシードされ（そのツールを使うようモデルに指示します）、各ターンのイベントが /log/ に書き出されます。独自のメモリを管理するジャーナルコンパニオンには適していますが、モデルに公開せずにポリシーからストアを利用できるようにするには、expose_tool=False, write_event_log=False を渡してください。Agent へのすべての設定にある「"KnowledgeConfig(store=...) は4つの処理を行います — そのうち3つはオプトアウト可能になりました」を参照してください。

KnowledgeStore の実装

実装	使用する場面
MemoryKnowledgeStore()	テスト、一時的なセッション
SqliteKnowledgeStore(path)	単一プロセスでの永続性 — 実用的なデフォルト
DiskKnowledgeStore(root)	ファイルをディスク上で人間が読み取れる形式にしたい場合 — 第1引数は root です。watchdog の追加パッケージが必要です（on_change ウォッチャーが使用します）。watchdog がインストールされていない場合、クラスのインポート時に依存関係の不足を示すエラーが発生します。
RedisKnowledgeStore(url_or_client)	複数プロセス間／ホスト間での共有 — 第1引数は url_or_client です。Redis の URL 文字列、または構築済みの redis.asyncio クライアントを渡します。URL 文字列を使う場合は、redis パッケージも必要です。
LockedKnowledgeStore(store, lock)	任意のストアをラップして、同時書き込みを直列化します — 第1引数は store、第2引数は lock です（どちらも位置引数）。読み取りはロックなしで通過し、write／delete／append のみがロックを取得します。

API（すべて非同期）：

await store.write("/artifacts/report.md", "# Q3...")
text = await store.read("/artifacts/report.md")
children = await store.list("/")            # immediate children, dirs end in '/'
await store.delete("/artifacts/old.md")
exists = await store.exists("/artifacts/report.md")

off = await store.append("/log/events.jsonl", '{"t":1}\n')   # WAL-style
new_slice = await store.read_range("/log/events.jsonl", off)  # only new bytes
sub = await store.on_change("/log/", on_change_callback)

アセンブリチェーン — LLM が実際に見るもの

assembly=[...] を介して AssemblyPolicy インスタンスを渡します。Agent は、最外部のミドルウェア位置に内部の AssemblerMiddleware を設定します。各ポリシーは (prompts, events) を変換して、次のポリシーに渡します。

ポリシーには2種類あります — 順序が重要です。削減より先に挿入を行ってください。

種類	目的	組み込み
挿入	prompts に追加する	WorkingMemoryPolicy、EpisodicMemoryPolicy、AlertPolicy
削減	events を削減する	ConversationPolicy、SlidingWindowPolicy、TokenBudgetPolicy

順序は手動で検証してください。

from ag2.assembly import AssemblerMiddleware
warnings = AssemblerMiddleware.validate_order(policies)  # returns list of warnings on known bad orderings

（AssemblerMiddleware と AssemblyPolicy プロトコルは、高度な用途や手動でのハーネス設定のために ag2.assembly にあります。assembly=[...] を介して組み込みポリシーを渡すだけの場合、インポートする必要はありません。）

組み込みポリシー

from ag2.policies import (
    AlertPolicy,
    ConversationPolicy,
    EpisodicMemoryPolicy,
    SlidingWindowPolicy,
    TokenBudgetPolicy,
    WorkingMemoryPolicy,
)

# Injection
WorkingMemoryPolicy()                                 # reads /memory/working.md
EpisodicMemoryPolicy(max_episodes=5, transparent=True) # reads recent /memory/conversations/
AlertPolicy()                                          # delivers ObserverAlerts to LLM, halts on FATAL

# Reduction
ConversationPolicy()                                  # drops non-conversation events
SlidingWindowPolicy(max_events=50, transparent=True)  # last N events
TokenBudgetPolicy(max_tokens=32_000, chars_per_token=4, transparent=True)

transparent=True を指定すると、プロンプトに [policy_name] Showing X of Y events. という注記が追加されます。調整時に便利です。実際のチェーンの例：

assembly=[
    WorkingMemoryPolicy(),
    EpisodicMemoryPolicy(max_episodes=3),
    AlertPolicy(),
    SlidingWindowPolicy(max_events=80),
]

集約 — ナレッジをストアに書き込む

AggregateStrategy.aggregate(events, ctx, store) → None は情報を抽出して永続化します。組み込みの戦略は 2 つあり、どちらも要約呼び出し用の ModelConfig を受け取ります（エージェントのメインモデルより安価なモデルを使用してください）。

戦略	書き込む内容	組み合わせるポリシー
WorkingMemoryAggregate(config=..., prompt="…")	/memory/working.md（単一のローリングファイル）。prompt= でマージテンプレート（{existing} / {events} プレースホルダー）を上書きします	WorkingMemoryPolicy
ConversationSummaryAggregate(config=...)	/memory/conversations/{ts}_{stream_id}.md	EpisodicMemoryPolicy

AggregateTrigger で実行間隔を制御します。every_n_turns、every_n_events、on_end を指定できます。AggregateTrigger() だけでは何も実行されないため、少なくとも 1 つを有効にしてください。on_end=True は、実行のたびに LLM 呼び出しが発生するため、デフォルトでは無効です。

圧縮 — ストリーム履歴を短縮する

CompactStrategy.compact(events, ctx, store) → list[BaseEvent] は、ストリームの履歴を置き換えます。組み込みの戦略は 2 つあります。

戦略	動作	コスト
TailWindowCompact(target=N)	直近の N 個のイベントを保持し、残りを削除します（必要に応じて /log/ に永続化します）	LLM 呼び出しなし
SummarizeCompact(target=N, config=...)	削除するイベントを 1 つの CompactionSummary に要約し、先頭に挿入します	実行ごとに LLM 呼び出し 1 回

CompactTrigger(max_events=N, max_tokens=M, chars_per_token=4) は、いずれかのしきい値を超えると実行されます。

from ag2.compact import CompactTrigger, TailWindowCompact, SummarizeCompact

SummarizeCompact は CompactionSummary イベントを先頭に挿入します。ConversationPolicy はこれを通過させるため、LLM は引き続きそのコンテキストを受け取れます。

すべてを Agent に組み込む

KnowledgeConfig はこれらをまとめた設定です。

from dataclasses import dataclass

@dataclass
class KnowledgeConfig:
    store: KnowledgeStore
    expose_tool: bool = True               # auto-inject the `knowledge` LLM tool?
    write_event_log: bool = True           # dump each turn's events to /log/{stream_id}.jsonl?
    compact: CompactStrategy | None = None
    compact_trigger: CompactTrigger | None = None
    aggregate: AggregateStrategy | None = None
    aggregate_trigger: AggregateTrigger | None = None
    bootstrap: StoreBootstrap | None = None    # default: DefaultBootstrap(mention_tool=expose_tool)

KnowledgeConfig(store=...) は4つの機能を提供します。そのうち3つはオプトアウトできるようになりました

knowledge=KnowledgeConfig(store=...) を渡すと、4つの機能がまとめて有効になります。デフォルトでは従来の動作が維持されますが、expose_tool フラグと write_event_log フラグを使うと、大部分を無効にできます。

1. ストアを登録するため、アセンブリポリシー（WorkingMemoryPolicy、…）から context.dependencies.get(KnowledgeStore) を利用できます。（通常はこれだけが必要です。これが本来の目的であるため、この機能にはフラグがありません。）
2. エージェントのツール一覧に knowledge アクショングループツール（knowledge(action="write"/"read"/"list"/...)）を自動的に追加します。expose_tool=False で無効にできます。
3. bootstrap= を明示的に渡さない限り、初回の ask 時に SKILL.md ファイル（/SKILL.md、/artifacts/SKILL.md、/memory/SKILL.md、/log/SKILL.md）をストアに登録します。デフォルトは DefaultBootstrap(mention_tool=expose_tool) です。そのため、expose_tool=False にすると、登録される SKILL.md で LLM に「knowledge ツールを使う」よう促す記述も自動的になくなります（使うツールがないためです）。登録を一切行わない場合は、独自の何もしない StoreBootstrap（async def bootstrap(self, store, actor_name): pass を持つクラス）を bootstrap= に渡してください。組み込みの NoBootstrap はありません。
4. ターンイベントを /log/ に永続化します。各ターンの後、エージェントのイベント履歴全体が /log/{stream_id}.jsonl に書き出されます。write_event_log=False で無効にできます。永続化に失敗すると、ストリームで EventLogFailed イベントが発行されます（あわせて logger.exception も呼び出されます）。

つまり、「ポリシーからはストアが見えるが、LLM からは見えず、/log/ のファイルもなく、SKILL.md ファイルもない」状態にするには、次のようにします。

agent = Agent(
    "summarizer",
    config=config,
    knowledge=KnowledgeConfig(
        store=DiskKnowledgeStore("./agent-state"),
        expose_tool=False, write_event_log=False,   # no `knowledge` tool, no /log/ dump
        aggregate=WorkingMemoryAggregate(config=cheap_config),  # aggregation/compaction wiring still works
        aggregate_trigger=AggregateTrigger(on_end=True),
    ),
    assembly=[WorkingMemoryPolicy()],   # reads /memory/working.md from context.dependencies
)

SKILL.md の登録も完全になくしたい場合は、独自の何もしない StoreBootstrap である NoBootstrap を使い、bootstrap=NoBootstrap() を追加してください。以前からある別の方法として、KnowledgeConfig を使わず、通常の Agent(..., dependencies={KnowledgeStore: store}) としてストアを登録する方法も引き続き使えます。これが最小構成です（ツールなし、ブートストラップなし、/log/ なし、さらに集約／コンパクトの連携もありません）。たとえば、別の「リフレクター」パスから store.write("/memory/working.md", ...) を自分で実行したい場合に使ってください。

全体の構成は次のとおりです。

agent = Agent(
    "assistant",
    config=main_config,
    knowledge=KnowledgeConfig(
        store=DiskKnowledgeStore("./state"),
        compact=TailWindowCompact(target=100),
        compact_trigger=CompactTrigger(max_events=200),
        aggregate=ConversationSummaryAggregate(config=summarizer_config),
        aggregate_trigger=AggregateTrigger(every_n_turns=10, on_end=True),
        # expose_tool / write_event_log default True; bootstrap defaults to
        # DefaultBootstrap(mention_tool=expose_tool). Set explicitly only to opt out.
    ),
    assembly=[
        WorkingMemoryPolicy(),
        EpisodicMemoryPolicy(max_episodes=3),
        AlertPolicy(),
        SlidingWindowPolicy(max_events=80),
    ],
)

ハーネスは、内部ミドルウェア（_AssemblerMiddleware、_HaltCheckMiddleware、_CompactionMiddleware、_AggregationMiddleware）を条件に応じて組み込みます。有効にした機能の分だけ負荷がかかります。

エージェントのストリームで発行されるライフサイクルイベント：

• CompactionStarted → CompactionCompleted（events_before／events_after／usage）または CompactionFailed（例外）
• AggregationStarted → AggregationCompleted（strategy／usage）または AggregationFailed（例外）
• /log/ のイベントストリームの永続化で例外が発生した場合の EventLogFailed
• AlertPolicy が FATAL アラートを検出した場合の HaltEvent

*Failed イベントは、永続的で確認可能な通知です。ハーネスはこれらのイベントについても logger.exception を呼び出しますが、問題の発生を確認するために Python のロギングを設定する必要はありません。ストリームを購読してください（ag2-observers-and-alerts を参照）。

さらに詳しく

• assets/journal_companion.py — 実行可能なエンドツーエンドの作業メモリのデモ（code_examples/06 に対応）。
• assets/long_doc_chat.py — アセンブリとコンパクトの負荷テスト（code_examples/07 に対応）。
• ソースドキュメント：
  • website/docs/user-guide/advanced/knowledge_store.mdx — ストア API、EventLogWriter、LockedKnowledgeStore。
  • website/docs/user-guide/advanced/assembly.mdx — ポリシーの詳細なリファレンスと順序に関するルール。
  • website/docs/user-guide/advanced/aggregation.mdx — 集約戦略とカスタム戦略。
  • website/docs/user-guide/advanced/compaction.mdx — コンパクト戦略とカスタム戦略。
  • website/docs/user-guide/agent_harness.mdx — KnowledgeConfig コンストラクターのリファレンス、ターンライフサイクルにおけるミドルウェアの順序。

よくある落とし穴

• 注入前に削減してしまう — WorkingMemoryPolicy より前に SlidingWindowPolicy を置くと、作業メモリの注入が予算にカウントされません。必ず、注入を最初に行い、次に AlertPolicy、最後に削減を行ってください。
• メモリポリシーに KnowledgeStore 依存関係を登録し忘れる — WorkingMemoryPolicy と EpisodicMemoryPolicy は、context.dependencies.get(KnowledgeStore) 経由でストアを検索します。KnowledgeConfig(store=...) を使えば自動で登録されます。ポリシーを手動で設定する場合は、dependencies にもストアを登録してください。
• 集約のたびに LLM 呼び出しが発生する — すべての会話で on_end=True にすると、呼び出し回数が積み重なります。WorkingMemoryAggregate と ConversationSummaryAggregate は慎重に組み合わせてください。大量のエージェントでは every_n_turns=N の使用を検討してください。
• HistoryLimiter ミドルウェアとアセンブリの削減ポリシーを併用する — どちらも履歴を削ります。どちらか一方の方式を選んでください。アセンブリのほうが柔軟で（詳細な整形や透明性に関する注記が可能）、HistoryLimiter はよりシンプルです。
• read_range は文字オフセットではなく、バイトオフセットを扱う — マルチバイトの UTF-8 シーケンスでは、オフセットを慎重に合わせる必要があります。
• WorkingMemoryAggregate が破壊的であることを忘れる — 実行のたびに /memory/working.md を上書きします。これは意図的な動作です（ログではなく、継続的に更新される状態です）が、以前の内容が統合されたり消えたりすることを想定してください。
• AlertPolicy を assembly= に指定せず、アラートが LLM に表示されると期待する — アラートは ObserverAlert イベントとしてストリーム上にありますが、AlertPolicy が注入した場合にのみ LLM に届きます。
• KnowledgeConfig(store=...) は「単なるストレージハンドル」ではありません — knowledge LLM ツールを自動で注入し、SKILL.md ファイルをストアにシードし、各ターンのイベントをすべて /log/ にダンプします。これらを無効にするには、KnowledgeConfig(store=..., expose_tool=False, write_event_log=False) を使います（デフォルトのブートストラップでもツールへの言及がなくなります。mention_tool のデフォルト値は expose_tool だからです）。または、KnowledgeConfig を使わず、最低限の構成として Agent(..., dependencies={KnowledgeStore: store}) を使ってください。上記の「KnowledgeConfig(store=...) は 4 つの処理を行い、そのうち 3 つは現在オプトアウト可能」を参照してください。
• WorkingMemoryAggregate のデフォルトのプロンプトは、内容を重視します — 初期設定では、戦略ではなく、会話が「何についてだったか」（「重要な既存のコンテキストを維持し、古くなった情報を削除する」）を保持します。そのため、戦術（どの表現や分野がうまくいったか）をメモリに記録したいリサーチエージェントでは、古くなったトピック情報が残ります。WorkingMemoryAggregate(config=..., prompt="…") を渡して上書きしてください。テンプレートには {existing}（現在の作業メモリ）と {events}（新しい会話）が補間されます。「同じ形式で別のプロンプト」より大幅な変更が必要な場合は、カスタムの AggregateStrategy.aggregate(events, ctx, store) を作成するか、リフレクターパスから直接ファイルに書き込んでください。
• カスタム戦略のデバッグでは *Failed イベントに注意する — カスタムの AggregateStrategy／CompactStrategy（またはそのトリガー）が例外を発生させると、ハーネスは logger.exception で記録し、例外を含む AggregationFailed／CompactionFailed をストリーム上に出力します。/log/ への書き込みに失敗した場合は、EventLogFailed も出力されます。そのため、Python ロギングを設定しなくても、「トリガーが発火しなかった」のか「戦略が例外を発生させた」のかを区別できます。購読するだけです（ag2-observers-and-alerts）。トリガーが実際に発火したことを確認するための AggregationStarted／CompactionStarted も受け取れます。
• ストアのファイルシステムルートとストア内パスの階層 — DiskKnowledgeStore("./memory") を使うと、ストアのルートは ./memory になり、ストア内の作業メモリのパスは /memory/working.md なので、ファイルは ./memory/memory/working.md に配置されます。「memory の中に memory」ができる混乱を避けるため、ファイルシステムのルートには ./agent-state や ./journal-state など、異なる名前を付けてください。