ag2-knowledge-and-memory

知识、记忆与上下文组装

本技能介绍三种相互关联、协同工作的基本组件：

基本组件	所在位置	作用
KnowledgeStore	ag2.knowledge	基于路径的持久化存储（memory / sqlite / disk / redis）
组装策略	ag2.policies	在调用大模型前，为每轮调整 (prompts, events)
聚合 / 压缩	ag2.aggregate / .compact	将结构化知识写入存储 / 精简事件历史

KnowledgeConfig 通过 Agent 的 knowledge= 构造参数将这三者连接起来；组装策略则通过 assembly= 传入。

何时使用什么

用户意图	使用
在不同对话间记住用户偏好 / 状态	WorkingMemoryAggregate + WorkingMemoryPolicy（以及持久化存储）
总结每次会话，供下次使用	ConversationSummaryAggregate + EpisodicMemoryPolicy
严格限制发送给大模型的事件历史	SlidingWindowPolicy(max_events=N)
按近似 token 数量限制	TokenBudgetPolicy(max_tokens=N)
从大模型视图中移除生命周期 / 观察者事件	ConversationPolicy()
精简流历史（不只是大模型视图）	TailWindowCompact 或 SummarizeCompact
将告警观察者路由给大模型	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 会读取该文件，并将其作为提示词上下文注入。Agent 无需重放聊天历史，就能“记住”信息。完整的可运行示例：assets/journal_companion.py。

提醒：knowledge=KnowledgeConfig(store=...) 还会向大模型提供一个 knowledge 工具、将 SKILL.md 文件写入存储（这些文件会告诉模型使用该工具），并将每轮事件转储到 /log/。对于自行管理记忆的日志伴侣来说，这样做很合适；但如果你希望策略能够使用存储，而不将其暴露给模型，请传入 expose_tool=False, write_event_log=False。请参阅 将所有内容连接到 Agent 下的 “KnowledgeConfig(store=...) 有四项作用——现在其中三项可以选择关闭”。

KnowledgeStore 实现

实现	适用场景
MemoryKnowledgeStore()	测试、临时会话
SqliteKnowledgeStore(path)	单进程持久化——实用的默认选择
DiskKnowledgeStore(root)	需要让文件在磁盘上可供人阅读——第一个参数是 root；需要安装 watchdog 扩展（on_change 监视器会用到）。若未安装 watchdog，导入该类时会引发缺少依赖项错误。
RedisKnowledgeStore(url_or_client)	多进程 / 跨主机共享——第一个参数是 url_or_client：传入 Redis URL 字符串，或已创建的 redis.asyncio 客户端。使用 URL 字符串时也需要安装 redis 包。
LockedKnowledgeStore(store, lock)	包装任意存储，以串行化并发写入——第一个参数是 store，第二个参数是 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)

组装链——大模型实际看到的内容

通过 assembly=[...] 传入 AssemblyPolicy 实例。Agent 会在最外层中间件位置配置内部的 AssemblerMiddleware。每个策略都会转换 (prompts, events)，并将结果传递给下一个策略。

策略分为两类——顺序很重要：先注入，再精简。

类型	用途	内置策略
注入	添加到 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 提取并持久化内容。内置两种策略，都会通过 ModelConfig 调用模型进行摘要（使用比 Agent 主模型更便宜的模型）：

策略	写入内容	配合使用
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() 不会触发任何操作；至少启用一项。on_end=True 默认关闭，因为每次触发都会调用一次大模型。

压缩 — 精简流历史记录

CompactStrategy.compact(events, ctx, store) → list[BaseEvent]。替换该流的历史记录。内置两种策略：

策略	行为	成本
TailWindowCompact(target=N)	保留最后 N 个事件；丢弃其余事件（可选将其持久化到 /log/）	零次大模型调用
SummarizeCompact(target=N, config=...)	将丢弃的事件汇总为一个 CompactionSummary；插入到开头	每次触发一次大模型调用

CompactTrigger(max_events=N, max_tokens=M, chars_per_token=4) — 任一阈值超过时触发。

from ag2.compact import CompactTrigger, TailWindowCompact, SummarizeCompact

SummarizeCompact 会在开头插入一个 CompactionSummary 事件；ConversationPolicy 会放行该事件，因此大模型仍能获取这部分上下文。

在 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=...) 会执行四项操作——其中三项现在可以选择关闭

传入 knowledge=KnowledgeConfig(store=...) 会打包四项功能。默认设置会保留原有行为，但现在可以通过 expose_tool 和 write_event_log 标志关闭其中大部分功能：

1. 注册存储，以便组装策略（WorkingMemoryPolicy 等）可以调用 context.dependencies.get(KnowledgeStore)。（通常这就是你想要的唯一功能——也是唯一没有对应标志的功能，因为这正是它的目的。）
2. 自动将一个 knowledge 动作组工具注入 Agent 的工具列表（knowledge(action="write"/"read"/"list"/...)）。设置 expose_tool=False 即可关闭。
3. 首次调用 ask 时，将 SKILL.md 文件写入存储——/SKILL.md、/artifacts/SKILL.md、/memory/SKILL.md、/log/SKILL.md——除非你传入显式的 bootstrap=。默认值为 DefaultBootstrap(mention_tool=expose_tool)，因此当你设置 expose_tool=False 时，写入的 SKILL.md 会自动停止告诉大模型“使用 knowledge 工具”（因为没有这个工具）。若要完全不写入任何文件，请将你自己的空操作 StoreBootstrap 传给 bootstrap=（这是一个包含 async def bootstrap(self, store, actor_name): pass 的类）；目前没有内置的 NoBootstrap。
4. 将轮次事件持久化到 /log/——每轮结束后，Agent 的完整事件历史都会写入 /log/{stream_id}.jsonl。设置 write_event_log=False 即可关闭。持久化失败时，流上会发出 EventLogFailed 事件（并调用 logger.exception）。

因此，要实现“策略可以访问存储、大模型不可见、没有 /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，请添加 bootstrap=NoBootstrap()，其中 NoBootstrap 是你自己的空操作 StoreBootstrap。较早的替代方案——完全跳过 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。只需承担你启用的功能。

Agent 流上会发出的生命周期事件：

• 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 构造函数参考、轮次生命周期中间件顺序。

常见陷阱

• 先缩减再注入——将 SlidingWindowPolicy 放在 WorkingMemoryPolicy 之前，意味着工作记忆注入不会计入预算。务必按以下顺序操作：先注入，再应用 AlertPolicy，最后缩减。
• 忘记为记忆策略添加 KnowledgeStore 依赖——WorkingMemoryPolicy 和 EpisodicMemoryPolicy 会通过 context.dependencies.get(KnowledgeStore) 查找存储。KnowledgeConfig(store=...) 会为你注册；如果手动接入策略，也要在 dependencies 中注册存储。
• 每次触发聚合都会产生一次大模型调用——在每次对话中设置 on_end=True，调用次数可能会累积。请谨慎搭配 WorkingMemoryAggregate 和 ConversationSummaryAggregate；对于高流量 Agent，可考虑设置 every_n_turns=N。
• 混用 HistoryLimiter 中间件和组装缩减策略——两者都会裁剪内容。请选择一种机制。组装方式更灵活（可进行丰富的内容整形并添加说明）；HistoryLimiter 则更简单。
• read_range 使用字节偏移量，而不是字符偏移量——多字节 UTF-8 序列需要仔细对齐。
• 忘记 WorkingMemoryAggregate 会破坏性地覆盖内容——每次触发时，它都会覆盖 /memory/working.md。这是有意为之（滚动状态，而非日志），但要做好先前内容被合并或消失的准备。
• 以为 AlertPolicy 不在 assembly= 中也会向大模型呈现提醒——提醒会作为 ObserverAlert 事件留在流中，但只有 AlertPolicy 注入后才会传递给大模型。
• KnowledgeConfig(store=...) ≠“只是一个存储句柄”——它还会自动注入一个 knowledge 大模型工具、将 SKILL.md 文件写入存储，并把每轮事件写入 /log/。可通过 KnowledgeConfig(store=..., expose_tool=False, write_event_log=False) 关闭这些功能（此时默认的引导流程也会停止提及该工具，因为 mention_tool 默认取值为 expose_tool）；或者跳过 KnowledgeConfig，使用 Agent(..., dependencies={KnowledgeStore: store}) 仅配置最基本内容。请参阅上文的“KnowledgeConfig(store=...) 会做四件事——现在其中三件可选择关闭”。
• WorkingMemoryAggregate 的默认提示词以内容为导向——开箱即用时，它保留的是对话讨论了什么（“保留重要的现有上下文，删除过时信息”），而不是策略；希望记忆追踪策略（哪些措辞／领域有效）的研究 Agent，反而会得到过时的主题事实。传入 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。请为文件系统根目录使用不同的名称（如 ./agent-state、./journal-state），避免遇到“memory 里还有个 memory”的困惑。