跳转到主要内容

Documentation Index

Fetch the complete documentation index at: https://qitor.mintlify.app/llms.txt

Use this file to discover all available pages before exploring further.

编写方法模板

QitOS 中的方法模板是实现经典 agentic 推理模式的 Agent + Critic 组合。本指南从零开始介绍如何创建一个方法模板。

方法模板的组成

每个方法模板由以下部分组成:
  1. Recipe 实现qitos/recipes/<method_name>/__init__.py,包含 AgentModule、Critic 和 State
  2. 模板资产templates/<method_name>/,包含配置、文档和脚手架代码
  3. 测试tests/test_<method_name>.py
  4. CLI 注册qitos/cli.py_METHOD_TEMPLATES 字典的条目

目录结构

qitos/recipes/<method_name>/
  __init__.py              # AgentModule + Critic + State dataclass

templates/<method_name>/
  __init__.py              # 空或最小导入
  agent.py                 # Config dataclass + build_registry()
  config.yaml              # 默认配置
  paper.md                 # 模式描述 + QitOS 映射

tests/test_<method_name>.py

第 1 步:定义 State

创建一个跟踪方法特定字段的数据类:
from dataclasses import dataclass, field
from typing import List

@dataclass
class MyMethodState:
    """MyMethod 的跨步骤状态跟踪。"""
    iterations: int = 0
    max_iterations: int = 3
    history: List[str] = field(default_factory=list)
    is_complete: bool = False
State 在执行后存储在 EngineResult.state 中,并且在每一步都可以被 Critic 访问。

第 2 步:实现 AgentModule

继承 AgentModule,实现 build_system_prompt()reduce()
from qitos.core.agent import AgentModule
from qitos.core.decision import Decision

class MyMethodAgent(AgentModule):
    """MyMethod 模式的 Agent。"""

    def build_system_prompt(self, task: str, state: dict | None = None) -> str:
        parts = [
            "你是一个实现 MyMethod 模式的 agent。",
            f"任务: {task}",
        ]
        # 注入状态感知上下文
        if state:
            my_state = MyMethodState(**state) if isinstance(state, dict) else state
            if my_state.history:
                parts.append(f"之前的尝试: {my_state.iterations}")
                parts.append("\n".join(my_state.history))
        return "\n\n".join(parts)

    def reduce(self, decision: Decision, state: dict | None = None) -> dict:
        """从 Decision 中提取状态更新。"""
        my_state = MyMethodState(**state) if state else MyMethodState()
        my_state.iterations += 1
        if decision.thought:
            my_state.history.append(decision.thought)
        return {
            "iterations": my_state.iterations,
            "history": my_state.history,
            "is_complete": my_state.is_complete,
        }

要点

  • build_system_prompt() 接收 task(str)和 state(dict)。使用 state 注入上下文,如反思、提议或任务账本。
  • reduce() 接收当前 Decisionstate,返回状态更新的 dict。引擎将其合并到运行状态中。
  • 始终优雅处理 state=None(第一步)。

第 3 步:实现 Critic

继承 Critic,实现 evaluate() 返回 CriticResult
from qitos.engine.critic import Critic
from qitos.engine.critic_result import CriticResult

class MyMethodCritic(Critic):
    """MyMethod 的 Critic — 完成时停止,失败时重试。"""

    def __init__(self, max_iterations: int = 3, quality_threshold: float = 0.7):
        super().__init__(name="my_method_critic")
        self.max_iterations = max_iterations
        self.quality_threshold = quality_threshold

    def evaluate(self, state, decision, results):
        my_state = MyMethodState(**state) if isinstance(state, dict) else state

        # 检查任务是否完成
        if my_state.is_complete:
            return CriticResult(
                action="stop",
                reason="Task completed successfully",
                score=1.0,
            )

        # 检查迭代预算
        if my_state.iterations >= self.max_iterations:
            return CriticResult(
                action="stop",
                reason=f"Max iterations ({self.max_iterations}) reached",
                score=0.0,
            )

        # 检查结果中的错误 — 带指导重试
        has_error = any(
            hasattr(r, "error") and r.error
            for r in (results or [])
        )
        if has_error:
            return CriticResult(
                action="retry",
                reason="Error detected in results",
                score=0.3,
                instruction_patch="Review the error and try a different approach.",
            )

        # 正常继续
        return CriticResult(
            action="continue",
            reason="Progress being made",
            score=0.5,
        )

CriticResult 动作

动作效果何时使用
continue引擎正常执行正在进展,没有问题
stop引擎停止,返回结果任务完成或预算耗尽
retry引擎重试,注入 instruction_patch检测到错误,需要改变策略

关键字段

  • instruction_patch:注入到下一个 prompt 中指导 agent 的字符串
  • state_patch:合并到 state 中更新方法特定字段的 dict
  • score:浮点数(0-1)表示质量;被 qita 用于可视化

第 4 步:创建模板资产

agent.py

"""MyMethod 模板配置。"""
from dataclasses import dataclass

@dataclass
class MyMethodConfig:
    agent_name: str = "my_method_agent"
    max_iterations: int = 3
    quality_threshold: float = 0.7
    max_steps: int = 15

def build_my_method_registry(config: MyMethodConfig) -> dict:
    return {
        "max_iterations": config.max_iterations,
        "quality_threshold": config.quality_threshold,
        "max_steps": config.max_steps,
    }

config.yaml

name: my_method_template
max_steps: 15
max_iterations: 3
quality_threshold: 0.7
model:
  provider: openai_compatible
  base_url: https://api.siliconflow.cn/v1/
  api_key: ${OPENAI_API_KEY}
  model: Qwen/Qwen3-8B
  model_name: Qwen/Qwen3-8B
  temperature: 0.0
  max_tokens: 2048

paper.md

# MyMethod 模板说明

## 来源思想
原始论文算法和关键洞察的简要描述。

## QitOS 映射
- `MyMethodAgent` 通过 `build_system_prompt()` 管理 ...
- `MyMethodCritic` 检测 ... 并返回 ...
- `MyMethodState` 跟踪 ...

## 与论文的主要差异
- 显著的适配或简化

## 本模板范围
模板涵盖的内容和用户可能添加的扩展

第 5 步:在 CLI 中注册

qitos/cli.py_METHOD_TEMPLATES 中添加:
_METHOD_TEMPLATES = {
    # ... 已有条目 ...
    "my_method": "MyMethod — 模式简要描述",
}
这会让它出现在 qit list-templates 的输出中。

第 6 步:编写测试

最低测试覆盖:
"""MyMethod 模板测试。"""
import pytest
from qitos.recipes.my_method import MyMethodAgent, MyMethodCritic, MyMethodState
from qitos.core.decision import Decision
from qitos.engine.critic_result import CriticResult

class TestMyMethodState:
    def test_default_values(self):
        state = MyMethodState()
        assert state.iterations == 0
        assert state.max_iterations == 3
        assert state.is_complete is False

class TestMyMethodAgent:
    def test_build_system_prompt(self):
        agent = MyMethodAgent(llm=None)
        prompt = agent.build_system_prompt(task="Test task", state=None)
        assert "MyMethod" in prompt
        assert "Test task" in prompt

    def test_build_system_prompt_with_state(self):
        agent = MyMethodAgent(llm=None)
        state = MyMethodState(iterations=2, history=["attempt 1", "attempt 2"])
        prompt = agent.build_system_prompt(task="Test", state=state.__dict__)
        assert "2" in prompt

    def test_reduce_updates_iterations(self):
        agent = MyMethodAgent(llm=None)
        decision = Decision(thought="trying again", actions=[])
        result = agent.reduce(decision, state={"iterations": 1, "history": [], "is_complete": False})
        assert result["iterations"] == 2

class TestMyMethodCritic:
    def test_stop_on_completion(self):
        critic = MyMethodCritic()
        state = MyMethodState(is_complete=True)
        result = critic.evaluate(state.__dict__, None, None)
        assert result.action == "stop"
        assert result.score == 1.0

    def test_stop_on_max_iterations(self):
        critic = MyMethodCritic(max_iterations=2)
        state = MyMethodState(iterations=2)
        result = critic.evaluate(state.__dict__, None, None)
        assert result.action == "stop"

    def test_retry_on_error(self):
        critic = MyMethodCritic()
        state = MyMethodState(iterations=1)
        class ErrResult:
            error = "timeout"
        result = critic.evaluate(state.__dict__, None, [ErrResult()])
        assert result.action == "retry"
        assert result.instruction_patch is not None

    def test_continue_on_progress(self):
        critic = MyMethodCritic()
        state = MyMethodState(iterations=1)
        result = critic.evaluate(state.__dict__, None, [])
        assert result.action == "continue"
每个模板至少 15 个测试,覆盖 Agent、Critic、State 和边界情况。

检查清单

提交方法模板 PR 前:
  • Agent 继承 AgentModule,实现 build_system_prompt()reduce()
  • Critic 继承 Critic,实现 evaluate() 返回 CriticResult
  • State dataclass 包含方法特定字段
  • templates/<method_name>/ 包含 __init__.pyagent.pyconfig.yamlpaper.md
  • 方法名称已添加到 qitos/cli.py_METHOD_TEMPLATES
  • 测试覆盖 Agent、Critic、State 和边界情况(至少 15 个测试)
  • paper.md 解释了从论文到 QitOS 的映射和主要差异
  • pytest tests/test_<method_name>.py 通过
  • 完整测试套件 pytest tests/ -q 无回归

参考模板

模板文件模式
Self-Refineqitos/recipes/self_refine/生成 → 批评 → 改进
Reflexionqitos/recipes/reflexion/行动 → 反思 → 重试
LATSqitos/recipes/lats/蒙特卡洛树搜索
MoAqitos/recipes/moa/并行提议 + 聚合
Magentic-Oneqitos/recipes/magentic_one/编排器 + 专家