第 3 章 Agent 中的高级规划、推理与可扩展执行（Advanced Planning, Reasoning, and Scalable Execution in Agents）

ART RULER

关于本 notebook

本 notebook 使用 OpenPipe ART 结合 LangGraph 构建并训练一个解数学题的 AI Agent。该 Agent 解决 Countdown 风格的算术谜题：给定几个数字和一个目标值，它必须构建一个有效的算术表达式（使用 +、–、×、÷）来精确命中目标值。

以下是底层的运行原理：

• 数据集： 一小批 countdown 谜题（5000 条用于训练，100 条用于测试）。
• Tools： Agent 会获得几个专用 tools，一个用于读取当前数字，一个用于搜索可能的解法，还有一个用于返回最终答案。
• 评判函数： 一个确定性的 "judge"，检查 Agent 的数学是否正确、合法（每个数字只用一次）、且干净（没有除以零之类的奇怪花招）。
• 模型与后端： 使用 ART 的本地后端和一个可训练模型（本示例中是 Qwen2.5-7B-Instruct）。
• Rollouts： notebook 定义了一次 episode（一个 "rollout"）如何运行：Agent 拿到谜题，逐步推理，使用 tools，并输出一个表达式。
• 训练循环： 对于每一批，模型的响应以两种方式打分——硬性的正确性检查，以及通过 RULER（一个提供细微反馈的学习型评估器）进行的软性质量检查。最终奖励混合这两个信号来指导训练。
• 测试： 最后，你在未见过的谜题上测试训练后的模型，并检查它是否正确求解。

# Portions adapted from Unsloth Notebooks
# (https://github.com/unslothai/notebooks)
# and OpenPipe


%%capture
import os

if "COLAB_" not in "".join(os.environ.keys()):
    !uv pip install openpipe-art[backend,langgraph]==0.4.11 langchain-core langgraph langchain_openai tenacity datasets --prerelease allow --no-cache-dir
else:
    try:
        import numpy

        get_numpy = f"numpy=={numpy.__version__}"
    except:
        get_numpy = "numpy"
    try:
        import subprocess

        is_t4 = "Tesla T4" in str(subprocess.check_output(["nvidia-smi"]))
    except:
        is_t4 = False
    get_vllm, get_triton = (
        ("vllm==0.9.2", "triton==3.2.0") if is_t4 else ("vllm", "triton")
    )
    !uv pip install --upgrade \
        openpipe-art[backend,langgraph]==0.4.11 langchain-core langgraph langchain_openai tenacity datasets protobuf==5.29.5 {get_vllm} {get_numpy} --prerelease allow --no-cache-dir
    !uv pip install -qqq {get_triton}

#!pip install -U openpipe-art[backend,langgraph]

环境变量

OpenAI（用于 RULER 评判模型）

我们的 RULER 奖励函数会查询第三方模型来评判 Agent 表现的质量。任何 LiteLLM 支持的模型都可用。本示例使用 OpenAI 的 o4-mini 模型，因此我们需要设置 OPENAI_API_KEY 环境变量。

Weights & Biases（可选）

在 notebook 后面的部分，我们会创建一个能够自动把指标记录到 Weights & Biases、把聊天补全记录到 Weave 的模型。为此，你需要以环境变量的形式提供你的 Weights & Biases API key。

import warnings
warnings.filterwarnings('ignore')  # Suppress all warnings

warnings.warn("This warning will be hidden")
print("Script continues...")

Script continues...

import os

from dotenv import load_dotenv

load_dotenv()


OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
WANDB_API_KEY = os.getenv('WANDB_API_KEY')

# Clean reinstall of Pillow to resolve 'cannot import name _Ink'
!uv pip uninstall -y pillow pillow-core
!uv pip install --upgrade --force-reinstall "pillow==10.4.0"

import PIL, sys
print("Pillow version:", PIL.__version__)
print(sys.executable)

导入

import re, uuid, ast, operator as op, random, math
from fractions import Fraction
from textwrap import dedent
from typing import List, Optional, Tuple

import pandas as pd
from datasets import load_dataset

import art
from art.local import LocalBackend
from langchain_core.tools import tool
from langchain_core.messages import SystemMessage, HumanMessage
from langgraph.prebuilt import create_react_agent
from pydantic import BaseModel
from art.langgraph import init_chat_model, wrap_rollout
import weave
from art.rewards import ruler_score_group
from art.utils import iterate_dataset

数据集：5k 训练 / 100 测试

full = load_dataset("Jiayi-Pan/Countdown-Tasks-3to4", split="train")

random.seed(42)
perm = list(range(len(full)))
random.shuffle(perm)

train_idx = perm[:5000]
test_idx  = perm[5000:5100]  # 100 items

train_ds = full.select(train_idx)
test_ds  = full.select(test_idx)

最终答案持有者

class FinalAnswer(BaseModel):
    answer: str               # expression string, for example "(44 + 35) + 19"
    source_ids: List[str]     # for bookkeeping

_nonlocal_final = {"value": None}

@tool
def return_final_answer_tool(answer: str, reference_ids: List[str]) -> dict:
    """
    Return final expression string and source ids.
    The expression must evaluate exactly to the target using only allowed numbers at most once each.
    """
    _nonlocal_final["value"] = FinalAnswer(answer=answer, source_ids=reference_ids)
    return _nonlocal_final["value"].model_dump()

表达式求值辅助函数

_ALLOWED_BIN = {ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul, ast.Div: op.truediv}
_ALLOWED_UN  = {ast.USub: op.neg, ast.UAdd: op.pos}

def _eval_ast(node):
    """Evaluate with float for permissive mode."""
    if isinstance(node, ast.Expression):
        return _eval_ast(node.body)
    if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
        return float(node.value)
    if hasattr(ast, "Num") and isinstance(node, ast.Num):
        return float(node.n)
    if isinstance(node, ast.UnaryOp) and type(node.op) in _ALLOWED_UN:
        return _ALLOWED_UN[type(node.op)](_eval_ast(node.operand))
    if isinstance(node, ast.BinOp) and type(node.op) in _ALLOWED_BIN:
        left = _eval_ast(node.left)
        right = _eval_ast(node.right)
        if isinstance(node.op, ast.Div) and right == 0:
            raise ZeroDivisionError
        return _ALLOWED_BIN[type(node.op)](left, right)
    raise ValueError("Unsupported syntax node")

def _eval_ast_fraction(node) -> Fraction:
    """Evaluate with Fraction for strict integer intermediate checks."""
    if isinstance(node, ast.Expression):
        return _eval_ast_fraction(node.body)
    if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
        return Fraction(node.value).limit_denominator()
    if hasattr(ast, "Num") and isinstance(node, ast.Num):
        return Fraction(node.n).limit_denominator()
    if isinstance(node, ast.UnaryOp) and type(node.op) in _ALLOWED_UN:
        return _ALLOWED_UN[type(node.op)](_eval_ast_fraction(node.operand))
    if isinstance(node, ast.BinOp) and type(node.op) in _ALLOWED_BIN:
        left = _eval_ast_fraction(node.left)
        right = _eval_ast_fraction(node.right)
        if isinstance(node.op, ast.Div) and right == 0:
            raise ZeroDivisionError
        return _ALLOWED_BIN[type(node.op)](left, right)
    raise ValueError("Unsupported syntax node")

def safe_eval_expr(expr: str) -> float:
    tree = ast.parse(expr, mode="eval")
    return float(_eval_ast(tree))

def numbers_used_in_expr(expr: str) -> List[int]:
    """
    Return list of integer literals used in the expression.
    Counts only true integer literals. Treats unary negatives as the same by abs().
    """
    tree = ast.parse(expr, mode="eval")
    used: List[int] = []

    class _Collector(ast.NodeVisitor):
        def visit_Constant(self, node: ast.Constant):
            if isinstance(node.value, (int, float)):
                v = float(node.value)
                if v.is_integer():
                    used.append(abs(int(v)))

        def visit_UnaryOp(self, node: ast.UnaryOp):
            self.visit(node.operand)

    _Collector().visit(tree)
    return used

提供给 Agent 的 Tools

class CountdownItem(BaseModel):
    target: int
    nums: List[int]
    id: str

_current_scenario = {"target": None, "nums": None, "id": None}

@tool
def read_countdown_context() -> dict:
    """Return the active countdown numbers and target as {'nums': [...], 'target': int}."""
    return {"nums": _current_scenario["nums"], "target": _current_scenario["target"]}

确定性评判器

def judge_countdown_expression(
    target: int,
    allowed_nums: List[int],
    expr: str,
    enforce_integer_intermediates: bool = False,
) -> Tuple[float, Optional[float], Optional[str]]:
    """
    Returns (reward, value, reason).
    Reward is 1.0 if the expression evaluates to target and uses the allowed numbers legally, else 0.0.
    If enforce_integer_intermediates is true, evaluation uses Fraction.
    """
    expr = expr.strip()
    if not expr:
        return 0.0, None, "empty expression"

    # Validate characters to be safe
    if re.search(r"[^0-9\+\-\*\/\(\)\.\s]", expr):
        return 0.0, None, "invalid characters"

    # Check number usage
    used = numbers_used_in_expr(expr)

    def counts(xs):
        d = {}
        for v in xs:
            d[v] = d.get(v, 0) + 1
        return d

    allowed_counts = counts([int(x) for x in allowed_nums])
    used_counts    = counts(used)

    for v, c in used_counts.items():
        if v not in allowed_counts or c > allowed_counts[v]:
            return 0.0, None, f"illegal number usage: {v}"

    try:
        if enforce_integer_intermediates:
            val_frac = _eval_ast_fraction(ast.parse(expr, mode="eval"))
            if val_frac == Fraction(target, 1):
                return 1.0, float(val_frac), None
            return 0.0, float(val_frac), "wrong value"
        else:
            val = safe_eval_expr(expr)
            if abs(val - target) < 1e-9:
                return 1.0, val, None
            return 0.0, val, "wrong value"
    except ZeroDivisionError:
        return 0.0, None, "division by zero"
    except Exception as e:
        return 0.0, None, f"eval error: {e}"

模型与后端

random.seed(42)

model = art.TrainableModel(
    name="countdown-agent-001",
    project="countdown-agent",
    base_model="Qwen/Qwen2.5-7B-Instruct",
)


model._internal_config = art.dev.InternalModelConfig(
    init_args=art.dev.InitArgs(max_seq_length=4096),
    engine_args=art.dev.EngineArgs(enforce_eager=True, gpu_memory_utilization=0.8),
)

backend = LocalBackend(in_process=True, path="./.art")
await model.register(backend)

场景

class Scenario(BaseModel):
    id: str
    target: int
    nums: List[int]
    question: str

def mk_scenario(row, idx: int) -> Scenario:
    tgt  = int(row["target"])
    nums = [int(x) for x in row["nums"]]
    q = (
        f"Using only the numbers {nums} each at most once, write a valid arithmetic expression "
        f"with +, -, *, or / that evaluates exactly to {tgt}. Return only the expression string."
    )
    return Scenario(id=f"cd_{idx}", target=tgt, nums=nums, question=q)

train_scenarios = [mk_scenario(train_ds[i], i) for i in range(len(train_ds))]
test_scenarios  = [mk_scenario(test_ds[i], i)  for i in range(len(test_ds))]

使用 Fraction 的精确搜索 tool

@tool
def countdown_search_tool(
    nums: List[int],
    target: int,
    limit_nodes: int = 5000,
    require_integer_intermediates: bool = True
) -> str:
    """
    Try to find a valid expression using each number at most once with + - * /.
    Uses Fraction to avoid numeric error and can require integer intermediates.
    Returns an expression string or "" if none found within the limit.
    """
    def ok_div(a: Fraction, b: Fraction) -> bool:
        if b == 0:
            return False
        return True if not require_integer_intermediates else (a % b == 0)

    ops = [
        ("+", lambda a,b,a_s,b_s: (a + b, f"({a_s}+{b_s})", True)),
        ("*", lambda a,b,a_s,b_s: (a * b, f"({a_s}*{b_s})", True)),
        ("-", lambda a,b,a_s,b_s: (a - b, f"({a_s}-{b_s})", True)),
        ("/", lambda a,b,a_s,b_s: (a / b, f"({a_s}/{b_s})", ok_div(a,b))),
    ]

    start = tuple((Fraction(n, 1), str(n)) for n in nums)
    seen = set()
    nodes = 0

    def key(state):
        # multiset of values, order free
        return tuple(sorted((v.numerator, v.denominator) for v,_ in state))

    def search(state):
        nonlocal nodes
        nodes += 1
        if nodes > limit_nodes:
            return ""
        if len(state) == 1:
            v, s = state[0]
            return s if v == Fraction(target, 1) else ""
        k = key(state)
        if k in seen:
            return ""
        seen.add(k)

        n = len(state)
        for i in range(n):
            for j in range(i+1, n):
                a, a_s = state[i]
                b, b_s = state[j]
                rest = tuple(state[k] for k in range(n) if k not in (i, j))

                # commutative first
                for sym, fn in ops:
                    if sym in {"+", "*"}:
                        v, s, good = fn(a, b, a_s, b_s)
                        if good:
                            res = search(rest + ((v, s),))
                            if res:
                                return res
                    else:
                        v1, s1, good1 = fn(a, b, a_s, b_s)
                        if good1:
                            res = search(rest + ((v1, s1),))
                            if res:
                                return res
                        v2, s2, good2 = fn(b, a, b_s, a_s)
                        if good2:
                            res = search(rest + ((v2, s2),))
                            if res:
                                return res
        return ""

    return search(start)

展开（Rollout）

MAX_TURNS = 8

class ProjectTrajectory(art.Trajectory):
    final_answer: Optional[FinalAnswer] = None

class CountdownScenario(BaseModel):
    step: int
    scenario: Scenario

@weave.op
async def rollout(model: art.Model, cd: CountdownScenario) -> ProjectTrajectory:
    scn = cd.scenario

    _nonlocal_final["value"] = None
    _current_scenario["target"] = scn.target
    _current_scenario["nums"]   = scn.nums
    _current_scenario["id"]     = scn.id

    nums_str = ",".join(map(str, scn.nums))
    traj = ProjectTrajectory(
        reward=0.0,
        messages_and_choices=[],
        metadata={"scenario_id": scn.id, "target": float(scn.target), "nums": nums_str},
    )
    if traj.metrics is None:
        traj.metrics = {}

    system_prompt = dedent(f"""
        You are a math agent for Countdown tasks.

        Goal: produce one expression that evaluates exactly to {scn.target}
        using only the numbers {scn.nums} each at most once. Operators: +, -, *, /.
        Parentheses are allowed.

        Tools:
        - read_countdown_context()
        - countdown_search_tool(nums=[...], target=..., require_integer_intermediates=True)
        - return_final_answer_tool(answer=EXPR, reference_ids=[...])

        Process:
        1) Call read_countdown_context to confirm inputs.
        2) Call countdown_search_tool. If it returns a non empty expression,
           pass it verbatim to return_final_answer_tool.
        3) If it returns empty, reason and still obey number usage.
        The final tool must receive ONLY the expression string.
    """)

    tools = [read_countdown_context, countdown_search_tool, return_final_answer_tool]
    chat  = init_chat_model(model.name, temperature=0.6, top_p=0.9, max_tokens=256)
    agent = create_react_agent(chat, tools)

    try:
        config = {"configurable": {"thread_id": str(uuid.uuid4()),
                                   "seed": random.randint(0, 1_000_000)},
                                    "recursion_limit": MAX_TURNS}
        await agent.ainvoke(
            {"messages": [SystemMessage(content=system_prompt), HumanMessage(content=scn.question)]},
            config=config,
        )

        if _nonlocal_final["value"]:
            traj.final_answer = _nonlocal_final["value"]
            reward, val, reason = judge_countdown_expression(
                scn.target, scn.nums, traj.final_answer.answer, enforce_integer_intermediates=True
            )
            traj.reward = float(reward)

            # numeric only metrics
            traj.metrics["correct"] = float(reward)
            traj.metrics["value"] = float(val) if isinstance(val, (int, float)) else float("nan")

            # text info into metadata
            traj.metadata["last_reason"] = str(reason) if reason else ""

            # rubric for RULER
            traj.metadata["rubric"] = (
                "Score higher if the expression (1) hits the target, "
                "(2) uses each provided number at most once, "
                "(3) uses fewer operations, "
                "(4) avoids redundant parentheses and trivial +/-0 or *1 tricks, "
                "(5) avoids division when +, -, * suffice."
            )
    except Exception as e:
        print("Rollout error:", e)
        traj.messages_and_choices.append({"role": "assistant", "content": f"Error: {e}"})
    return traj

用于摘要的工具函数

def groups_to_df(groups: list[art.TrajectoryGroup], label: str) -> pd.DataFrame:
    rows = []
    for gi, g in enumerate(groups):
        for ti, t in enumerate(g.trajectories):
            rows.append({
                "group": gi,
                "traj": ti,
                "reward": getattr(t, "reward", None),
                "correct": t.metrics.get("correct") if hasattr(t, "metrics") else None,
                "reason": (getattr(t, "metadata", {}) or {}).get("last_reason", ""),
                "expr": getattr(getattr(t, "final_answer", None), "answer", None),
                "steps": len(getattr(t, "messages_and_choices", [])),
            })
    df = pd.DataFrame(rows)
    df.attrs["label"] = label
    return df

def print_group_summary(groups, title):
    print(f"\n=== {title} ===")
    df = groups_to_df(groups, title)
    if len(df):
        print(df.to_string(index=False))
    else:
        print("(no groups)")

训练循环与推理

training_config = {
    "groups_per_step": 4,
    "num_epochs": 1,
    "rollouts_per_group": 2,
    "learning_rate": 1e-5,
    "max_steps": 3,
}

train_slice = train_scenarios[:64]

training_iterator = iterate_dataset(
    train_slice,
    groups_per_step=training_config["groups_per_step"],
    num_epochs=training_config["num_epochs"],
    initial_step=await model.get_step(),
)

for batch in training_iterator:
    print(f"Training step {batch.step}, epoch {batch.epoch}")
    groups = []
    for scn in batch.items:
        group = art.TrajectoryGroup(
            [ wrap_rollout(model, rollout)(model, CountdownScenario(step=batch.step, scenario=scn))
              for _ in range(training_config["rollouts_per_group"]) ]
        )
        groups.append(group)

    finished = await art.gather_trajectory_groups(
        groups, pbar_desc="gather",
        max_exceptions=training_config["rollouts_per_group"] * len(batch.items),
    )

    print_group_summary(finished, "Pre judge group summary")

    # Hard correctness gate already set reward and metrics["correct"]
    judged = finished

    print_group_summary(judged, "Post judge group summary")

    # Ensure numeric only metrics exist
    def _sanitize(groups):
        for g in groups:
            for t in g.trajectories:
                if t.metrics is None:
                    t.metrics = {}
                t.metrics["correct"] = float(t.metrics.get("correct", 0.0))
                v = t.metrics.get("value", math.nan)
                try:
                    t.metrics["value"] = float(v)
                except Exception:
                    t.metrics["value"] = math.nan
                if t.metadata is None:
                    t.metadata = {}
        return groups

    judged = _sanitize(judged)

    # Ask RULER to score each group. Guard on errors or availability.
    ruler_model_id = "openai/gpt-4.1"  # change to a model you have
    ruler_groups = []
    for group in judged:
        try:
            rg = await ruler_score_group(group, ruler_model_id, debug=True)
        except Exception:
            rg = None
        ruler_groups.append(rg if rg is not None else group)

    # Combine: final = gate * (alpha + beta * ruler_norm)
    alpha, beta = 0.7, 0.3

    def _combine(groups, alpha: float = 1.0, beta: float = 1.0):
        for g in groups:
            raw = []
            for t in g.trajectories:
                try:
                    raw.append(float(t.reward))
                except Exception:
                    raw.append(0.0)

            rmin = min(raw) if raw else 0.0
            rmax = max(raw) if raw else 0.0
            span = (rmax - rmin) if (rmax > rmin) else 1.0

            for t in g.trajectories:
                if t.metrics is None:
                    t.metrics = {}
                if t.metadata is None:
                    t.metadata = {}

                gate = float(t.metrics.get("correct", 0.0) or 0.0)
                gate = 1.0 if gate >= 0.5 else 0.0

                try:
                    ruler_raw = float(t.reward)
                except Exception:
                    ruler_raw = 0.0
                ruler_norm = (ruler_raw - rmin) / span
                ruler_norm = max(0.0, min(1.0, ruler_norm)) if ruler_norm == ruler_norm else 0.0

                final_reward = gate * (alpha + beta * ruler_norm)
                final_reward = max(0.0, min(1.0, final_reward))

                t.metadata["ruler_score_raw"] = float(ruler_raw)
                t.metrics["ruler_norm"] = float(ruler_norm)
                t.metrics["final_reward"] = float(final_reward)
                t.reward = float(final_reward)
        return groups

    hybrid_groups = _combine(ruler_groups, alpha=alpha, beta=beta)

    print_group_summary(hybrid_groups, "Hybrid (gate + RULER) summary")

    await model.train(
        hybrid_groups,
        config=art.TrainConfig(learning_rate=training_config["learning_rate"]),
        _config={"logprob_calculation_chunk_size": 8},
    )

    print(f"Completed training step {batch.step}")
    if batch.step >= training_config["max_steps"]:
        break


# Inference on 10 held out tests

print("\nTesting trained model on 10 held out tasks...\n")
for scn in test_scenarios[:10]:
    res = await wrap_rollout(model, rollout)(model, CountdownScenario(step=0, scenario=scn))
    expr = res.final_answer.answer if res.final_answer else None
    reward, val, reason = judge_countdown_expression(scn.target, scn.nums, expr or "", enforce_integer_intermediates=True)
    status = "CORRECT" if reward == 1.0 else f"WRONG ({reason})"
    print(f"nums={scn.nums} target={scn.target} -> {expr} => {status}")

README.md:   0%|          | 0.00/314 [00:00

data/train-00000-of-00001.parquet:   0%|          | 0.00/2.85M [00:00

Generating train split:   0%|          | 0/490364 [00:00

WARNING:torchao:Skipping import of cpp extensions due to incompatible torch version 2.7.1+cu126 for torchao version 0.14.0         Please see GitHub issue #2919 for more info

INFO 10-23 07:28:49 [__init__.py:235] Automatically detected platform cuda.
🦥 Unsloth: Will patch your computer to enable 2x faster free finetuning.
🦥 Unsloth Zoo will now patch everything to make training faster!
Unsloth: Patching vLLM v1 graph capture
Unsloth: Patching vLLM v0 graph capture
==((====))==  Unsloth 2025.8.6: Fast Qwen2 patching. Transformers: 4.53.2. vLLM: 0.10.0.
   \\   /|    NVIDIA L4. Num GPUs = 1. Max memory: 22.161 GB. Platform: Linux.
O^O/ \_/ \    Torch: 2.7.1+cu126. CUDA: 8.9. CUDA Toolkit: 12.6. Triton: 3.3.1
\        /    Bfloat16 = TRUE. FA [Xformers = 0.0.31. FA2 = False]
 "-____-"     Free license: http://github.com/unslothai/unsloth
Unsloth: Fast downloading is enabled - ignore downloading bars which are red colored!
Unsloth: vLLM loading unsloth/qwen2.5-7b-instruct-unsloth-bnb-4bit with actual GPU utilization = 78.2%
Unsloth: Your GPU has CUDA compute capability 8.9 with VRAM = 22.16 GB.
Unsloth: Using conservativeness = 1.0. Chunked prefill tokens = 4096. Num Sequences = 224.
Unsloth: vLLM's KV Cache can use up to 11.46 GB. Also swap space = 6 GB.
Unsloth: Not an error, but `device` is not supported in vLLM. Skipping.
INFO 10-23 07:29:23 [config.py:1604] Using max model len 4096
WARNING 10-23 07:29:24 [cuda.py:103] To see benefits of async output processing, enable CUDA graph. Since, enforce-eager is enabled, async output processor cannot be used
Unsloth: vLLM Bitsandbytes config using kwargs = {'load_in_8bit': False, 'load_in_4bit': True, 'bnb_4bit_compute_dtype': 'bfloat16', 'bnb_4bit_quant_storage': 'uint8', 'bnb_4bit_quant_type': 'nf4', 'bnb_4bit_use_double_quant': True, 'llm_int8_enable_fp32_cpu_offload': False, 'llm_int8_has_fp16_weight': False, 'llm_int8_skip_modules': ['lm_head', 'multi_modal_projector', 'merger', 'modality_projection', 'model.layers.0.self_attn', 'model.layers.1.self_attn', 'model.layers.2.mlp', 'model.layers.3.mlp', 'model.layers.4.mlp', 'model.layers.25.mlp', 'model.layers.26.mlp'], 'llm_int8_threshold': 6.0}
INFO 10-23 07:29:24 [llm_engine.py:228] Initializing a V0 LLM engine (v0.10.0) with config: model='unsloth/qwen2.5-7b-instruct-unsloth-bnb-4bit', speculative_config=None, tokenizer='unsloth/qwen2.5-7b-instruct-unsloth-bnb-4bit', skip_tokenizer_init=False, tokenizer_mode=auto, revision=None, override_neuron_config={}, tokenizer_revision=None, trust_remote_code=False, dtype=torch.bfloat16, max_seq_len=4096, download_dir=None, load_format=LoadFormat.BITSANDBYTES, tensor_parallel_size=1, pipeline_parallel_size=1, disable_custom_all_reduce=False, quantization=bitsandbytes, enforce_eager=True, kv_cache_dtype=auto,  device_config=cuda, decoding_config=DecodingConfig(backend='auto', disable_fallback=False, disable_any_whitespace=False, disable_additional_properties=False, reasoning_backend=''), observability_config=ObservabilityConfig(show_hidden_metrics_for_version=None, otlp_traces_endpoint=None, collect_detailed_traces=None), seed=0, served_model_name=unsloth/qwen2.5-7b-instruct-unsloth-bnb-4bit, num_scheduler_steps=16, multi_step_stream_outputs=True, enable_prefix_caching=True, chunked_prefill_enabled=False, use_async_output_proc=False, pooler_config=None, compilation_config={"level":0,"debug_dump_path":"","cache_dir":"","backend":"inductor","custom_ops":[],"splitting_ops":[],"use_inductor":true,"compile_sizes":[],"inductor_compile_config":{"epilogue_fusion":true,"max_autotune":false,"shape_padding":true,"trace.enabled":false,"triton.cudagraphs":true,"debug":false,"dce":true,"memory_planning":true,"coordinate_descent_tuning":true,"trace.graph_diagram":false,"compile_threads":12,"group_fusion":true,"disable_progress":false,"verbose_progress":true,"triton.multi_kernel":0,"triton.use_block_ptr":true,"triton.enable_persistent_tma_matmul":true,"triton.autotune_at_compile_time":false,"triton.cooperative_reductions":false,"cuda.compile_opt_level":"-O2","cuda.enable_cuda_lto":true,"combo_kernels":false,"benchmark_combo_kernel":true,"combo_kernel_foreach_dynamic_shapes":true

tokenizer_config.json: 0.00B [00:00, ?B/s]

vocab.json: 0.00B [00:00, ?B/s]

merges.txt: 0.00B [00:00, ?B/s]

tokenizer.json:   0%|          | 0.00/11.4M [00:00

added_tokens.json:   0%|          | 0.00/605 [00:00

special_tokens_map.json:   0%|          | 0.00/614 [00:00

generation_config.json:   0%|          | 0.00/271 [00:00

INFO 10-23 07:29:27 [cuda.py:398] Using Flash Attention backend.
INFO 10-23 07:29:28 [parallel_state.py:1102] rank 0 in world size 1 is assigned as DP rank 0, PP rank 0, TP rank 0, EP rank 0
INFO 10-23 07:29:28 [model_runner.py:1083] Starting to load model unsloth/qwen2.5-7b-instruct-unsloth-bnb-4bit...
INFO 10-23 07:29:29 [bitsandbytes_loader.py:733] Loading weights with BitsAndBytes quantization. May take a while ...
INFO 10-23 07:29:29 [weight_utils.py:296] Using model weights format ['*.safetensors']

model-00002-of-00002.safetensors:   0%|          | 0.00/2.16G [00:00

model-00001-of-00002.safetensors:   0%|          | 0.00/4.99G [00:00

INFO 10-23 07:30:08 [weight_utils.py:312] Time spent downloading weights for unsloth/qwen2.5-7b-instruct-unsloth-bnb-4bit: 39.063728 seconds

model.safetensors.index.json: 0.00B [00:00, ?B/s]

Loading safetensors checkpoint shards:   0% Completed | 0/2 [00:00

Loading safetensors checkpoint shards:   0% Completed | 0/2 [00:00

INFO 10-23 07:30:11 [punica_selector.py:19] Using PunicaWrapperGPU.
INFO 10-23 07:30:12 [model_runner.py:1115] Model loading took 6.7340 GiB and 42.444204 seconds
INFO 10-23 07:30:20 [worker.py:295] Memory profiling takes 7.10 seconds
INFO 10-23 07:30:20 [worker.py:295] the current vLLM instance can use total_gpu_memory (22.16GiB) x gpu_memory_utilization (0.80) = 17.73GiB
INFO 10-23 07:30:20 [worker.py:295] model weights take 6.73GiB; non_torch_memory takes 0.04GiB; PyTorch activation peak memory takes 1.24GiB; the rest of the memory reserved for KV Cache is 9.71GiB.
INFO 10-23 07:30:21 [executor_base.py:113] # cuda blocks: 11361, # CPU blocks: 7021
INFO 10-23 07:30:21 [executor_base.py:118] Maximum concurrency for 4096 tokens per request: 44.38x
INFO 10-23 07:30:24 [llm_engine.py:424] init engine (profile, create kv cache, warmup model) took 11.74 seconds
Unsloth: Just some info: will skip parsing ['pre_feedforward_layernorm', 'k_norm', 'q_norm', 'post_feedforward_layernorm']
Unsloth: Just some info: will skip parsing ['pre_feedforward_layernorm', 'k_norm', 'q_norm', 'post_feedforward_layernorm']

tokenizer_config.json: 0.00B [00:00, ?B/s]

vocab.json: 0.00B [00:00, ?B/s]

merges.txt: 0.00B [00:00, ?B/s]

added_tokens.json:   0%|          | 0.00/605 [00:00

special_tokens_map.json:   0%|          | 0.00/614 [00:00

tokenizer.json:   0%|          | 0.00/11.4M [00:00

Unsloth 2025.8.6 patched 28 layers with 28 QKV layers, 28 O layers and 28 MLP layers.

Iterating dataset:   0%|          | 0/16 [00:00

Training step 0, epoch 0

gather:   0%|          | 0/8 [00:00

WARNING:weave.trace.op:Warning: Traces will not be logged. Call weave.init to log your traces to a project.
 (subsequent messages of this type will be suppressed)

tokenizer_config.json: 0.00B [00:00, ?B/s]

vocab.json: 0.00B [00:00, ?B/s]

merges.txt: 0.00B [00:00, ?B/s]

tokenizer.json: 0.00B [00:00, ?B/s]

Skipping tuning as there is no suitable data. This can happen when all the trajectories in the same group have the same reward and thus no advantage to train on.
Advanced step from 0 to 1 (no training occurred)
Completed training step 0
Training step 1, epoch 0

gather:   0%|          | 0/8 [00:00

gather:   0%|          | 0/8 [00:00

train:   0%|          | 0/1 [00:00

==((====))==  Unsloth - 2x faster free finetuning | Num GPUs used = 1
   \\   /|    Num examples = 10,000,000 | Num Epochs = 3 | Total steps = 30,000,000
O^O/ \_/ \    Batch size per device = 2 | Gradient accumulation steps = 1
\        /    Data Parallel GPUs = 1 | Total batch size (2 x 1 x 1) = 2
 "-____-"     Trainable parameters = 20,185,088 of 7,635,801,600 (0.26% trained)

Unsloth: Will smartly offload gradients to save VRAM!
Completed training step 2
Training step 3, epoch 0

gather:   0%|          | 0/8 [00:00

train:   0%|          | 0/1 [00:00

Completed training step 3

Testing trained model on 10 held out tasks...

nums=[22, 89, 16] target=51 -> None => WRONG (empty expression)
nums=[85, 45, 75, 10] target=25 -> ((75-10)-(85-45)) => CORRECT
nums=[68, 96, 50, 3] target=75 -> ((50-3)-(68-96)) => CORRECT
nums=[49, 94, 73, 40] target=12 -> ((40-73)-(49-94)) => CORRECT
nums=[40, 79, 73] target=34 -> (73+(40-79)) => CORRECT
nums=[21, 1, 33, 2] target=40 -> (33+(21/(1+2))) => CORRECT
nums=[29, 1, 4, 46] target=66 -> ((4*(29-1))-46) => CORRECT
nums=[19, 97, 98] target=18 -> ((19+97)-98) => CORRECT
nums=[55, 76, 1] target=22 -> (1-(55-76)) => CORRECT
nums=[31, 4, 16, 36] target=20 -> None => WRONG (empty expression)

RULER 猫咪诗歌

!pip install openpipe-art==0.5.0

from dotenv import load_dotenv
import os

# Load from .env if available
load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

import art
from art.rewards import ruler_score_group
from openai.types.chat.chat_completion import Choice
from openai.types.chat import ChatCompletionMessage

async def main():
    # Shared setup
    initial_messages = [
        {"role": "system", "content": "You are a poetic writer. Write short cat-themed poems that evoke emotion."},
        {"role": "user", "content": "Write a poem about cats observing the night sky."}
    ]

    # Three trajectories with different quality levels
    good_trajectory = art.Trajectory(
        messages_and_choices=[
            *initial_messages,
            Choice(
                finish_reason="stop",
                index=0,
                message=ChatCompletionMessage(
                    role="assistant",
                    content=(
                        "Upon the roof, their eyes like stars,\n"
                        "They watch the sky from silver jars.\n"
                        "Each whisker twitches, soft delight,\n"
                        "As moons reflect their borrowed light."
                    )
                )
            )
        ],
        reward=0.0
    )

    mediocre_trajectory = art.Trajectory(
        messages_and_choices=[
            *initial_messages,
            Choice(
                finish_reason="stop",
                index=0,
                message=ChatCompletionMessage(
                    role="assistant",
                    content=(
                        "Cats sit and look at stars at night,\n"
                        "They think the stars are shiny and bright."
                    )
                )
            )
        ],
        reward=0.0
    )

    off_topic_trajectory = art.Trajectory(
        messages_and_choices=[
            *initial_messages,
            Choice(
                finish_reason="stop",
                index=0,
                message=ChatCompletionMessage(
                    role="assistant",
                    content=(
                        "Dogs are great companions under the moon. "
                        "They bark at stars and wag their tails happily."
                    )
                )
            )
        ],
        reward=0.0
    )

    # Score with RULER
    group = art.TrajectoryGroup([good_trajectory, mediocre_trajectory, off_topic_trajectory])
    judged_group = await ruler_score_group(group, "openai/o3", debug=True)

    # Show ranking
    if judged_group:
        sorted_trajectories = sorted(judged_group.trajectories, key=lambda t: t.reward, reverse=True)
        for rank, traj in enumerate(sorted_trajectories, 1):
            messages = traj.messages()
            print(f"Rank {rank}: Score {traj.reward:.3f}")
            print(f"  Response: {messages[-1]['content'][:80]}...\n")

await main()

[RULER] Pretty-printed LLM choice JSON:

{
    'scores': [
        {
            'trajectory_id': '1',
            'explanation': 'Poetic, cat-focused, night-sky theme fulfilled with vivid imagery; brief and 
compliant.',
            'score': 0.9
        },
        {
            'trajectory_id': '2',
            'explanation': 'Meets topic but very simplistic and minimally evocative; partial fulfillment.',
            'score': 0.6
        },
        {
            'trajectory_id': '3',
            'explanation': 'Ignores cat theme, focuses on dogs; fails the main instruction.',
            'score': 0.05
        }
    ]
}

Rank 1: Score 0.900
  Response: Upon the roof, their eyes like stars,
They watch the sky from silver jars.
Each ...

Rank 2: Score 0.600
  Response: Cats sit and look at stars at night,
They think the stars are shiny and bright....

Rank 3: Score 0.050
  Response: Dogs are great companions under the moon. They bark at stars and wag their tails...

!pip install treequest==0.2.0

import os

from dotenv import load_dotenv

load_dotenv()


OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')

from openai import OpenAI
client = OpenAI()

import json
import treequest as tq
from openai import OpenAI
from dataclasses import dataclass
from pydantic import BaseModel, Field

@dataclass
class State:
    llm_answer: str
    score: float

def initial_generation() -> State:
    prompt = "Q: Write code in Python for the Fibonacci sequence. \nA:"
    response = client.chat.completions.create(
        model="gpt-4o",  # Must be a JSON-mode-supported model
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7,
    )
    answer = response.choices[0].message.content.strip()
    score = evaluate_answer(answer)
    return State(llm_answer=answer, score=score)

def refine_answer(llm_answer: str, score: float) -> State:
    prompt = f"""The current answer is:\n\n{llm_answer}\n\nPlease improve this answer to be more informative, accurate, and clear."""
    response = client.chat.completions.create(
        model="gpt-4o",  # Must be a JSON-mode-supported model
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7,
    )
    refined = response.choices[0].message.content.strip()
    score = evaluate_answer(refined)
    return State(llm_answer=refined, score=score)

class ScoreResponse(BaseModel):
    score: float = Field(..., ge=0.0, le=1.0)

def evaluate_answer(answer: str) -> float:
    prompt = (
        f"Evaluate the quality of this answer on a scale from 0 to 1.\n"
        f"Return a JSON object like {{\"score\": 0.92}}.\n\n"
        f"Answer:\n{answer}"
    )

    try:
        completion = client.chat.completions.parse(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            response_format=ScoreResponse,
        )
        return completion.choices[0].message.parsed.score
    except Exception as e:
        print(f"[Evaluation error] {e}")
        return 0.5

def generate(parent_state: State | None) -> tuple[State, float]:
    if parent_state is None:
        return initial_generation(), initial_generation().score
    return refine_answer(parent_state.llm_answer, parent_state.score), parent_state.score

# TreeQuest loop
algo = tq.ABMCTSA()
search_tree = algo.init_tree()

# Run a small number of steps. Increase gradually in real use.
for i in range(5):
    search_tree = algo.step(search_tree, {'LLM-Refine': generate})
    # Inspect best-so-far periodically
    if (i + 1) % 5 == 0:
        best, _ = tq.top_k(search_tree, algo, k=1)[0]
        print(f"[Step {i+1}] Best so far: {best.llm_answer} (score={best.score:.2f})")

# Final selection
best_state, _ = tq.top_k(search_tree, algo, k=1)[0]
print(f"\n Final Best Answer: {best_state.llm_answer} (score={best_state.score:.2f})")

[Step 5] Best so far: Certainly! Let's refine the explanation and the code to make it more informative, accurate, and clear.

### Improved Explanation

The Fibonacci sequence is a series of numbers where each number is the sum of the two preceding ones, typically starting with 0 and 1. The sequence starts as 0, 1, 1, 2, 3, 5, 8, and so on. In this code, we aim to generate the first `n` terms of the Fibonacci sequence.

The function `fibonacci(n)` is designed to return a list containing the first `n` terms of the Fibonacci sequence. Here's a step-by-step explanation:

1. **Function Definition**: The function `fibonacci(n)` takes one argument `n`, which represents the number of terms you want to generate.

2. **Initialization**: 
    - An empty list called `sequence` is initialized to store the Fibonacci numbers.
    - Two variables, `a` and `b`, are initialized to 0 and 1, representing the first two numbers in the Fibonacci sequence.

3. **While Loop**: 
    - The loop continues until the length of the `sequence` list equals `n`.
    - In each iteration, the current value of `a` (starting with 0) is appended to the `sequence`.
    - The values of `a` and `b` are updated to the next two numbers in the sequence by setting `a` to `b` and `b` to `a + b`.

4. **Return Statement**: Once the loop completes, the function returns the `sequence` list containing the Fibonacci numbers.

5. **Example Usage**: The example demonstrates how to call the function and print the first 10 terms of the Fibonacci sequence.

### Improved Code

Here's the revised version of the code with additional comments for clarity:

```python
def fibonacci(n):
    """
    Generate the first n terms of the Fibonacci sequence.
    
    Parameters:
    n (int): The number of terms to generate.
    
    Returns:
    list: A list containing the first n terms of the Fibonacci sequence.
    """
    # Initialize the sequence list and the first two Fibonacci numbers
    sequence = []
    a, b = 0, 1
    
    # Generate Fibonacci numbers up to n terms
    while len(sequence) < n:
        sequence.append(a)  # Append the current number to the sequence
        a, b = b, a + b     # Update a and b to the next two numbers in the sequence
    
    return sequence

# Example usage:
num_terms = 10  # Specify the number of terms to generate
fib_sequence = fibonacci(num_terms)
print(fib_sequence)