Context 与 memory

从网页完整代码学习 QitOS 资料整理项目。

​

目标与前置条件

资料 Agent 需要项目指令和记忆事实。在 extensions 中绑定具体实现，再选择 memory.sources: [project_memory] 和 compaction.provider: closed_window。MemorySourceAdapter 同时适配 MemdirMemory 与 MarkdownFileMemory，用户不需要手写 contribution、provider messages 或 compaction receipt。

fixture 发出四次脚本化工具调用（对两篇笔记各处理两次），然后结束。只有请求超过配置预算时，ClosedExchangeWindowCompactor 才按最旧优先省略符合条件的 closed exchanges，不生成摘要。ContextBudget.protected_recent_exchanges=2 保护最近窗口。没有显式选择 compactor 时，历史超预算会失败，不会静默丢弃 exchanges。

公开 YAML loader 接受 context.budget_policy: budget 与 context.allow_codec_loss: true。在 extensions 中将 budget 显式绑定到预算 policy factory。loss 必须是真实布尔值；未知字段、未绑定 factory 会在首个模型请求前失败。下方完整教程同时展示其他 context 服务的 Python 组合；B 的安装消费者直接在 YAML 中配置这两个字段。

需要 Python 基础；声明支持 Python ≥3.10，本轮本地验证使用 Python 3.12.7。每章可独立运行；继续已有项目时可复用环境和同名文件。以下命令为 macOS/Linux shell。

​

准备项目

python3 -m venv .venv
source .venv/bin/activate
python -m pip install "qitos @ git+https://github.com/WhitzardAgent/WhitzardOS.git@f7d4b2d666a156d361da496a41868278f84ffabf"
mkdir notes_lesson
cd notes_lesson

本教程使用下方固定的 R1 融合实现；源码安装不代表已发布 PyPI 版本。

将下方完整文件保存到当前目录。无需 clone 仓库、安装 editable 包或复制 tests。

​

运行并验证

python context.py --root context-run

预期出现以下输出片段（随机 ID 不固定）。每次运行都检查工具结果或持久化断言，成功退出码为 0。

compaction loss recorded

​

行为与支持边界

Memory 是 user placement 的请求上下文数据，不是 system 指令、权限或 checkpoint store。namespace 是应用 factory 绑定到 memory 资源的逻辑名称，不授予读取任意目录的权限。默认不读取 global memory。

Memdir 文本记录的独立 record_id 跨进程保持稳定。两条独立记录即使文本相同也保留两个 identity；修改内容会改变 contribution revision/digest。召回重新读取磁盘，不合并追加时的旧缓存；外部修改和删除会反映在 retrieve 与 summarize 中。contribution ID、内容和 metadata 不会自动加入 host path。只有明确初始化时使用 create=True；普通构造恢复已有 root，缺失时抛出 MemoryResourceError。这改变了原来的隐式创建默认值，也适用于 memdir_memory(...)。

adapter 借用 memory，adapter 的 reset/close 不会重置、关闭或删除它。Memdir 的 reset/evict 仅影响内存缓存，不删除持久文件。durable reference 仅承诺文本记录，不承诺任意 Python/JSON 记录或历史 metadata 无损恢复。MarkdownFileMemory 可在当前实例内消费，但新实例不会重载已有 append-only 日志；不会执行 metadata，也不会猜测重建。

selection 不修改 ExchangeLog bytes 或工具完成顺序；tool declaration/result 要么一起保留，要么一起省略。未完成 exchange、未消费 steering、required context/artifact 保持受保护。由于没有声明更小的依赖闭包，opaque continuation 状态会保守保护完整历史。受保护内容无法放入预算时，selection 返回 typed failure。相同 revision 在此前省略或恢复后再次召回时会重新注入。

安装后的两进程 consumer 写入 remembered-value=17 后退出，再根据配置恢复 memory，验证实际编码请求和重复预算压缩。它使用 deterministic provider，不调用真实模型。旧 CompactHistory 保持兼容。Markdown durable reader、摘要模型、完整旧 history 退休、跨 Agent memory 策略和原 Agent 迁移实验仍不在本轮范围。

​

练习与参考答案

修改持久文本但保留 record ID，验证 revision 变化。在第二个进程恢复相同资源；factory 绑定的另一个 namespace 应保持空。

​

常见错误与清理

ModuleNotFoundError：确认已激活安装指定 wheel/源码版本的环境，并保存本页所有文件。root 已存在：换一个新 --root，不要覆盖需要保留的证据。断言失败：检查第一个失败的工具或 typed error；不要只依赖最终文字。退出所有进程、停止 board 后，可自行删除本章新建且不再需要的运行目录；保留要调试的 SQLite、journal 和报告。

​

完整文件：复制到项目根目录

notes.py

"""Synthetic notes; fake model, real tools, composition, Session and journal."""
import argparse
from dataclasses import replace
import json
from pathlib import Path

from qitos.config import build_agent_composition, load_agent_config
from qitos.core.function_tool_decorator import function_tool
from qitos.engine.runtime import LifecyclePolicy

# docs:start fixture
NOTES = (
    "Session: A durable session can resume after a process exits.",
    "Artifact: Large tool outputs can be retained outside model context.",
)


@function_tool(read_only=True, concurrency_safe=True)
def summarize_note(index: int) -> dict:
    """Extract a title and word count from a synthetic in-memory note."""
    text = NOTES[index]
    return {"title": text.split(":", 1)[0], "words": len(text.split())}
# docs:end fixture


# docs:start provider
class FakeProvider:
    """Scripted responses; this does not summarize or reason like a real model."""
    model = "notes-fake"
    qitos_protocol = "react_text_v1"

    def __init__(self, start=0):
        self.stage = start

    def call_raw(self, messages, **options):
        if self.stage < len(NOTES):
            content = f"Thought: inspect a note\nAction: summarize_note(index={self.stage})"
        else:
            content = "Final Answer: Indexed 2 notes: Session, Artifact."
        self.stage += 1
        return {"choices": [{"message": {"content": content}}]}
# docs:end provider


class PauseAfterTool(LifecyclePolicy):
    policy_id = "notes.pause_after_tool"

    def should_pause(self, context):
        return context.step_id == 0


# docs:start composition
def configuration(root):
    config = load_agent_config(Path(__file__).with_name("agent.yaml"))
    return replace(config, runtime=replace(
        config.runtime, data_root=str(root / "data"),
        environment=replace(config.runtime.environment, workspace=str(root)),
        session=replace(config.runtime.session, path=str(root / "sessions.sqlite3")),
        trajectory=replace(config.runtime.trajectory, output=str(root / "trajectory.journal")),
    ))


def compose(root, *, start=0, pause=False):
    config = configuration(root)
    if pause:
        config = replace(config, lifecycle={"policy": "pause"})
    composition = build_agent_composition(
        config, model_override=FakeProvider(start), extensions={"pause": PauseAfterTool},
    )
    composition.tool_registry.register(summarize_note)
    return composition
# docs:end composition


# docs:start run
def run(root):
    root.mkdir(parents=True, exist_ok=False)
    with compose(root) as composition:
        session = composition.session("Index both synthetic notes")
        result = session.run()
        outputs = [action.output for record in result.records for action in record.action_results
                   if action.tool_name == "summarize_note"]
        assert [output["title"] for output in outputs] == ["Session", "Artifact"]
        assert result.state.final_result == "Indexed 2 notes: Session, Artifact."
        config = composition.config.to_dict()
        config["runtime"]["environment"] = {"type": "unsafe_host", "workspace": str(root)}
        (root / "agent.json").write_text(json.dumps(config), encoding="utf-8")
        control = {"session_id": session.session_id.value, "run_id": result.run_id}
        (root / "control.json").write_text(json.dumps(control), encoding="utf-8")
        print(json.dumps({**control, "result": result.state.final_result, "outputs": outputs}))
# docs:end run


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", type=Path, default=Path("notes-run"))
    run(parser.parse_args().root.resolve())

agent.yaml

schema: qitos.agent
agent:
  name: notes_agent
  protocol: react_text_v1
model:
  provider: openai_compatible
  model: notes-fake
  credential:
    ref: notes-provider
  request:
    max_tokens: 512
    timeout_seconds: 30
    retries: 0
tools:
  preset: none
runtime:
  environment:
    type: unsafe_host
    workspace: .
  session:
    mode: durable
    store: sqlite
    path: ./notes-run/sessions.sqlite3
  trajectory:
    enabled: true
    output: ./notes-run/trajectory.journal
budgets:
  max_steps: 6
  max_requests: 6
  max_runtime_seconds: 30
failure_policy:
  tool: fail_closed

context.py

"""Explicit contributor, memory, selector and compactor factories; no network."""
import argparse
from dataclasses import replace
from pathlib import Path

from qitos.config import build_agent_composition
from qitos.core.context import (
    DeclaredContextBudgetPolicy, PriorityContextSelectionPolicy, StaticContextContributor,
)
from qitos.core.memory import MemoryRecord
from qitos.kit.context.compaction import ClosedExchangeWindowCompactor
from qitos.kit.memory.adapter import MemorySourceAdapter
from qitos.kit.memory.memdir_memory import MemdirMemory
from qitos.qita.reader import default_reader
from qitos.tracing.trajectory import PrivacyView

from notes import FakeProvider, summarize_note, configuration


class ContextProvider(FakeProvider):
    """Four scripted tool turns make closed-exchange omission observable."""

    def __init__(self):
        self.stage = 0

    def call_raw(self, messages, **options):
        content = (f"Thought: inspect\nAction: summarize_note(index={self.stage % 2})"
                   if self.stage < 4 else "Final Answer: Indexed 2 notes: Session, Artifact.")
        self.stage += 1
        return {"choices": [{"message": {"content": content}}]}


class AuditedSelector(PriorityContextSelectionPolicy):
    def __init__(self):
        self.seen = set()

    def select(self, contributions, **options):
        contributions = tuple(contributions)
        self.seen.update(item.contribution_id for item in contributions)
        return super().select(contributions, **options)


def run(root):
    root.mkdir(parents=True, exist_ok=False)
    config = replace(
        configuration(root),
        context={"contributors": ["project"], "selector": "selector", "budget_policy": "budget", "allow_codec_loss": True},
        memory={"sources": ["project_memory"]}, compaction={"provider": "closed_window"})
    config = replace(config, budgets=replace(config.budgets, max_steps=6, max_requests=6))
    selector = AuditedSelector()
    memory = MemdirMemory(str(root / "memory"), create=True)
    memory.append(MemoryRecord("user", 'Session and Artifact are the two note titles.', 0))
    with build_agent_composition(config, model_override=ContextProvider(), extensions={
        "project": lambda: StaticContextContributor("lesson.project", "project", "Use only the supplied notes."),
        "project_memory": lambda: MemorySourceAdapter(memory, namespace="notes", required=True),
        "selector": selector, "closed_window": ClosedExchangeWindowCompactor,
        "budget": lambda: DeclaredContextBudgetPolicy(default_max_input_units=30000, protected_recent_exchanges=2),
    }) as composition:
        composition.tool_registry.register(summarize_note)
        session = composition.session("Index both notes")
        assert session.run().state.final_result == "Indexed 2 notes: Session, Artifact."
        assert "lesson.project" in selector.seen
        assert any(identity.startswith("memory:notes:") for identity in selector.seen)
        trajectory = default_reader(root).read_session(session.session_id.value, view=PrivacyView.RAW_PRIVATE)
        assert any(record.kind.value == "compaction" and not record.loss.is_lossless for record in trajectory.records)
    print("context selected; memory selected; compaction loss recorded")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", type=Path, required=True)
    run(parser.parse_args().root.resolve())

​

下一步与 API

API Reference · Configuration · Learning path · Next

Source file (可选；全部所需代码已在本页)