工具与并行调用

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

Observation 保留 dict 兼容性，但每个字段只保存一次。 obs.task = value 与 obs["task"] = value 写入同一权威状态；映射中的 step 与 step_id 均对应属性 step_id。别名矛盾、非法类型和删除 schema 字段会原子拒绝。 update 与 |= 先校验整次更新；扩展键支持 pop、del、setdefault、popitem。 clear 会删除必需 identity，因此拒绝。state/metadata 的嵌套修改在两种访问方式下一致。 构造输入防御性复制，to_dict 与 to_legacy_dict 返回独立快照。属性 action results 使用 canonical ToolResult，to_dict 序列化其 canonical shape；映射中的 action results 按需生成 legacy projection，以兼容既有 reducer。请通过 obs["action_results"] 赋值 立即校验并替换结果；直接调用未绑定的 dict 修改方法不在此契约范围内。

旧 host Read 使用从 0 开始的 offset 与正整数 limit：offset=2、limit=2 只选择第 3/4 行。空文件与 EOF 返回空 output；负数或非整数窗口失败，selection 元数据保留截断与后续行信息。Edit 默认要求唯一匹配，replace_all=True 才替换全部。 无匹配、非唯一匹配、权限拒绝与显式 expected_sha256 冲突均不修改文件。 这两个 alias 现在返回 canonical ToolResult，编号文本在 .output，状态与错误在对应字段， 不再把错误包装成成功字符串。原先直接使用字符串的调用方应先检查 .status 再读取 .output。 既有 validation/permission pipeline 继续决定权限。Host compatibility 仍在 Host 执行； Env read_file/edit_file 仍只使用 Env ops，保留从 1 开始的 line_offset。 不增加 Host 回退或隔离保证。

​

目标与前置条件

函数经过装饰并注册到 ToolRegistry 后才能被 Agent 调用。函数名和参数类型构成调用接口；read_only、concurrency_safe 是工具作者对行为的声明，不会自动授予权限。

先运行顺序和并行资料示例。再运行 parallel.py：第一个声明的工具等待第二个工具发出的内存事件，因此完成顺序固定反转，而 reducer 收到的仍是声明顺序。事件与列表是公开可见、仅用于本次执行的教学观测变量。

五秒等待是 fixture 的截止时间，不是性能基准。只给一个 worker 无法完成此例。工具错误、超时、不完整输出和未知外部副作用必须检查 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

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

​

资料与工具

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())}

​

运行并验证

python custom_agent.py
python parallel.py

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

completion=[1, 0]; declaration=[0, 1]

​

行为与支持边界

timeout 不等于硬取消。遇到 worker_still_running 或 outcome_unknown 应先核查实际结果，不自动重放未知副作用。

​

练习与参考答案

把逆序 fixture 的 max_concurrency 改为 1 并解释失败；恢复为 2 后通过，不通过延长超时隐藏依赖。

​

常见错误与清理

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()

parallel.py

"""Force reverse completion while retaining declaration order in reduce."""
from threading import Event

from qitos import Action, Decision, Engine, ToolRegistry
from qitos.core.function_tool_decorator import function_tool
from qitos.engine.action_executor import ActionExecutionPolicy
from qitos.engine.runtime import RuntimeComposition
from custom_agent import NotesAgent
from notes import NOTES

# Explicit test instrumentation for one local invocation, not persistent state.
second_finished = Event()
completion_order = []


@function_tool(read_only=True, concurrency_safe=True)
def analyze_note(index: int) -> dict:
    """An in-memory fixture that controls completion order without file/network I/O."""
    if index == 0:
        if not second_finished.wait(5):
            raise RuntimeError("This fixture requires two parallel workers")
    completion_order.append(index)
    if index == 1:
        second_finished.set()
    return {"title": NOTES[index].split(":", 1)[0]}


class ParallelNotesAgent(NotesAgent):
    def __init__(self):
        super().__init__()
        self.tool_registry = ToolRegistry()
        self.tool_registry.register(analyze_note)

    def decide(self, state, observation):
        if state.titles:
            return Decision.final(", ".join(state.titles))
        return Decision.act([Action(name="analyze_note", args={"index": i}) for i in (0, 1)])


def run():
    second_finished.clear()
    completion_order.clear()
    engine = Engine(ParallelNotesAgent(), runtime=RuntimeComposition(),
                    action_execution_policy=ActionExecutionPolicy(mode="parallel", max_concurrency=2))
    result = engine.session("Compare completion with declaration order").run()
    assert completion_order == [1, 0]
    assert result.state.titles == ["Session", "Artifact"]
    print("completion=[1, 0]; declaration=[0, 1]; titles=Session, Artifact")


if __name__ == "__main__":
    run()

​

下一步与 API

API Reference · Configuration · Learning path · Next

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