使用 qita 检查与导出

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

目标与前置条件

先运行资料项目，再用 control.json 中的准确 Session ID 检查结果。default_reader 默认读取 canonical journal，同时保留明确的历史 trace 支持。read_session 重建观察视图，不恢复执行。

inspect_run.py 读取快照分页，通过本次自有 staging 流式写出 public canonical export。小规模 re-import 检查逐条比较投影后的记录；re-import 本身仍会物化数据。教程保留旧 replay/HTML CLI 所需的空 selector 目录。新的 canonical export 直接接受 journal 与 run ID。

使用下方命令检查、回放和导出。replay 展示已记录执行，不调用模型、不重做工具。只读 board 用 Ctrl-C 退出。

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

准备项目

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

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

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

如果继续 Quickstart 且 notes-run 已存在，跳过下面第一条 notes.py 创建命令，直接检查已有运行。replay 启动本地网页服务；先用 Ctrl-C 停止再执行 export/board。

运行并验证

python notes.py --root notes-run
python inspect_run.py --root notes-run

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

"records":

检查保存的 Session

session_id=$(python -c 'import json; print(json.load(open("notes-run/control.json"))["session_id"])')
qit session inspect --config notes-run/agent.json --session-id "$session_id"
qita inspect session "$session_id" --logdir ./notes-run

run_id=$(python -c 'import json; print(json.load(open("notes-run/control.json"))["run_id"])')
qita replay --run "notes-run/$run_id"
qita export --run "notes-run/$run_id" --html notes-run/trajectory.html
qita export --run "$run_id" --journal notes-run/trajectory.journal --canonical notes-run/streamed.json
qita board --logdir ./notes-run

行为与支持边界

inspect 不需要 Docker 或 credentials。read_page 仅保留当前页、有 frame 大小上限的缓冲和固定 I/O/SQLite cache；临时磁盘索引随日志增长。每页仍 hash 快照全部历史字节。完整 read_run/read_session、旧 HTML export 和 re-import 仍会物化数据。public export 声明损失，不能替代私有恢复数据。

cursor 绑定查询/视图、来源、committed head 与字节边界，reader close 后失效。并发 append 不混入当前遍历。末页后以该页 watermark 作为新查询的 after_sequence，显式捕获新 head 来 polling。查询变化、来源替换/截断和快照损坏均 typed reject。不支持此能力的历史或第三方来源抛出 BoundedReadUnsupported；旧完整读取保持原行为。

练习与参考答案

在不公开私有 payload 的前提下比较 private 与 public 记录；记录数相同不能证明 payload 无损。

常见错误与清理

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

连续 iter_records 与文件导出在开始、结束时校验完整快照，遍历时逐 frame 校验。迭代值在耗尽前属于 partial；提前 close 释放资源，但不证明遍历完成。文件仅在最终校验后发布。独立 read_page 每页仍校验全部历史字节。

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

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

inspect_run.py

"""Page an existing notes journal, stream a public export and verify re-import."""
import argparse
import json
from pathlib import Path

from qitos.qita.reader import default_reader
from qitos.tracing.exporter import CanonicalTrajectoryExporter, ExportArtifact
from qitos.tracing.paging import iter_records
from qitos.tracing.trajectory import LossReport, PrivacyView, TrajectoryQuery


def inspect(root):
    control = json.loads((root / "control.json").read_text())
    reader = default_reader(root)
    try:
        query = TrajectoryQuery(session_id=control["session_id"], limit=128)
        page = reader.read_page(query)
        assert page.records
        exporter = CanonicalTrajectoryExporter()
        target = root / "public-trajectory.json"
        receipt = exporter.export_file(reader, query, target)
        # Re-import remains a materializing API; this small lesson checks every record.
        data = target.read_bytes()
        value = json.loads(data)
        artifact = ExportArtifact(exporter.capabilities.exporter_id, exporter.capabilities.format_version,
                                  "application/json", data, receipt.digest, PrivacyView.REDACTED_PUBLIC,
                                  True, value["provenance"], LossReport.from_dict(value["loss"]))
        imported = exporter.reimport(artifact)
        expected = iter_records(reader, query)
        for actual in imported.records:
            assert actual.to_dict() == next(expected).to_dict()
        assert next(expected, None) is None
        # Legacy replay/HTML CLI still accepts an existing directory selector.
        (root / control["run_id"]).mkdir(exist_ok=True)
        print(json.dumps({"records": receipt.record_count, "lossless": imported.loss.is_lossless,
                          "run_selector": str(root / control["run_id"]), "bounded_export": True}))
    finally:
        reader.close()


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

下一步与 API

API Reference · Configuration · Learning path · Next

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