配置真实模型

Use the same notes project with explicit credentials and bounded requests.

R1 context 配置支持以下严格 YAML。budget 通过现有显式 extensions factory 边界解析；不持久化 callable、client 或凭据值。未启用 loss 时普通无损请求仍可运行，需要损失授权的请求在 provider dispatch 前失败。

context:
  budget_policy: budget
  allow_codec_loss: true

本章继续同一个资料项目。默认只校验配置；必须显式传 --live 才读取凭据并请求模型。本轮文档验收不会执行该选项。先完成 Quickstart 安装；独立开始时，创建空目录并保存本页全部文件。

​

依赖和配置

python -m pip install "qitos[openai] @ git+https://github.com/WhitzardAgent/WhitzardOS.git@f7d4b2d666a156d361da496a41868278f84ffabf"

将 real_agent.yaml 的 example-model 和 https://provider.example/v1 替换为你的模型和 endpoint；此保留域名不是服务地址。本例使用受信任纯函数，不需要 Docker。若引入文件/命令工具，必须先按 sandbox 章节配置 Docker，不能直接扩展这里的 unsafe_host 工具权限。

预算为 6 次模型请求、每次最多 512 输出 token、6 steps、30 秒 runtime；provider timeout 30 秒、retries 0。预算限制不承诺任务成功，timeout 也不是硬取消。请分别确认文本协议、工具往返、continuation 与 loss policy 兼容性。

​

私有 credentials 文件

mkdir -p "$HOME/.config/qitos"
chmod 700 "$HOME/.config/qitos"
touch "$HOME/.config/qitos/credentials.yaml"
chmod 600 "$HOME/.config/qitos/credentials.yaml"

用本地编辑器在上面的文件中填写以下结构。必须是当前用户拥有的普通非符号链接文件，父目录权限 0700、文件权限 0600。不要放进项目、Git、命令历史或截图。

credentials:
  notes-provider: REPLACE_IN_PRIVATE_EDITOR

model.credential.ref 对应 notes-provider，配置只保存 CredentialRef。LocalCredentialFileResolver 在组合边界显式解析；环境变量不是默认路径。生成模板的 credentials 示例可能缺少所需根键，请使用这里的完整结构。

​

先验证，再显式运行

python real_notes.py

configuration valid; no credentials read; no model request

下面的命令会产生真实模型请求，需你自己配置后执行。断言检查实际工具结果，不以模型自称完成为依据。

python real_notes.py --live --credentials "$HOME/.config/qitos/credentials.yaml"

本项目动态注册 summarize_note，因此启动入口是该 Python 文件。qit run --config 不会凭空导入你的扩展；仅使用内置工具的配置可使用 CLI，详见 CLI reference。

​

常见错误、限制与练习

未知配置键、权限不符、缺少引用、provider/codec 不兼容应保留 typed failure。不要把 key 写进 YAML 或放宽 loss 检查。连续运行会新建 Session；恢复必须使用匹配的配置与 resolver。练习：修改 YAML 中 request 的 max_tokens，再加载并读取该值；这一步不需要凭据。

​

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

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

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

real_agent.yaml

schema: qitos.agent
agent:
  name: notes_agent
  protocol: react_text_v1
model:
  provider: openai_compatible
  model: example-model
  base_url: https://provider.example/v1
  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: ./real-notes-run/sessions.sqlite3
  trajectory:
    enabled: true
    output: ./real-notes-run/trajectory.journal
budgets:
  max_steps: 6
  max_requests: 6
  max_runtime_seconds: 30
failure_policy:
  tool: fail_closed

real_notes.py

"""Validate configuration by default; --live explicitly opts into model requests."""
import argparse
from pathlib import Path

from notes import summarize_note
from qitos.config import LocalCredentialFileResolver, build_agent_composition, load_agent_config


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--live", action="store_true")
    parser.add_argument("--credentials", type=Path, default=Path.home() / ".config/qitos/credentials.yaml")
    args = parser.parse_args()
    config = load_agent_config(Path(__file__).with_name("real_agent.yaml"))
    assert config.budgets.max_requests == 6
    if not args.live:
        print("configuration valid; no credentials read; no model request")
        return
    if config.model.base_url == "https://provider.example/v1":
        raise ValueError("Replace the reserved provider.example endpoint and example-model first")
    resolver = LocalCredentialFileResolver(args.credentials, repository_root=Path.cwd())
    with build_agent_composition(config, credential_resolver=resolver) as composition:
        composition.tool_registry.register(summarize_note)
        result = composition.session(
            "Call summarize_note for indices 0 and 1. Report both titles and word counts."
        ).run()
        print(composition.config.name, result.state.stop_reason, result.state.final_result)
        outputs = [a.output for record in result.records for a in record.action_results
                   if a.tool_name == "summarize_note" and a.status == "success"]
        assert {item["title"] for item in outputs} == {"Session", "Artifact"}, outputs


if __name__ == "__main__":
    main()

Composition API · Sandbox · CLI · Next: extensions

Source file