> ## Documentation Index
> Fetch the complete documentation index at: https://qitor.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 替换 provider 并添加 context

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

## 目标与前置条件

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。

## 准备项目

```bash theme={null}
python3 -m venv .venv
source .venv/bin/activate
python -m pip install "qitos @ git+https://github.com/WhitzardAgent/WhitzardOS.git@60809b3be388d22ea40ea41b4aaa1f5540c76fda"
mkdir notes_lesson
cd notes_lesson
```

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

## 运行并验证

```bash theme={null}
python provider_extension.py --root extension-run
```

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

```text theme={null}
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 和报告。

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

```python title="notes.py" theme={null}
"""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())
```

```yaml title="agent.yaml" theme={null}
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
```

```python title="provider_extension.py" theme={null}
"""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())
```

## 下一步与 API

[API Reference](/zh/reference/api) · [Configuration](/zh/reference/configuration) · [Learning path](/zh/tutorials/index) · [Next](/zh/quickstart)

[Source file](https://github.com/WhitzardAgent/WhitzardOS/blob/master/examples/tutorials/notes/provider_extension.py) (可选；全部所需代码已在本页)