目标与前置条件
自定义 Agent 把决策策略写成可读代码：NotesState.titles 起初为空，decide 声明两个动作，reduce 把观察合入状态，下一次 decide 返回最终决定。本章不需要模型。
仍使用 Quickstart 的资料，但这次替换决策逻辑。AgentComposition 构造的是配置式 Agent，没有任意 agent_override 参数。使用公共 Engine 组合自己的 AgentModule 与 RuntimeComposition，再创建 Session。
顺序与并行两种模式必须得到相同的声明顺序标题列表。注意 reduce 收到的动作观察是字典，而 EngineResult.records 中是 typed ToolResult；两种表示不能混用。
需要 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
状态与决策
@dataclass
class NotesState(StateSchema):
titles: list[str] = field(default_factory=list)
class NotesAgent(AgentModule):
def __init__(self):
registry = ToolRegistry()
registry.register(summarize_note)
super().__init__(tool_registry=registry)
def init_state(self, task, **kwargs):
return NotesState(task=task, max_steps=3)
def decide(self, state, observation):
if state.titles:
return Decision.final(", ".join(state.titles))
return Decision.act([
Action(name="summarize_note", args={"index": index}) for index in (0, 1)
])
def reduce(self, state, observation, decision):
state.titles.extend(item["output"]["title"] for item in observation.get("action_results", []))
return state
运行与断言
def main():
for mode in ("sequential", "parallel"):
engine = Engine(NotesAgent(), runtime=RuntimeComposition(),
action_execution_policy=ActionExecutionPolicy(mode=mode, max_concurrency=2))
result = engine.session("Index the two notes").run()
assert result.state.titles == ["Session", "Artifact"]
assert result.state.final_result == "Session, Artifact"
print(f"{mode}: Session, Artifact; process_local=true")
运行并验证
python custom_agent.py
sequential: Session, Artifact
parallel: Session, Artifact
行为与支持边界
默认 RuntimeComposition 使用进程内 Memory checkpoint store。有 Session ID 不意味着数据已经落盘；跨进程恢复请使用生命周期章节的 SQLite 组合。练习与参考答案
只处理第一条：把动作索引改成(0,)，两种模式都应输出 Session。
常见错误与清理
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
custom_agent.py
"""A custom notes AgentModule through the public Engine/Session path."""
from dataclasses import dataclass, field
from qitos import Action, AgentModule, Decision, Engine, StateSchema, ToolRegistry
from qitos.engine.action_executor import ActionExecutionPolicy
from qitos.engine.runtime import RuntimeComposition
from notes import summarize_note
# docs:start agent
@dataclass
class NotesState(StateSchema):
titles: list[str] = field(default_factory=list)
class NotesAgent(AgentModule):
def __init__(self):
registry = ToolRegistry()
registry.register(summarize_note)
super().__init__(tool_registry=registry)
def init_state(self, task, **kwargs):
return NotesState(task=task, max_steps=3)
def decide(self, state, observation):
if state.titles:
return Decision.final(", ".join(state.titles))
return Decision.act([
Action(name="summarize_note", args={"index": index}) for index in (0, 1)
])
def reduce(self, state, observation, decision):
state.titles.extend(item["output"]["title"] for item in observation.get("action_results", []))
return state
# docs:end agent
# docs:start run
def main():
for mode in ("sequential", "parallel"):
engine = Engine(NotesAgent(), runtime=RuntimeComposition(),
action_execution_policy=ActionExecutionPolicy(mode=mode, max_concurrency=2))
result = engine.session("Index the two notes").run()
assert result.state.titles == ["Session", "Artifact"]
assert result.state.final_result == "Session, Artifact"
print(f"{mode}: Session, Artifact; process_local=true")
# docs:end run
if __name__ == "__main__":
main()