你将完成什么
创建一个资料整理项目，调用工具提取两条合成资料的标题和词数，通过 Session 执行，再用 qita 查看记录。你只需 Python 基础，不需要模型凭据或 Docker；这里的 fake provider 是显式脚本替身。准备项目
python3 -m venv .venv
source .venv/bin/activate
python -m pip install "qitos @ git+https://github.com/WhitzardAgent/WhitzardOS.git@f7d4b2d666a156d361da496a41868278f84ffabf"
python -m pip install pytest
qit new --agent-name notes_agent --output-dir . --no-input
cd notes_agent
资料与工具
NOTES 是全部输入。装饰器把普通 Python 函数变成可注册工具；这个函数只读取内存。返回字典是真实工具结果，随后会被断言验证。
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())}
明确的 fake provider
这个 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}}]}
配置与组合
把本页完整文件保存到刚生成的notes_agent/ 根目录，并替换 agent.yaml。配置选择 SQLite 和 journal；程序只注册受信任的纯函数。unsafe_host 不提供隔离，不要在此配置加入文件或 shell 工具。
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
运行与断言
with 负责资源关闭；composition.session 创建 Session，run 返回 EngineResult。检查实际工具输出，再检查最终文字。筛选工具名是因为记录还包含环境观察。
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}))
运行并验证
python -m pip install .
python -m pytest -q tests
python notes.py --root notes-run
Indexed 2 notes: Session, Artifact.
检查保存的 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
行为与支持边界
工具的实际输出与 provider 的回答分开验证。build_agent_composition 组合配置，composition.session 创建 Session，session.run 返回 EngineResult；with 退出时关闭所拥有的资源。
练习与参考答案
把第一条标题改为Recovery，同时更新标题断言和 fake provider 的固定回答；词数仍由工具计算。
常见错误与清理
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