目标与前置条件
composition 接受公共 model override 和显式命名的扩展工厂。ObservedFakeProvider 保持调用接口，计数实际请求，并断言每个请求都包含选中的项目 context，随后委托给网页可见的确定性 provider。
项目 contributor 工厂返回 StaticContextContributor，注册名与 context.contributors 一致。不从 YAML 中任意导入代码，执行哪些扩展由应用程序明确决定。
这是完整的 provider 替换和 context 集成示例，不是新的传输客户端。真实 provider 使用配置章节中的配置与显式 credential resolver。替换 store、sink、sandbox 前阅读扩展索引：只有相同 Python 方法签名不能保证持久化和安全契约不变。
需要 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
运行并验证
python provider_extension.py --root extension-run
provider replaced; context observed; requests=3
行为与支持边界
不存在对所有 provider 的通用兼容保证。codec、continuation、tool schema 与 loss policy 必须匹配。检查 typed failure，不能关闭 loss 检查接受不兼容响应。练习与参考答案
同时修改 config 与 extensions 中 contributor 名称；只改一边应产生配置或扩展失败。常见错误与清理
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
provider_extension.py
"""Replace the provider and inject context, without changing the kernel."""
import argparse
from dataclasses import replace
from pathlib import Path
from notes import FakeProvider, configuration, summarize_note
from qitos.config import build_agent_composition
from qitos.core.context import StaticContextContributor
class ObservedFakeProvider(FakeProvider):
"""Keep the public provider call shape and validate selected context."""
def __init__(self):
super().__init__()
self.requests = 0
def call_raw(self, messages, **options):
assert "notes-project-context" in str(messages)
self.requests += 1
return super().call_raw(messages, **options)
def run(root):
root.mkdir(parents=True, exist_ok=False)
config = replace(configuration(root), context={"contributors": ["project"]})
provider = ObservedFakeProvider()
with build_agent_composition(config, model_override=provider, extensions={
"project": lambda: StaticContextContributor("notes.project", "project", "notes-project-context"),
}) as composition:
composition.tool_registry.register(summarize_note)
result = composition.session("Index both notes").run()
assert result.state.final_result == "Indexed 2 notes: Session, Artifact."
assert provider.requests == 3
print("provider replaced; context observed; requests=3")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--root", type=Path, required=True)
run(parser.parse_args().root.resolve())
consumer.py
"""Offline installed-wheel mechanism proof; no model capability claim."""
from __future__ import annotations
import json
import sys
from pathlib import Path
from tempfile import TemporaryDirectory
from threading import Event
from types import SimpleNamespace as NS
from unittest.mock import patch
from qitos.config import FakeCredentialResolver, load_agent_config, build_agent_composition
from qitos.core.conversation import ExchangeLog
from qitos.core.function_tool_decorator import function_tool
from qitos.tracing.journal_store import JournalTrajectoryStore
def run(interrupted=False):
second_finished = Event()
completions = []
executions = []
requests = []
resources = []
@function_tool(read_only=True, concurrency_safe=True)
def add(a: int, b: int) -> int:
"""Add two integers."""
executions.append(('add', a, b))
if (a, b) == (2, 3):
assert second_finished.wait(5)
return a + b
@function_tool(read_only=True, concurrency_safe=True)
def multiply(a: int, b: int) -> int:
"""Multiply two integers."""
executions.append(('multiply', a, b))
return a * b
class CompletionHook:
def on_event(self, event, state, record, engine):
if event.payload.get('stage') == 'tool_slot_terminal':
name = event.payload['tool_result']['tool_name']
completions.append(name)
if name == 'multiply':
second_finished.set()
def chunk(text='', calls=None, finish=None):
return NS(choices=[NS(delta=NS(content=text, tool_calls=calls),
finish_reason=finish)], usage=None)
def call(index, identifier, name, arguments):
return NS(index=index, id=identifier, type='function',
function=NS(name=name, arguments=arguments))
class Stream:
def __init__(self, values):
self.values = iter(values)
self.closed = 0
resources.append(self)
def __iter__(self):
return self
def __next__(self):
return next(self.values)
def close(self):
self.closed += 1
class Client:
def __init__(self, **kwargs):
assert kwargs['max_retries'] == 0
self.closed = 0
self.chat = NS(completions=self)
resources.append(self)
def create(self, **kwargs):
assert kwargs['stream'] is True
assert kwargs['max_tokens'] == 10240
requests.append(kwargs)
results = {m['tool_call_id']: json.loads(m['content'])
for m in kwargs['messages'] if m['role'] == 'tool'}
if len(requests) == 1:
return Stream([chunk(calls=[call(0, 'one', 'add', '{"a":2,"b":3}'),
call(1, 'two', 'multiply', '{"a":2,"b":3}')]),
chunk(finish='tool_calls')])
if len(requests) == 2:
assert results == {'one': 5, 'two': 6}, results
values = [chunk(calls=[call(0, 'three', 'add', '{"a":5,"b":6}' if not interrupted else '{"a":5')])]
if not interrupted:
values.append(chunk(finish='tool_calls'))
return Stream(values)
assert len(requests) == 3
assert results['three'] == 11
return Stream([chunk('{"final_answer":"11"}'), chunk(finish='stop')])
def close(self):
self.closed += 1
with TemporaryDirectory(prefix='qitos-r1-consumer-') as directory:
root = Path(directory)
path = root / 'agent.yaml'
journal = root / 'trajectory.journal'
path.write_text(f'''schema: qitos.agent
agent:
name: offline-multi-round
protocol: json_decision_multi_v1
parser: auto
model:
provider: openai_compatible
model: offline
base_url: https://offline.invalid/v1
credential:
ref: offline
request:
max_tokens: 10240
retries: 0
tools:
preset: none
policy: auto
runtime:
environment:
type: unsafe_host
workspace: {root}
session:
enabled: true
store: memory
trajectory:
enabled: true
output: {journal}
privacy: private
budgets:
max_steps: 5
max_requests: 3
''')
with patch.dict(sys.modules, {'openai': NS(OpenAI=Client, APIConnectionError=ConnectionError, APITimeoutError=TimeoutError)}):
with build_agent_composition(load_agent_config(path),
credential_resolver=FakeCredentialResolver()) as composition:
composition.tool_registry.register(add)
composition.tool_registry.register(multiply)
composition.engine.hooks.append(CompletionHook())
composition.engine.stream_callback = lambda text: None
session = composition.session('Compute add(2,3) and multiply(2,3), then add their results.')
result = session.run()
inspection = session.inspect()
assert inspection.budget['model_requests_consumed'] == len(requests)
snapshot = composition.runtime.checkpoint_store.get_session_snapshot(
session.current_head.snapshot_id.value)
conversation = next(c for c in snapshot.payload['components'] if c['slot'] == 'conversation')
log = ExchangeLog.from_dict(conversation['payload']['exchange_log'])
assert log.open_batch_id() is None
batches = [item.batch_id for item in log.items if getattr(item, 'batch_id', None)
and getattr(item, 'tool_calls', None)]
assert len(batches) == (1 if interrupted else 2)
for batch in batches:
calls = log.declared_calls(batch)
terminals = log.results_for_batch(batch)
assert len(calls) == len(terminals)
assert len({item.identity.key() for item in terminals}) == len(calls)
ordered = log.results_for_batch_in_declaration_order(batch)
assert [item.identity for item in ordered] == [item.identity for item in calls]
assert [item.identity.call_id for item in log.results_for_batch(batches[0])] == ['two', 'one']
assert completions[:2] == ['multiply', 'add']
assert len(completions) == len(executions)
assert len(executions) == (2 if interrupted else 3)
assert len(requests) == (2 if interrupted else 3)
if interrupted:
assert ('add', 5, 6) not in executions
assert result.error_code == 'provider_stream_protocol_error'
assert result.state.final_result is None
else:
assert str(result.state.final_result) == '11', result
session_id = session.session_id.value
run_id = session.run_id.value
store = JournalTrajectoryStore(journal, read_only=True)
trajectory = store.read_session(session_id)
assert trajectory.records
assert all(record.session_id == session_id for record in trajectory.records)
assert all(record.run_id == run_id for record in trajectory.records)
assert all(resource.closed == 1 for resource in resources)
return {'interrupted': interrupted, 'requests': len(requests),
'executions': len(executions), 'completion_order': completions,
'final': result.state.final_result}
if __name__ == '__main__':
print(json.dumps([run(), run(interrupted=True)], ensure_ascii=False))
下一步与 API
API Reference · Configuration · Learning path · Next Source file (可选；全部所需代码已在本页)原生模型 I/O 机制验证
新增的consumer.py 使用内置 OpenAI-compatible adapter、离线 SDK transport、公开 composition 和 Session 以及真实并行工具。它断言三次请求、三次执行、首批逆序完成、声明顺序结果、最终结果 11 及 journal 身份一致。第二批流中断时不执行该工具，也不产生成功 final。这证明框架机制，不代表真实模型能力。请先安装 qitos[openai]，再从源码仓库外运行。
Stream 构造失败抛出经过净化的 typed failure，不会变成回答文本。原生 stream 必须收到明确终态，收集尾部 usage 一次，并在正常、异常及 consumer close 时清理自有资源。缺失 usage 保持 unknown。Python worker 取消仍是协作式。