从第一性原理构建 Agent：环境、行动与强化学习

Anshuman Mishra 2026-05-21 85.7 KB 𝕏 文章列表

本文旨在从第一性原理出发探讨如何构建和训练 AI Agent。文章批评了过度依赖框架的教程，转而从底层构建了一个极简的“文本转图表”Agent。作者通过 Python 实现了一个包含 Canvas 环境的系统，深入解析了 Agent 如何通过 JSON 动作空间与环境交互，阐述了从监督微调（SFT）到强化学习（RL）的完整闭环，即“提示-行动-环境-奖励-梯度更新”，揭示了模型学习可执行指令逻辑的核心机制。

Agent开发强化学习第一性原理环境构建 LLM 代码实战 RLHF 策略学习 JSON交互后端

# On Building Agents From First Principles **作者**: anshuman **日期**: 2026-05-20T13:30:53.000Z **来源**: [https://x.com/athleticKoder/status/2057091692235481560](https://x.com/athleticKoder/status/2057091692235481560) --- ![Image](/doc/2106/image/img_6221ce934759.jpg) How to define an environment, generate teacher trajectories, fine-tune a student, and improve it with reinforcement learning. Authors: Anshuman Mishra & GPT 5.5 May 20, 2026 [Editor’s note: Arguments are mine; writing and structure were refined with GPT 5.5. This is partly an experiment in using AI to help speed-write technical research blogs from rough notes, while keeping the actual taste, direction, and claims human.] The tutorials on post training start too high up the stack. They begin with a framework. Install this library, define this reward function, run this trainer, watch the reward curve move. That is useful once you already understand what is happening. It is less useful when you are trying to build a mental model of the whole system. I find it more helpful to start lower down. Before there is a trainer, there is an environment. Before there is reinforcement learning, there is an action space. Before there is an agent, there is a policy producing actions that change some state of the world. This post is an attempt to build that picture from first principles. The example will be deliberately small: a text-to-diagram agent. The user asks for a simple diagram, and the model outputs structured JSON actions that create shapes on a canvas. You can think of this as a tiny version of a tldraw-style agent. Instead of clicking around a real editor, the model emits actions like “create a rectangle,” “add a label,” and “connect these two nodes.” The goal is not to build the world’s best diagram agent. The goal is to understand the shape of agent training itself. At a high level, the loop is: prompt -> model action -> environment -> reward -> gradient update Almost every agent-training system is a scaled-up version of this loop. A browser agent, a coding agent, a spreadsheet agent, a robotics planner, a math solver, and a diagramming agent all have the same basic structure. They differ in the environment, the action space, and the reward function. This is the part that is easy to miss when using frameworks. Libraries like TRL, Unsloth, PRIME-RL, verl, OpenRLHF, or custom internal trainers are not magic. They are mostly infrastructure around this loop: batching, rollout generation, distributed inference, reward computation, logging, reference models, clipping, checkpointing, and scaling. The conceptual core is much smaller. ## An Agent Is a Policy Inside an Environment A language model is a distribution over sequences. When we use it as an agent, we are asking this distribution to produce actions instead of ordinary prose. Given an observation, the model emits an action. In a chat model, the observation is the conversation and the action is the assistant message. In a browser agent, the observation might be the DOM and the action might be a click or a keystroke. In a coding agent, the observation might be the repository state and the action might be a patch. In a diagram agent, the observation is the user request and the action is a JSON object describing shapes and connections. So the first question is not “which RL trainer should I use?” The first question is: what is the environment? The environment defines what actions are valid, what happens when those actions are executed, and how success is measured. In ordinary supervised fine-tuning, this environment is often implicit. We show the model examples of good behavior and ask it to imitate them. In reinforcement learning, the environment becomes explicit. The model tries something, the environment responds, and the reward function decides whether that attempt was good. For a diagram agent, a completion is not good merely because it sounds plausible. It is good if the JSON parses, the schema is valid, the canvas accepts the actions, the requested objects appear, the arrows connect the right nodes, and the final layout is understandable. That is the core difference between ordinary chat fine-tuning and agent training. Agent training grounds the model’s output in an executable world. ## The Action Language Let us start with the smallest possible action space. The model must return JSON with an actions array. Each action either creates a shape or connects two shapes. A valid completion might look like this: ``` { "actions": [ { "type": "create_shape", "id": "frontend", "shape": "rectangle", "x": 80, "y": 100, "w": 180, "h": 80, "text": "Frontend" }, { "type": "create_shape", "id": "api", "shape": "rectangle", "x": 340, "y": 100, "w": 180, "h": 80, "text": "API" }, { "type": "connect", "from": "frontend", "to": "api", "text": "request" } ] } ``` This looks simple, but it already contains the essential structure of tool use. The model is no longer just generating text. It is generating instructions that another system will execute. That changes the training problem. The model has to learn not just what to say, but what the environment will accept. It has to create shapes before connecting them. It has to use stable IDs. It has to avoid duplicate IDs. It has to keep coordinates finite. It has to avoid invalid shape types. It has to emit parseable JSON. These are not philosophical details. They determine whether the policy can even enter the valid region of the action space. This is why SFT is often necessary before RL. Before the model can optimize reward, it has to learn the language of the environment. ## A Tiny Environment Without a Framework An environment needs only three things: an input prompt, an action format, and a reward function. Here is a minimal pure-Python canvas environment. It supports rectangles, ellipses, diamonds, text blocks, and arrows. The point is not that this canvas is sophisticated. The point is that it gives us a deterministic world in which model outputs can succeed or fail. ``` # env.py from __future__ import annotations import json import math from dataclasses import dataclass, field from typing import Any ALLOWED_SHAPES = {"rectangle", "ellipse", "diamond", "text"} @dataclass class Shape: id: str shape: str x: float y: float w: float h: float text: str = "" @dataclass class Arrow: source: str target: str text: str = "" @dataclass class Canvas: shapes: dict[str, Shape] = field(default_factory=dict) arrows: list[Arrow] = field(default_factory=list) def create_shape(self, action: dict[str, Any]) -> None: shape_id = require_str(action, "id") shape_type = require_str(action, "shape") if shape_type not in ALLOWED_SHAPES: raise ValueError(f"unknown shape type: {shape_type}") if shape_id in self.shapes: raise ValueError(f"duplicate shape id: {shape_id}") x = require_number(action, "x") y = require_number(action, "y") w = require_number(action, "w") h = require_number(action, "h") if w <= 0 or h <= 0: raise ValueError("shape width and height must be positive") if w > 1000 or h > 1000: raise ValueError("shape too large") self.shapes[shape_id] = Shape( id=shape_id, shape=shape_type, x=x, y=y, w=w, h=h, text=str(action.get("text", "")), ) def connect(self, action: dict[str, Any]) -> None: source = require_str(action, "from") target = require_str(action, "to") if source not in self.shapes: raise ValueError(f"arrow source does not exist: {source}") if target not in self.shapes: raise ValueError(f"arrow target does not exist: {target}") if source == target: raise ValueError("arrow cannot connect a shape to itself") self.arrows.append(Arrow(source=source, target=target, text=str(action.get("text", "")))) def apply(self, action: dict[str, Any]) -> None: action_type = require_str(action, "type") if action_type == "create_shape": self.create_shape(action) elif action_type == "connect": self.connect(action) else: raise ValueError(f"unknown action type: {action_type}") def require_str(obj: dict[str, Any], key: str) -> str: value = obj.get(key) if not isinstance(value, str) or not value: raise ValueError(f"{key} must be a non-empty string") return value def require_number(obj: dict[str, Any], key: str) -> float: value = obj.get(key) if not isinstance(value, int | float) or not math.isfinite(value): raise ValueError(f"{key} must be a finite number") return float(value) def parse_actions(text: str) -> list[dict[str, Any]]: try: data = json.loads(text) except json.JSONDecodeError: start = text.find("{") end = text.rfind("}") if start == -1 or end == -1 or end <= start: raise ValueError("model output does not contain JSON") data = json.loads(text[start : end + 1]) actions = data.get("actions") if not isinstance(actions, list): raise ValueError("missing actions array") if not actions: raise ValueError("actions array is empty") if len(actions) > 40: raise ValueError("too many actions") if not all(isinstance(action, dict) for action in actions): raise ValueError("each action must be an object") return actions def validate_completion(text: str) -> tuple[Canvas | None, list[str]]: errors: list[str] = [] canvas = Canvas() try: actions = parse_actions(text) except Exception as exc: return None, [str(exc)] for i, action in enumerate(actions): try: canvas.apply(action) except Exception as exc: errors.append(f"action {i}: {exc}") return canvas, errors ``` This already gives us the core of a verifier environment. The model emits text. The environment parses that text, executes the action sequence, and returns errors if something goes wrong. A framework would make this more scalable. It would not make it more conceptually different. ## Reward Is Where the Task Lives Once the environment can execute actions, we need to define success. This is where most of the real difficulty lives. The trainer can only optimize the reward you give it. If the reward is mostly syntactic, the model will learn syntax. If the reward measures task satisfaction, the model has a chance of learning useful behavior. If the reward is brittle, the model will eventually find the brittleness. For the toy diagram agent, we can combine a few signals. We can reward outputs that parse, actions that validate, layouts that do not overlap, labels that are present, arrows that connect objects, and labels that cover important words from the user request. ``` # reward.py from __future__ import annotations import re from env import Canvas, validate_completion def score_layout(canvas: Canvas) -> float: if not canvas.shapes: return 0.0 score = 1.0 shapes = list(canvas.shapes.values()) for i, a in enumerate(shapes): for b in shapes[i + 1 :]: ax2, ay2 = a.x + a.w, a.y + a.h bx2, by2 = b.x + b.w, b.y + b.h overlap = not (ax2 < b.x or bx2 < a.x or ay2 < b.y or by2 < a.y) if overlap: score -= 0.15 labeled = sum(1 for shape in shapes if shape.text.strip()) score += 0.1 * min(labeled, 5) score += 0.1 * min(len(canvas.arrows), 5) return max(0.0, min(1.0, score)) def score_semantics(prompt: str, canvas: Canvas) -> float: prompt_words = set(re.findall(r"[a-zA-Z][a-zA-Z0-9_-]+", prompt.lower())) label_words: set[str] = set() for shape in canvas.shapes.values(): label_words.update(re.findall(r"[a-zA-Z][a-zA-Z0-9_-]+", shape.text.lower())) important = {w for w in prompt_words if len(w) >= 4} if not important: return 0.5 coverage = len(important & label_words) / max(1, len(important)) return max(0.0, min(1.0, coverage)) def reward(prompt: str, completion: str) -> float: canvas, errors = validate_completion(completion) if errors or canvas is None: return 0.0 validity = 1.0 layout = score_layout(canvas) semantics = score_semantics(prompt, canvas) return 0.4 * validity + 0.3 * layout + 0.3 * semantics ``` This reward is intentionally imperfect. It will miss many things humans care about. It may overvalue labels. It may undervalue aesthetics. It may fail on synonyms. It may reward a diagram that contains the right words but has the wrong structure. That is fine for a toy environment. In fact, it is useful because it exposes the central problem. The hard part of RL for agents is usually not the policy-gradient equation. The hard part is constructing an environment where reward is correlated with the behavior you actually want. A weak reward for this task would be: 1 if JSON parses, else 0 That teaches the model to be valid, but not useful. A stronger reward might look like this: ``` reward = 0.25 * parses_as_json 0.20 * schema_valid 0.20 * renderer_accepts 0.15 * requested_entities_present 0.10 * arrows_connect_expected_entities 0.10 * layout_quality ``` For a real tldraw-like agent, some of these can be checked with code. You can validate the schema, execute actions inside the real editor, inspect final shapes, check arrow bindings, count overlaps, and export a screenshot. Other parts may require a judge model. You might ask a VLM or LLM judge whether the screenshot satisfies the user’s intent. But even then, the judge should be treated as a noisy component of the environment, not as an oracle. Log its judgments. Inspect failures. Compare against human review. Add negative tests. Assume the model will eventually exploit whatever reward signal is easiest to exploit. ## Why Teacher Trajectories Come Before RL If we start RL from a model that cannot produce valid actions, almost every rollout receives zero reward. This is not merely an optimization inconvenience. It is a state-distribution problem. Before SFT, the model’s policy places most of its probability mass outside the valid region of the environment. It writes explanations, markdown, malformed JSON, invalid IDs, impossible connections, or plausible-looking actions that the canvas rejects. The environment returns zero. The gradient has very little useful information. This is why teacher trajectories matter. A stronger model, such as Gemini, can generate examples of the action language. We can sample multiple completions, validate them, keep the ones that work, and turn them into an SFT dataset. In a single-turn environment, a trajectory is just: observation: user prompt action: JSON actions reward: validation score In a multi-turn environment, it becomes: obs_0 -> action_0 -> obs_1 -> action_1 -> obs_2 -> reward For a richer tldraw-style setup, the trajectory might contain the user request, the visible canvas state, the selected shapes, the model’s action batch, the validator output, the final screenshot, and the reward. The key point is that teacher generation is not magic. It is sampling from a stronger policy, executing the output in the environment, and keeping the traces that survive. Here is a minimal sketch using Gemini structured output. This produces the dataset we need for the first phase. The teacher moves the student into the valid action manifold. It does not solve the whole task. It gives RL somewhere useful to start. ``` # teacher_generate.py from __future__ import annotations import json from pathlib import Path from typing import Literal from google import genai from pydantic import BaseModel, Field from reward import reward from env import validate_completion class CreateShape(BaseModel): type: Literal["create_shape"] id: str shape: Literal["rectangle", "ellipse", "diamond", "text"] x: float y: float w: float h: float text: str = "" class Connect(BaseModel): type: Literal["connect"] from_: str = Field(alias="from") to: str text: str = "" class AgentResponse(BaseModel): actions: list[CreateShape | Connect] SYSTEM_PROMPT = """You are a diagramming agent. The user will ask for a small diagram. Return only JSON with an `actions` array. You can create shapes and connect them. Rules: - Create every shape before connecting it. - Use stable, meaningful shape ids. - Prefer simple left-to-right or top-to-bottom layouts. - Avoid overlaps. - Put short readable labels inside shapes. - Return only JSON. No markdown. """ PROMPTS = [ "Draw a three step data pipeline: client, API, database.", "Draw a login flow with user, auth service, token, and dashboard.", "Draw a simple RAG system with documents, embeddings, vector DB, retriever, and LLM.", "Draw a CI pipeline with commit, tests, build, deploy.", "Draw a checkout flow with cart, payment, fraud check, and receipt.", ] def generate_one(client: genai.Client, model: str, prompt: str) -> str: response = client.models.generate_content( model=model, contents=f"{SYSTEM_PROMPT}\n\nUser request: {prompt}", config={ "response_mime_type": "application/json", "response_json_schema": AgentResponse.model_json_schema(), }, ) return response.text def main() -> None: client = genai.Client() model = "gemini-2.5-flash" out_dir = Path("data/diagram_teacher") out_dir.mkdir(parents=True, exist_ok=True) accepted = out_dir / "train.jsonl" rejected = out_dir / "rejected.jsonl" with accepted.open("w") as good, rejected.open("w") as bad: for prompt in PROMPTS: for attempt in range(4): completion = generate_one(client, model, prompt) canvas, errors = validate_completion(completion) score = reward(prompt, completion) row = { "prompt": prompt, "completion": completion, "reward": score, "errors": errors, } if canvas is not None and not errors and score >= 0.6: good.write( json.dumps( { "messages": [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": prompt}, {"role": "assistant", "content": completion}, ], "reward": score, } ) + "\n" ) else: bad.write(json.dumps(row) + "\n") if __name__ == "__main__": main() ``` ## SFT Buys You Syntax Supervised fine-tuning is imitation learning. Given a prompt and a teacher action, the model is trained to increase the probability of that action. For a chat model, each row looks like this: ``` { "messages": [ {"role": "system", "content": "..."}, {"role": "user", "content": "Draw a three step data pipeline."}, {"role": "assistant", "content": "{\"actions\": [...]}"} ] } ``` The important thing to notice is what SFT is and is not doing. SFT is not asking whether the completion is better than another completion. It is not directly optimizing the reward. It is not exploring the environment. It is pulling the student toward demonstrated trajectories. For agent training, this is still extremely useful. SFT teaches the model the grammar of the action space. It teaches the model to return JSON, use the right schema, create shapes before connections, and avoid obviously invalid action sequences. In this sense, SFT buys syntax. It moves the policy from a region where almost every rollout fails into a region where some rollouts are valid enough for reinforcement learning to become meaningful. A minimal TRL SFT script looks like this: ``` # train_sft.py from datasets import load_dataset from trl import SFTConfig, SFTTrainer dataset = load_dataset("json", data_files="data/diagram_teacher/train.jsonl", split="train") args = SFTConfig( output_dir="outputs/diagram-sft", per_device_train_batch_size=2, gradient_accumulation_steps=8, learning_rate=2e-5, num_train_epochs=3, max_length=4096, logging_steps=10, save_steps=100, ) trainer = SFTTrainer( model="Qwen/Qwen3-4B-Instruct-2507", args=args, train_dataset=dataset, ) trainer.train() trainer.save_model("outputs/diagram-sft/final") ``` Run it with: ``` accelerate launch train_sft.py ``` You can add LoRA or QLoRA if you want. You can scale the dataset. You can use multiple teachers. Those are important engineering choices, but they do not change the conceptual role of this phase. After SFT, the model should produce syntactically valid action sequences much more often. That is enough to unlock the next phase. ## RL Buys You Optimization If SFT teaches the model to imitate good traces, why do RL at all? Because the teacher traces are static. They show what the teacher did on the prompts we generated. They do not directly optimize the metric we care about. SFT says: make this teacher answer more likely. RL says: make high-reward answers more likely. That difference matters. Suppose the teacher creates a valid diagram, but the layout is cramped. SFT will copy the cramped layout. RL can discover a cleaner layout if the reward values spacing. Suppose the teacher labels a node “DB,” but the reward gives higher semantic coverage for “Database.” RL can move toward “Database.” Suppose there are many valid diagrams for the same prompt. SFT collapses toward the teacher’s style. RL can reinforce whichever variants score better under the environment. This is the basic division of labor: SFT buys you syntax. RL buys you optimization. The usual pipeline is therefore: base model -> SFT on teacher traces -> RL on environment reward The SFT checkpoint is not the final model. It is the initialization that makes on-policy exploration productive. ## RL From First Principles The simplest online RL loop for LLMs is not complicated. Sample a batch of prompts. Generate multiple completions for each prompt. Score each completion with the environment. Convert those scores into advantages. Increase the probability of completions that did better than their siblings. Decrease, or at least stop increasing, the probability of completions that did worse. This is the intuition behind group-relative methods like GRPO. For a prompt x, the model samples a group of completions: y_1, y_2, ..., y_G The environment scores them: r_1, r_2, ..., r_G Then we compute a group-relative advantage: A_i = (r_i - mean(r)) / (std(r) + eps) If a completion is better than its siblings, its advantage is positive. If it is worse, its advantage is negative. The simplified loss is: loss = - advantage * logprob(completion tokens) Real implementations add clipping, KL penalties, reference models, token masks, distributed rollout generation, vLLM inference, caching, and many stability tricks. But the core idea is still this: sample from the current policy, evaluate with the environment, and move probability mass toward better samples. Here is a deliberately bare trainer sketch. It is not production code. It is meant to expose the mechanism. ``` # train_rl_minimal.py from __future__ import annotations import json import random import torch from transformers import AutoModelForCausalLM, AutoTokenizer from reward import reward SYSTEM_PROMPT = """You are a diagramming agent. Return only JSON with an actions array.""" PROMPTS = [ "Draw a three step data pipeline: client, API, database.", "Draw a login flow with user, auth service, token, and dashboard.", "Draw a simple RAG system with documents, embeddings, vector DB, retriever, and LLM.", "Draw a CI pipeline with commit, tests, build, deploy.", "Draw a checkout flow with cart, payment, fraud check, and receipt.", ] def format_prompt(tokenizer, user_prompt: str) -> str: messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_prompt}, ] return tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) def generate_group(model, tokenizer, prompt: str, group_size: int, max_new_tokens: int) -> list[str]: encoded = tokenizer(prompt, return_tensors="pt").to(model.device) outputs = model.generate( **encoded, do_sample=True, temperature=0.8, top_p=0.95, max_new_tokens=max_new_tokens, num_return_sequences=group_size, pad_token_id=tokenizer.eos_token_id, ) completions = [] prompt_len = encoded["input_ids"].shape[1] for output in outputs: completion_ids = output[prompt_len:] completions.append(tokenizer.decode(completion_ids, skip_special_tokens=True)) return completions def completion_logprob(model, tokenizer, prompt: str, completion: str) -> torch.Tensor: full_text = prompt + completion full = tokenizer(full_text, return_tensors="pt").to(model.device) prompt_ids = tokenizer(prompt, return_tensors="pt").to(model.device)["input_ids"] prompt_len = prompt_ids.shape[1] input_ids = full["input_ids"] attention_mask = full["attention_mask"] logits = model(input_ids=input_ids, attention_mask=attention_mask).logits logprobs = torch.log_softmax(logits[:, :-1, :], dim=-1) target_ids = input_ids[:, 1:] token_logprobs = logprobs.gather(-1, target_ids.unsqueeze(-1)).squeeze(-1) completion_token_logprobs = token_logprobs[:, prompt_len - 1 :] return completion_token_logprobs.mean() def main() -> None: model_name = "outputs/diagram-sft/final" tokenizer = AutoTokenizer.from_pretrained(model_name) model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch.bfloat16, device_map="cuda", ) model.train() optimizer = torch.optim.AdamW(model.parameters(), lr=1e-6) group_size = 4 max_new_tokens = 1024 for step in range(500): user_prompt = random.choice(PROMPTS) prompt = format_prompt(tokenizer, user_prompt) completions = generate_group(model, tokenizer, prompt, group_size, max_new_tokens) rewards = torch.tensor([reward(user_prompt, c) for c in completions], device=model.device) if rewards.std() < 1e-6: print(f"step={step} skipped zero-variance rewards={rewards.tolist()}") continue advantages = (rewards - rewards.mean()) / (rewards.std() + 1e-6) losses = [] for completion, advantage in zip(completions, advantages, strict=True): logp = completion_logprob(model, tokenizer, prompt, completion) losses.append(-advantage.detach() * logp) loss = torch.stack(losses).mean() optimizer.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) optimizer.step() print( json.dumps( { "step": step, "loss": float(loss.detach().cpu()), "reward_mean": float(rewards.mean().cpu()), "reward_max": float(rewards.max().cpu()), "rewards": [float(r) for r in rewards.cpu()], } ) ) if step and step % 100 == 0: model.save_pretrained(f"outputs/diagram-rl/step-{step}") tokenizer.save_pretrained(f"outputs/diagram-rl/step-{step}") if __name__ == "__main__": main() ``` This trainer has many flaws. It computes log probabilities inefficiently. It does not use a frozen reference model. It does not use PPO or GRPO clipping. It does not separate generation from training. It does not handle distributed training. It uses a very simplified token objective. But it shows the loop. The model samples multiple diagrams for the same prompt. The environment scores them. The loss nudges the model toward the better diagrams. That is the basic mechanism behind RL for LLM agents. ## Why RL Trainers Feel More Complicated Than SFT Trainers SFT data is fixed. RL data is produced by the model during training. That one difference explains a lot of the complexity. In SFT, the dataset exists before optimization starts. You can shuffle it, batch it, and train on it. The model does not affect which examples appear next. In RL, the policy determines the training data. As the model changes, the sampled completions change. As the completions change, the reward distribution changes. As the reward distribution changes, the gradient changes. Generation and optimization become coupled. This is why RL trainers need guardrails. A completion that gets a high reward might get reinforced too aggressively. A low-reward completion might be punished too strongly. The model can drift away from general language quality. It can exploit the reward. It can collapse into short, safe completions. It can overfit quirks of the verifier. Modern RL trainers usually introduce some combination of a frozen reference model, KL penalties, ratio clipping, reward normalization, length masking, and completion masking. The reference model is often the SFT checkpoint. The policy model is the one being updated. If the policy starts assigning much higher probability to a completion than the old model or reference model did, the update can be clipped or penalized. A simplified PPO-style ratio looks like this: ratio = exp(logprob_policy - logprob_old) The clipped objective prevents the policy from moving too far in one step: min(ratio * advantage, clip(ratio, 1 - eps, 1 + eps) * advantage) An educational version would store the log probability from the model that generated the completion: old_logp = completion_logprob(model, tokenizer, prompt, completion).detach() Then, during optimization: ``` new_logp = completion_logprob(model, tokenizer, prompt, completion) ratio = torch.exp(new_logp - old_logp) unclipped = ratio * advantage clipped = torch.clamp(ratio, 0.8, 1.2) * advantage loss = -torch.minimum(unclipped, clipped) ``` You can also add a reference-model KL term: ``` policy_logp = completion_logprob(policy, tokenizer, prompt, completion) ref_logp = completion_logprob(reference, tokenizer, prompt, completion).detach() kl_estimate = policy_logp - ref_logp loss = rl_loss + beta * kl_estimate ``` This is still simplified, but it explains why PPO and GRPO implementations look more complex than SFT implementations. They are not just fitting a static dataset. They are controlling the movement of a policy while the policy is also generating its own training data. ## The On-Policy View The most useful way to think about this is through state distributions. In SFT, the model trains on external trajectories. These trajectories may come from humans, a stronger teacher model, or a curated dataset. The current model did not produce them. The data distribution is fixed before training. In RL, the model trains on its own trajectories. The completions come from the current policy. This creates a very different geometry. The model is not being pulled toward an arbitrary external behavior distribution. It is sampling from its current behavior distribution, receiving reward, and locally reshaping probability mass toward the parts of that distribution that worked better. This is one reason RL often feels less like imitation and more like capability refinement. SFT can teach a model to speak the action language, but it may also pull the model toward the teacher’s quirks. RL starts from the model’s own attempts and asks which of those attempts succeed in the environment. For agent training, this matters a lot. A model after SFT may produce valid JSON but still make poor diagrams. RL does not need to know the one correct answer. It only needs enough variation in the model’s own samples and a reward function that can distinguish better attempts from worse attempts. This is also why RL can sometimes improve beyond the teacher traces. The teacher dataset may contain one valid diagram per prompt. The student’s rollouts may contain many variants. If the environment can score those variants, optimization can move toward completions the teacher never demonstrated. ## Mapping This to TRL Once the loop is clear, TRL becomes much easier to understand. GRPOTrainer accepts a model, a dataset of prompts, and one or more reward functions. It handles generation, reward computation, advantage estimation, and optimization. The diagram task becomes: ``` # train_grpo_trl.py from datasets import Dataset from trl import GRPOConfig, GRPOTrainer from reward import reward prompts = [ "Draw a three step data pipeline: client, API, database.", "Draw a login flow with user, auth service, token, and dashboard.", "Draw a simple RAG system with documents, embeddings, vector DB, retriever, and LLM.", "Draw a CI pipeline with commit, tests, build, deploy.", ] dataset = Dataset.from_list([{"prompt": prompt} for prompt in prompts]) def diagram_reward_func(prompts, completions, **kwargs): scores = [] for prompt, completion in zip(prompts, completions, strict=True): text = completion if isinstance(completion, str) else completion[0]["content"] scores.append(reward(prompt, text)) return scores args = GRPOConfig( output_dir="outputs/diagram-grpo", learning_rate=1e-6, per_device_train_batch_size=2, gradient_accumulation_steps=4, num_generations=4, max_prompt_length=1024, max_completion_length=1024, logging_steps=1, ) trainer = GRPOTrainer( model="outputs/diagram-sft/final", reward_funcs=diagram_reward_func, args=args, train_dataset=dataset, ) trainer.train() ``` This looks much shorter than the minimal trainer because the infrastructure is hidden. But conceptually it is the same loop: prompt batch -> generate completions -> score completions -> compute advantages -> update policy This is the right way to use frameworks. Not as magic boxes, but as implementations of mechanisms you already understand. ## What Makes an Environment Good? The quality of an RL agent is often bottlenecked by the quality of the environment. This is especially true for tool-using agents. The model can only optimize what the environment can measure. If your validator only checks JSON validity, your agent will learn to emit valid JSON. If your reward checks schema validity and task completion, it can learn task completion. If your reward includes a brittle LLM judge, the model may learn the judge’s preferences instead of the user’s. A good environment has a few properties. First, it gives partial credit. If every failure receives zero, learning becomes difficult. A malformed JSON output should score worse than a valid-but-incomplete diagram, and a valid-but-incomplete diagram should score worse than a complete, readable one. Second, it separates syntax from semantics. Validity is necessary, but it is not the task. The reward should distinguish “the action sequence executed” from “the result satisfied the request.” Third, it should be hard to exploit. If the model can maximize reward by creating huge text boxes containing every prompt word, it will eventually learn that. If it can get high reward with disconnected shapes, it will do that. If it can satisfy a judge with superficial labels, it will do that. Fourth, it should produce useful logs. Every rollout should be inspectable: prompt, completion, parse errors, canvas state, reward components, final screenshot, and judge rationale if a judge is used. Without this, reward hacking becomes invisible. This is why I increasingly think of agent training as environment design first and algorithm selection second. The trainer decides how to use the signal. The environment decides what signal exists. ## Common Failure Modes Many RL failures look mysterious until you view them as environment or state-distribution failures. If the model never gets reward, the policy is probably not visiting valid states. This usually means the action schema is too strict, the base model cannot speak the action language, or SFT was skipped. The fix is not to stare at the RL algorithm. The fix is to make the valid region reachable: simplify the action space, generate teacher traces, add partial rewards, and train the model to produce valid actions. If the model learns to emit tiny trivial diagrams, the reward probably overvalues validity and undervalues task satisfaction. The environment is saying “valid is enough,” and the model is listening. Add entity coverage, minimum shape counts, structure checks, and prompt-conditioned semantic scoring. If the model emits valid JSON but ignores the prompt, the reward is mostly syntactic. The action language has been learned, but the task has not. Add checks that compare the prompt to the final canvas. If the model overfits teacher style, the SFT data is too narrow or the RL phase is too weak. Add teacher diversity, sample multiple attempts per prompt, use multiple teacher models, or let the environment reward alternative valid solutions. If the RL run is unstable, the update is probably too aggressive relative to reward quality and batch size. Lower the learning rate, shorten completions, add clipping, add a reference-model KL penalty, increase batch size, and inspect reward variance. If reward goes up but human quality does not, the reward is being gamed. This is the most important failure mode. Inspect examples manually. Decompose the reward. Add negative tests. Penalize the exploit directly. Avoid relying on a single scalar from one brittle heuristic. ## What Changes for a Real tldraw Agent? The toy Python canvas is educational. A real tldraw environment changes the validator, not the loop. Instead of applying actions to a fake Canvas, you would execute them inside the real editor. Start the tldraw app or a validator page. Send JSON actions into the editor. Let tldraw sanitize and apply them. Read back the final shapes, bindings, selections, and errors. Export a screenshot. Compute reward from the shape state, the screenshot, and any validator errors. The policy still receives an observation and emits an action. The environment still executes the action. The reward still scores the result. The trainer still updates the model. This is why environment design is such a powerful abstraction. Once you can expose: reward(prompt, completion) -> float or, for richer settings: step(state, action) -> next_state, reward, done, info you can plug the environment into a tiny educational trainer, TRL, Unsloth, PRIME-RL, verl, or a custom distributed training stack. The framework changes. The shape of the problem does not. ## The Full Pipeline The complete pipeline looks like this. First, define the environment. This means the action schema, parser, validator, simulator or renderer, and reward function. This is the most important part because it defines what the model is actually optimizing. Then generate prompts. Start with hand-written seed prompts, then expand with synthetic prompts. Make the prompts varied enough that the model cannot survive by memorizing a narrow pattern. Then generate teacher trajectories. Use a stronger model to sample candidate actions. Validate them through the environment. Keep the traces that pass. Save rejected traces too, because they are useful for debugging the schema and reward. Then SFT the student. The purpose of SFT is to move the model into the valid action space. Measure valid-action rate and average reward before doing RL. Then run RL. Sample prompts, generate multiple completions per prompt, score them with the environment, compute group-relative advantages, and update the policy. Then evaluate. Use held-out prompts. Track parse rate, schema-valid rate, environment reward, semantic score, reward distribution, and human review of rendered diagrams. For visual tasks, inspect screenshots. For tool tasks, inspect final states. Then iterate. Improve the reward. Add harder prompts. Refresh teacher traces. Tune RL hyperparameters. Add negative tests. Repeat. A useful compressed mental model is: Teacher trajectories solve cold start. SFT teaches the action language. RL optimizes behavior inside the environment. Evaluation tells you whether the environment measures the right thing. ## Closing The magic of agent training is not in the framework. The framework matters, especially at scale. Distributed rollout generation, vLLM serving, batching, LoRA training, checkpointing, reference models, logging, and reproducibility are all real engineering problems. But they are not the conceptual center. The conceptual center is smaller. Define the action space. Define success. Generate traces. Imitate good traces. Sample new attempts. Reward good attempts. Update the model. That is the shape of the system. SFT and RL play different roles in this shape. SFT moves the model into the valid region of the action space. RL uses the environment to reshape the policy toward higher-reward behavior. Teacher trajectories solve the cold-start problem. Verifiers turn execution into supervision. Once you see this, a tldraw agent is not a special case. A browser agent, coding agent, spreadsheet agent, robotics planner, and math solver all have the same skeleton: state -> action -> validation -> reward -> learning The rest is engineering. ## References - Gemini text generation docs: https://ai.google.dev/gemini-api/docs/text-generation - Gemini structured output docs: https://ai.google.dev/gemini-api/docs/structured-output - TRL documentation: https://huggingface.co/docs/trl/index - TRL GRPOTrainer docs: https://huggingface.co/docs/trl/grpo_trainer - Unsloth RL guide: https://docs.unsloth.ai/get-started/reinforcement-learning-rl-guide ## 相关链接 - [anshuman](https://x.com/athleticKoder) - [@athleticKoder](https://x.com/athleticKoder) - [34K](https://x.com/athleticKoder/status/2057091692235481560/analytics) - [https://ai.google.dev/gemini-api/docs/text-generation](https://ai.google.dev/gemini-api/docs/text-generation) - [https://ai.google.dev/gemini-api/docs/structured-output](https://ai.google.dev/gemini-api/docs/structured-output) - [https://huggingface.co/docs/trl/index](https://huggingface.co/docs/trl/index) - [https://huggingface.co/docs/trl/grpo_trainer](https://huggingface.co/docs/trl/grpo_trainer) - [https://docs.unsloth.ai/get-started/reinforcement-learning-rl-guide](https://docs.unsloth.ai/get-started/reinforcement-learning-rl-guide) - [Upgrade to Premium](https://x.com/i/premium_sign_up) - [9:30 PM · May 20, 2026](https://x.com/athleticKoder/status/2057091692235481560) - [34.3K Views](https://x.com/athleticKoder/status/2057091692235481560/analytics) - [View quotes](https://x.com/athleticKoder/status/2057091692235481560/quotes) --- *导出时间: 2026/5/21 09:54:27* --- ## 中文翻译 # 从第一性原理构建智能体 **作者**: anshuman **日期**: 2026-05-20T13:30:53.000Z **来源**: [https://x.com/athleticKoder/status/2057091692235481560](https://x.com/athleticKoder/status/2057091692235481560) --- ![Image](/doc/2106/image/img_6221ce934759.jpg) 如何定义环境、生成教师轨迹、微调学生模型，并通过强化学习加以改进。作者：Anshuman Mishra & GPT 5.5，2026年5月20日 [编者按：观点属于我个人；文字和结构借助 GPT 5.5 进行了润色。这在某种程度上是一个实验——利用 AI 帮助我从粗略的笔记快速写成技术研究博客，同时保持真正品味、方向和论断由人来把控。] 关于后训练的教程起点都太高了。它们从框架开始。安装这个库，定义这个奖励函数，运行这个训练器，看着奖励曲线上升。这在你已经理解发生了什么时是有用的。但当你试图为整个系统建立心智模型时，就没那么有用了。我发现从更低处开始更有帮助。在有训练器之前，先有环境。在强化学习之前，先有动作空间。在智能体之前，先有一个策略在产生能改变世界某种状态的动作。这篇文章就是一次尝试，从第一性原理构建这幅图景。例子刻意做得很小：一个文本转图表的智能体。用户请求一个简单的图表，模型输出结构化的 JSON 动作，在画布上创建图形。你可以把它看作一个迷你版的 tldraw 风格智能体。模型不是在真实编辑器里点击操作，而是发出诸如“创建一个矩形”、“添加一个标签”、“连接这两个节点”之类的动作。目标不是构建世界上最好的图表智能体。目标是理解智能体训练本身的结构。从高层次看，这个循环是：提示 -> 模型动作 -> 环境 -> 奖励 -> 梯度更新几乎每个智能体训练系统都是这个循环的放大版。浏览器智能体、编码智能体、电子表格智能体、机器人规划器、数学求解器和图表绘制智能体都具有相同的基本结构。它们的区别在于环境、动作空间和奖励函数。这是使用框架时容易忽略的部分。像 TRL、Unsloth、PRIME-RL、verl、OpenRLHF 或自定义内部训练器这样的库并非魔法。它们大多是围绕这个循环构建的基础设施：批处理、rollout 生成、分布式推理、奖励计算、日志记录、参考模型、裁剪、检查点保存和扩展。概念核心要小得多。 ## 智能体是环境中的一个策略语言模型是序列上的一个分布。当我们把它用作智能体时，我们是在要求这个分布产生动作，而不是普通的文字。给定一个观测，模型发出一个动作。在聊天模型中，观测是对话内容，动作是助手消息。在浏览器智能体中，观测可能是 DOM，动作可能是一次点击或一次按键。在编码智能体中，观测可能是仓库状态，动作可能是一个补丁。在图表智能体中，观测是用户请求，动作是一个描述图形和连接的 JSON 对象。所以第一个问题不是“我该用哪个 RL 训练器？” 第一个问题是：环境是什么？环境定义了哪些动作是有效的、这些动作被执行时会发生什么，以及如何衡量成功。在普通的监督微调中，这个环境往往是隐式的。我们向模型展示良好行为的示例，并要求它模仿。在强化学习中，环境变得显式。模型尝试某些操作，环境做出响应，奖励函数判断这次尝试是否良好。对于图表智能体而言，一个补全不是因为它听起来合理就是好的。它必须满足：JSON 可以解析、模式合法、画布接受这些动作、请求的对象出现了、箭头连接了正确的节点、最终布局可以理解。这就是普通聊天微调和智能体训练之间的核心区别。智能体训练将模型的输出锚定在一个可执行的世界中。 ## 动作语言让我们从最小的动作空间开始。模型必须返回带有 actions 数组的 JSON。每个动作要么创建一个图形，要么连接两个图形。一个有效的补全可能如下所示： ``` { "actions": [ { "type": "create_shape", "id": "frontend", "shape": "rectangle", "x": 80, "y": 100, "w": 180, "h": 80, "text": "Frontend" }, { "type": "create_shape", "id": "api", "shape": "rectangle", "x": 340, "y": 100, "w": 180, "h": 80, "text": "API" }, { "type": "connect", "from": "frontend", "to": "api", "text": "request" } ] } ``` 这看起来很简单，但它已经包含了工具使用的本质结构。模型不再只是生成文本。它在生成将被另一个系统执行的指令。这改变了训练问题。模型不仅要学会说什么，还要学会环境会接受什么。它必须先创建图形再连接它们。它必须使用稳定的 ID。它必须避免重复的 ID。它必须保持坐标有限。它必须避免无效的图形类型。它必须发出可解析的 JSON。这些不是哲学上的细节。它们决定了一个策略能否进入动作空间的有效区域。这就是为什么在 RL 之前通常需要 SFT。在模型能够优化奖励之前，它必须学会环境的语言。 ## 一个不用框架的微型环境一个环境只需要三样东西：一个输入提示、一个动作格式和一个奖励函数。下面是一个极简的纯 Python 画布环境。它支持矩形、椭圆、菱形、文本块和箭头。重点不在于这个画布有多精巧，而在于它为我们提供了一个确定性世界，模型输出在其中可以成功或失败。 ``` # env.py from __future__ import annotations import json import math from dataclasses import dataclass, field from typing import Any ALLOWED_SHAPES = {"rectangle", "ellipse", "diamond", "text"} @dataclass class Shape: id: str shape: str x: float y: float w: float h: float text: str = "" @dataclass class Arrow: source: str target: str text: str = "" @dataclass class Canvas: shapes: dict[str, Shape] = field(default_factory=dict) arrows: list[Arrow] = field(default_factory=list) def create_shape(self, action: dict[str, Any]) -> None: shape_id = require_str(action, "id") shape_type = require_str(action, "shape") if shape_type not in ALLOWED_SHAPES: raise ValueError(f"unknown shape type: {shape_type}") if shape_id in self.shapes: raise ValueError(f"duplicate shape id: {shape_id}") x = require_number(action, "x") y = require_number(action, "y") w = require_number(action, "w") h = require_number(action, "h") if w <= 0 or h <= 0: raise ValueError("shape width and height must be positive") if w > 1000 or h > 1000: raise ValueError("shape too large") self.shapes[shape_id] = Shape( id=shape_id, shape=shape_type, x=x, y=y, w=w, h=h, text=str(action.get("text", "")), ) def connect(self, action: dict[str, Any]) -> None: source = require_str(action, "from") target = require_str(action, "to") if source not in self.shapes: raise ValueError(f"arrow source does not exist: {source}") if target not in self.shapes: raise ValueError(f"arrow target does not exist: {target}") if source == target: raise ValueError("arrow cannot connect a shape to itself") self.arrows.append(Arrow(source=source, target=target, text=str(action.get("text", "")))) def apply(self, action: dict[str, Any]) -> None: action_type = require_str(action, "type") if action_type == "create_shape": self.create_shape(action) elif action_type == "connect": self.connect(action) else: raise ValueError(f"unknown action type: {action_type}") def require_str(obj: dict[str, Any], key: str) -> str: value = obj.get(key) if not isinstance(value, str) or not value: raise ValueError(f"{key} must be a non-empty string") return value def require_number(obj: dict[str, Any], key: str) -> float: value = obj.get(key) if not isinstance(value, int | float) or not math.isfinite(value): raise ValueError(f"{key} must be a finite number") return float(value) def parse_actions(text: str) -> list[dict[str, Any]]: try: data = json.loads(text) except json.JSONDecodeError: start = text.find("{") end = text.rfind("}") if start == -1 or end == -1 or end <= start: raise ValueError("model output does not contain JSON") data = json.loads(text[start : end + 1]) actions = data.get("actions") if not isinstance(actions, list): raise ValueError("missing actions array") if not actions: raise ValueError("actions array is empty") if len(actions) > 40: raise ValueError("too many actions") if not all(isinstance(action, dict) for action in actions): raise ValueError("each action must be an object") return actions def validate_completion(text: str) -> tuple[Canvas | None, list[str]]: errors: list[str] = [] canvas = Canvas() try: actions = parse_actions(text) except Exception as exc: return None, [str(exc)] for i, action in enumerate(actions): try: canvas.apply(action) except Exception as exc: errors.append(f"action {i}: {exc}") return canvas, errors ``` 这已经为我们提供了验证器环境的核心。模型生成文本，环境解析该文本、执行动作序列，并在出现问题时返回错误。一个框架可以让这更具可扩展性，但不会让它产生概念上的根本差异。 ## 奖励才是任务的核心所在一旦环境能够执行动作，我们就需要定义什么是成功。这正是大部分真正困难所在。训练器只能优化你给它的奖励。如果奖励主要衡量语法，模型就会学习语法。如果奖励衡量任务完成度，模型就有机会学到有用的行为。如果奖励很脆弱，模型最终会发现并利用这种脆弱性。对于这个玩具图示智能体，我们可以组合几个信号。我们可以奖励能被解析的输出、通过验证的动作、不重叠的布局、存在的标签、连接对象的箭头，以及覆盖了用户请求中重要词语的标签。 ``` # reward.py from __future__ import annotations import re from env import Canvas, validate_completion def score_layout(canvas: Canvas) -> float: if not canvas.shapes: return 0.0 score = 1.0 shapes = list(canvas.shapes.values()) for i, a in enumerate(shapes): for b in shapes[i + 1 :]: ax2, ay2 = a.x + a.w, a.y + a.h bx2, by2 = b.x + b.w, b.y + b.h overlap = not (ax2 < b.x or bx2 < a.x or ay2 < b.y or by2 < a.y) if overlap: score -= 0.15 labeled = sum(1 for shape in shapes if shape.text.strip()) score += 0.1 * min(labeled, 5) score += 0.1 * min(len(canvas.arrows), 5) return max(0.0, min(1.0, score)) def score_semantics(prompt: str, canvas: Canvas) -> float: prompt_words = set(re.findall(r"[a-zA-Z][a-zA-Z0-9_-]+", prompt.lower())) label_words: set[str] = set() for shape in canvas.shapes.values(): label_words.update(re.findall(r"[a-zA-Z][a-zA-Z0-9_-]+", shape.text.lower())) important = {w for w in prompt_words if len(w) >= 4} if not important: return 0.5 coverage = len(important & label_words) / max(1, len(important)) return max(0.0, min(1.0, coverage)) def reward(prompt: str, completion: str) -> float: canvas, errors = validate_completion(completion) if errors or canvas is None: return 0.0 validity = 1.0 layout = score_layout(canvas) semantics = score_semantics(prompt, canvas) return 0.4 * validity + 0.3 * layout + 0.3 * semantics ``` 这个奖励函数是有意设计得不完美的。它会遗漏许多人类在意的方面。它可能高估标签的价值。它可能低估美观的价值。它可能无法处理同义词。它可能奖励一个包含正确词语但结构错误的图表。对于一个玩具环境来说，这已经足够了。事实上，这很有用，因为它暴露了核心问题。智能体强化学习（RL）的难点通常不在于策略梯度方程，而在于构建一个奖励与您真正想要的行为相关联的环境。一个较弱的奖励可能是这样的：如果 JSON 能解析则为 1，否则为 0 这只会教会模型生成有效输出，而不是有用输出。一个更强的奖励可能看起来像这样： ``` reward = 0.25 * parses_as_json 0.20 * schema_valid 0.20 * renderer_accepts 0.15 * requested_entities_present 0.10 * arrows_connect_expected_entities 0.10 * layout_quality ``` 对于一个真实的类 tldraw 智能体，其中一些指标可以用代码来检验。你可以验证 schema、在真实编辑器中执行操作、检查最终的形状、检查箭头绑定、统计重叠数量，并导出截图。其他部分可能需要一个评判模型。你可以让 VLM 或 LLM 评判者判断截图是否满足用户的意图。但即便如此，评判者也应该被视为环境中一个有噪声的组件，而不是神谕。记录它的判断结果。检查失败案例。与人工评审进行对比。添加负面测试。要假设模型最终会利用任何最容易利用的奖励信号。 ## 为什么教师轨迹要先于强化学习如果我们从一个无法产生有效动作的模型开始强化学习，几乎每一次 rollout 的奖励都会是零。这不仅仅是一个优化上的麻烦。这是一个状态分布问题。在 SFT 之前，模型的策略把大部分概率质量放在了环境有效区域之外。它会写出解释性文字、markdown、格式错误的 JSON、无效的 ID、不可能的连接，或者画布会拒绝的看似合理的动作。环境返回零奖励。梯度中几乎没有有用的信息。这就是为什么教师轨迹如此重要。一个更强的模型（例如 Gemini）可以生成动作语言的示例。我们可以采样多个补全、验证它们、保留有效的那些，并把它们转化为 SFT 数据集。在单轮环境中，一条轨迹就是： observation: 用户提示词 action: JSON 动作 reward: 验证得分在多轮环境中，它变成： obs_0 -> action_0 -> obs_1 -> action_1 -> obs_2 -> reward 对于一个更丰富的 tldraw 风格设置，轨迹可能包含用户请求、可见的画布状态、选中的形状、模型的动作批次、验证器输出、最终截图以及奖励。关键在于，教师生成并不是什么魔法。它就是从更强的策略中采样，在环境中执行输出，并保留那些存活下来的轨迹。下面是一个使用 Gemini 结构化输出的最小示例。这将生成第一阶段所需的数据集。教师模型把学生模型带入了有效的动作流形。它并没有解决整个任务，但它为强化学习提供了一个有用的起点。 ``` # teacher_generate.py from __future__ import annotations import json from pathlib import Path from typing import Literal from google import genai from pydantic import BaseModel, Field from reward import reward from env import validate_completion class CreateShape(BaseModel): type: Literal["create_shape"] id: str shape: Literal["rectangle", "ellipse", "diamond", "text"] x: float y: float w: float h: float text: str = "" class Connect(BaseModel): type: Literal["connect"] from_: str = Field(alias="from") to: str text: str = "" class AgentResponse(BaseModel): actions: list[CreateShape | Connect] SYSTEM_PROMPT = """You are a diagramming agent. The user will ask for a small diagram. Return only JSON with an `actions` array. You can create shapes and connect them. Rules: - Create every shape before connecting it. - Use stable, meaningful shape ids. - Prefer simple left-to-right or top-to-bottom layouts. - Avoid overlaps. - Put short readable labels inside shapes. - Return only JSON. No markdown. """ PROMPTS = [ "Draw a three step data pipeline: client, API, database.", "Draw a login flow with user, auth service, token, and dashboard.", "Draw a simple RAG system with documents, embeddings, vector DB, retriever, and LLM.", "Draw a CI pipeline with commit, tests, build, deploy.", "Draw a checkout flow with cart, payment, fraud check, and receipt.", ] def generate_one(client: genai.Client, model: str, prompt: str) -> str: response = client.models.generate_content( model=model, contents=f"{SYSTEM_PROMPT}\n\nUser request: {prompt}", config={ "response_mime_type": "application/json", "response_json_schema": AgentResponse.model_json_schema(), }, ) return response.text def main() -> None: client = genai.Client() model = "gemini-2.5-flash" out_dir = Path("data/diagram_teacher") out_dir.mkdir(parents=True, exist_ok=True) accepted = out_dir / "train.jsonl" rejected = out_dir / "rejected.jsonl" with accepted.open("w") as good, rejected.open("w") as bad: for prompt in PROMPTS: for attempt in range(4): completion = generate_one(client, model, prompt) canvas, errors = validate_completion(completion) score = reward(prompt, completion) row = { "prompt": prompt, "completion": completion, "reward": score, "errors": errors, } if canvas is not None and not errors and score >= 0.6: good.write( json.dumps( { "messages": [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": prompt}, {"role": "assistant", "content": completion}, ], "reward": score, } ) + "\n" ) else: bad.write(json.dumps(row) + "\n") if __name__ == "__main__": main() ``` ## SFT 为你买来语法监督微调是模仿学习。给定一个提示词和一个教师动作，模型被训练来提高该动作的概率。对于聊天模型，每一行数据看起来像这样： ``` { "messages": [ {"role": "system", "content": "..."}, {"role": "user", "content": "Draw a three step data pipeline."}, {"role": "assistant", "content": "{\"actions\": [...]}"} ] } ``` 需要注意的重点是 SFT 在做什么、不在做什么。 SFT 并不是在判断一个补全是否比另一个补全更好。它不直接优化奖励。它不探索环境。它是把学生模型拉向被演示的轨迹。对于智能体训练来说，这仍然非常有用。SFT 教会模型动作空间的语法。它教会模型返回 JSON、使用正确的模式、在建立连接之前先创建形状，以及避免明显无效的动作序列。从这个意义上说，SFT 买来的是语法。它把策略从几乎每次 rollout 都失败的区域，移动到某些 rollout 足够有效、以至于强化学习能够变得有意义的区域。一个最小化的 TRL SFT 脚本如下： ``` # train_sft.py from datasets import load_dataset from trl import SFTConfig, SFTTrainer dataset = load_dataset("json", data_files="data/diagram_teacher/train.jsonl", split="train") args = SFTConfig( output_dir="outputs/diagram-sft", per_device_train_batch_size=2, gradient_accumulation_steps=8, learning_rate=2e-5, num_train_epochs=3, max_length=4096, logging_steps=10, save_steps=100, ) trainer = SFTTrainer( model="Qwen/Qwen3-4B-Instruct-2507", args=args, train_dataset=dataset, ) trainer.train() trainer.save_model("outputs/diagram-sft/final") ``` 运行方式： ``` accelerate launch train_sft.py ``` 你可以根据需要添加 LoRA 或 QLoRA。你可以扩充数据集。你可以使用多个教师。这些都是重要的工程选择，但它们不会改变这个阶段的概念性作用。 SFT 之后，模型应该更频繁地生成语法上有效的动作序列。这就足以解锁下一个阶段了。 ## RL 为你买来优化如果 SFT 教会模型模仿好的轨迹，为什么还要做 RL？因为教师轨迹是静态的。它们展示的是教师在我们生成的提示词上的行为。它们并不能直接优化我们关心的指标。 SFT 说：让这个教师的回答更可能出现。 RL 说：让高奖励的回答更可能出现。这个区别很重要。假设教师创建了一个有效的图表，但布局很拥挤。SFT 会照搬这个拥挤的布局。而如果奖励函数重视间距，RL 可以发现更整洁的布局。假设教师把一个节点标注为"DB"，但奖励函数对"Database"给出更高的语义覆盖分。RL 可以向"Database"移动。假设同一个提示词有很多有效的图表。SFT 会向教师的风格坍缩。而 RL 可以强化任何在环境下得分更高的变体。这就是基本的分工： SFT 为你买来语法。 RL 为你买来优化。因此通常的流程是：基础模型 -> 在教师轨迹上做 SFT -> 在环境奖励上做 RL SFT 检查点并不是最终模型。它是让同策略探索变得富有成效的初始化。 ## 从第一性原理理解 RL 对于大语言模型来说，最简单的在线 RL 循环并不复杂。采样一批提示词。为每个提示词生成多个补全。用环境给每个补全打分。将这些分数转换为优势。提高那些比同组表现更好的补全的概率。降低——或至少停止提高——那些表现更差的补全的概率。这就是 GRPO 这类组相对方法背后的直觉。对于一个提示词 x，模型采样一组补全： y_1, y_2, ..., y_G 环境给它们打分： r_1, r_2, ..., r_G 然后我们计算组相对优势： A_i = (r_i - mean(r)) / (std(r) + eps) 如果一个补全比它的同组更好，它的优势为正。如果更差，优势为负。简化的损失函数是： loss = - advantage * logprob(completion tokens) 实际实现还会加入裁剪、KL 惩罚、参考模型、token 掩码、分布式 rollout 生成、vLLM 推理、缓存以及许多稳定性技巧。但核心思想仍然是：从当前策略采样，用环境评估，并把概率质量向更好的样本移动。这是一个刻意简化的训练器草图。它不是生产级代码，只是用来展示其中的机制。 ``` # train_rl_minimal.py from __future__ import annotations import json import random import torch from transformers import AutoModelForCausalLM, AutoTokenizer from reward import reward SYSTEM_PROMPT = """You are a diagramming agent. Return only JSON with an actions array.""" PROMPTS = [ "Draw a three step data pipeline: client, API, database.", "Draw a login flow with user, auth service, token, and dashboard.", "Draw a simple RAG system with documents, embeddings, vector DB, retriever, and LLM.", "Draw a CI pipeline with commit, tests, build, deploy.", "Draw a checkout flow with cart, payment, fraud check, and receipt.", ] def format_prompt(tokenizer, user_prompt: str) -> str: messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_prompt}, ] return tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) def generate_group(model, tokenizer, prompt: str, group_size: int, max_new_tokens: int) -> list[str]: encoded = tokenizer(prompt, return_tensors="pt").to(model.device) outputs = model.generate( **encoded, do_sample=True, temperature=0.8, top_p=0.95, max_new_tokens=max_new_tokens, num_return_sequences=group_size, pad_token_id=tokenizer.eos_token_id, ) completions = [] prompt_len = encoded["input_ids"].shape[1] for output in outputs: completion_ids = output[prompt_len:] completions.append(tokenizer.decode(completion_ids, skip_special_tokens=True)) return completions def completion_logprob(model, tokenizer, prompt: str, completion: str) -> torch.Tensor: full_text = prompt + completion full = tokenizer(full_text, return_tensors="pt").to(model.device) prompt_ids = tokenizer(prompt, return_tensors="pt").to(model.device)["input_ids"] prompt_len = prompt_ids.shape[1] input_ids = full["input_ids"] attention_mask = full["attention_mask"] logits = model(input_ids=input_ids, attention_mask=attention_mask).logits logprobs = torch.log_softmax(logits[:, :-1, :], dim=-1) target_ids = input_ids[:, 1:] token_logprobs = logprobs.gather(-1, target_ids.unsqueeze(-1)).squeeze(-1) completion_token_logprobs = token_logprobs[:, prompt_len - 1 :] return completion_token_logprobs.mean() def main() -> None: model_name = "outputs/diagram-sft/final" tokenizer = AutoTokenizer.from_pretrained(model_name) model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch.bfloat16, device_map="cuda", ) model.train() optimizer = torch.optim.AdamW(model.parameters(), lr=1e-6) group_size = 4 max_new_tokens = 1024 for step in range(500): user_prompt = random.choice(PROMPTS) prompt = format_prompt(tokenizer, user_prompt) completions = generate_group(model, tokenizer, prompt, group_size, max_new_tokens) rewards = torch.tensor([reward(user_prompt, c) for c in completions], device=model.device) if rewards.std() < 1e-6: print(f"step={step} skipped zero-variance rewards={rewards.tolist()}") continue advantages = (rewards - rewards.mean()) / (rewards.std() + 1e-6) losses = [] for completion, advantage in zip(completions, advantages, strict=True): logp = completion_logprob(model, tokenizer, prompt, completion) losses.append(-advantage.detach() * logp) loss = torch.stack(losses).mean() optimizer.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) optimizer.step() print( json.dumps( { "step": step, "loss": float(loss.detach().cpu()), "reward_mean": float(rewards.mean().cpu()), "reward_max": float(rewards.max().cpu()), "rewards": [float(r) for r in rewards.cpu()], } ) ) if step and step % 100 == 0: model.save_pretrained(f"outputs/diagram-rl/step-{step}") tokenizer.save_pretrained(f"outputs/diagram-rl/step-{step}") if __name__ == "__main__": main() ``` 这个训练器有很多缺陷。它计算对数概率的方式效率很低。它没有使用冻结的参考模型。它没有使用 PPO 或 GRPO 的裁剪。它没有将生成与训练分离。它没有处理分布式训练。它使用的是一个非常简化的 token 目标函数。但它展示了那个循环。模型针对同一个提示词采样生成多张图表。环境为它们打分。损失函数引导模型向更好的图表靠拢。这就是 LLM 智能体强化学习背后的基本机制。 ## 为什么 RL 训练器比 SFT 训练器感觉更复杂 SFT 的数据是固定的。RL 的数据是模型在训练过程中产生的。仅这一个差异就解释了大部分复杂性。在 SFT 中，数据集在优化开始之前就已存在。你可以对它进行打乱、分批和训练。模型不会影响接下来出现哪些样本。在 RL 中，策略本身决定了训练数据。随着模型变化，采样出的补全结果也随之变化。随着补全结果变化，奖励分布也随之变化。随着奖励分布变化，梯度也随之变化。生成和优化由此耦合在一起。这就是 RL 训练器需要防护机制的原因。获得高奖励的补全可能被过度强化。低奖励的补全可能被过度惩罚。模型可能偏离通用的语言质量。它可能钻奖励的空子。它可能退化为简短而保守的补全。它可能过拟合验证器的各种怪癖。现代 RL 训练器通常会引入以下机制的某种组合：冻结的参考模型、KL 惩罚、比率裁剪、奖励归一化、长度掩码和补全掩码。参考模型通常是 SFT 检查点。策略模型则是被更新的那个模型。如果策略开始给某个补全分配比旧模型或参考模型高得多的概率，该更新就会被裁剪或惩罚。简化的 PPO 风格比率如下： ratio = exp(logprob_policy - logprob_old) 裁剪后的目标函数防止策略一步走得太远： min(ratio * advantage, clip(ratio, 1 - eps, 1 + eps) * advantage) 教学版本会存储生成该补全的模型的对数概率： old_logp = completion_logprob(model, tokenizer, prompt, completion).detach() 然后，在优化过程中： ``` new_logp = completion_logprob(model, tokenizer, prompt, completion) ratio = torch.exp(new_logp - old_logp) unclipped = ratio * advantage clipped = torch.clamp(ratio, 0.8, 1.2) * advantage loss = -torch.minimum(unclipped, clipped) ``` 你还可以加入参考模型的 KL 项： ``` policy_logp = completion_logprob(policy, tokenizer, prompt, completion) ref_logp = completion_logprob(reference, tokenizer, prompt, completion).detach() kl_estimate = policy_logp - ref_logp loss = rl_loss + beta * kl_estimate ``` 这仍然是简化版，但它解释了为什么 PPO 和 GRPO 的实现看起来比 SFT 实现更复杂。它们不仅仅是在拟合一个静态数据集。它们在控制策略的移动，同时策略也在生成自己的训练数据。 ## 在策略视角理解这一点最有用的方式是通过状态分布。在 SFT 中，模型在外部轨迹上训练。这些轨迹可能来自人类、更强的教师模型或精心整理的数据集。当前模型并没有产生它们。数据分布在训练前就已固定。在 RL 中，模型在自身的轨迹上训练。补全来自当前策略。这创造了一种截然不同的几何结构。模型不是被拉向某个任意的的部行为分布。它从当前的行为分布中采样，获得奖励，并局部地将概率质量重塑到该分布中表现更好的部分。这是 RL 往往不像模仿、更像能力精炼的原因之一。SFT 可以教会模型说动作语言，但它也可能把模型拉向教师的怪癖。RL 从模型自身的尝试出发，追问这些尝试中哪些在环境中成功了。对于智能体训练来说，这一点非常重要。 SFT 之后的模型可能生成合法的 JSON，但仍然画出糟糕的图表。RL 不需要知道唯一的正确答案。它只需要模型自身样本有足够的多样性，以及一个能够区分更好尝试与更差尝试的奖励函数。这也是 RL 有时能超越教师轨迹的原因。教师数据集中每个提示可能只包含一张合法图表。学生的 rollout 中可能包含许多变体。如果环境能够为这些变体打分，优化过程就能朝向教师从未演示过的补全移动。 ## 将此映射到 TRL 一旦理解了这个循环，TRL 就变得容易理解得多。 GRPOTrainer 接受一个模型、一个提示数据集以及一个或多个奖励函数。它负责生成、奖励计算、优势估计和优化。图表任务变成： ``` # train_grpo_trl.py from datasets import Dataset from trl import GRPOConfig, GRPOTrainer from reward import reward prompts = [ "Draw a three step data pipeline: client, API, database.", "Draw a login flow with user, auth service, token, and dashboard.", "Draw a simple RAG system with documents, embeddings, vector DB, retriever, and LLM.", "Draw a CI pipeline with commit, tests, build, deploy.", ] dataset = Dataset.from_list([{"prompt": prompt} for prompt in prompts]) def diagram_reward_func(prompts, completions, **kwargs): scores = [] for prompt, completion in zip(prompts, completions, strict=True): text = completion if isinstance(completion, str) else completion[0]["content"] scores.append(reward(prompt, text)) return scores args = GRPOConfig( output_dir="outputs/diagram-grpo", learning_rate=1e-6, per_device_train_batch_size=2, gradient_accumulation_steps=4, num_generations=4, max_prompt_length=1024, max_completion_length=1024, logging_steps=1, ) trainer = GRPOTrainer( model="outputs/diagram-sft/final", reward_funcs=diagram_reward_func, args=args, train_dataset=dataset, ) trainer.train() ``` 这看起来比极简训练器短得多，因为基础设施被隐藏了。但在概念上，它是同一个循环：提示批次 -> 生成补全 -> 为补全打分 -> 计算优势 -> 更新策略这才是使用框架的正确方式。不是把它们当作魔法盒子，而是当作你已经理解的机制的实现。 ## 什么造就一个好的环境？ RL 智能体的质量往往受限于环境的质量。对于使用工具的智能体来说尤其如此。模型只能优化环境能够度量的东西。如果你的验证器只检查 JSON 合法性，你的智能体就会学会输出合法的 JSON。如果你的奖励检查模式合法性和任务完成情况，它就能学会完成任务。如果你的奖励包含一个脆弱的 LLM 裁判，模型可能学会的是裁判的偏好而不是用户的。一个好的环境具有几个特性。首先，它给出部分得分。如果每次失败都得零分，学习就会变得困难。格式错误的 JSON 输出应该比合法但不完整的图表得分更低，而合法但不完整的图表应该比完整、可读的图表得分更低。其次，它将语法与语义分开。合法性是必要的，但它不是任务本身。奖励应该区分“动作序列执行了”和“结果满足了请求”。第三，奖励应当难以被钻空子。如果模型可以通过创建包含所有提示词的巨大文本框来最大化奖励，它最终会学会这么做。如果它可以通过不相连的形状获得高奖励，它就会那么做。如果它能用表面的标签糊弄评判者，它就会那么做。第四，它应当产生有用的日志。每次 rollout 都应可检查：提示词、补全内容、解析错误、画布状态、奖励各组成部分、最终截图，以及（如果使用了评判者的话）评判理由。没有这些，奖励作弊将变得不可见。这就是为什么我越来越认为智能体训练首先是环境设计，其次才是算法选择。训练器决定如何使用信号。环境决定存在什么样的信号。 ## 常见的失败模式许多 RL 失败看起来很神秘，但如果你把它们视为环境或状态分布的失败，就不难理解了。如果模型始终得不到奖励，策略很可能没有访问到有效状态。这通常意味着动作模式过于严格，基座模型无法“说”这种动作语言，或者跳过了 SFT。解决办法不是盯着 RL 算法看，而是让有效区域变得可达：简化动作空间，生成教师轨迹，添加部分奖励，并训练模型产生有效动作。如果模型学会输出微小的琐碎图形，那可能是奖励过度看重有效性而低估了任务完成度。环境在说“有效就够了”，而模型在听。要增加实体覆盖率、最小形状数量、结构检查，以及基于提示词的语义评分。如果模型输出了有效的 JSON 却忽略了提示词，说明奖励大多是语法层面的。动作语言学会了，但任务没有。要增加将提示词与最终画布进行对比的检查。如果模型过拟合了教师的风格，说明 SFT 数据太窄，或者 RL 阶段太弱。增加教师多样性，为每个提示词采样多次尝试，使用多个教师模型，或者让环境奖励其他有效的解决方案。如果 RL 训练不稳定，说明相对于奖励质量和批大小而言，更新可能过于激进。降低学习率，缩短补全长度，添加梯度裁剪，添加参考模型的 KL 惩罚，增大批大小，并检查奖励方差。如果奖励上升但人工评估的质量没有上升，说明奖励被钻了空子。这是最重要的失败模式。人工检查样例，分解奖励，添加负面测试，直接惩罚漏洞行为。避免依赖来自单一脆弱启发式规则的单一标量。 ## 对于真实的 tldraw 智能体，有什么变化？玩具版的 Python 画布很有教育意义。真实的 tldraw 环境改变的是验证器，而不是循环本身。你不再需要把动作应用到一个假的 Canvas 上，而是在真实的编辑器中执行它们。启动 tldraw 应用或一个验证器页面。将 JSON 动作发送到编辑器中。让 tldraw 对其进行清理和应用。读回最终的形状、绑定、选区和错误。导出截图。根据形状状态、截图和任何验证器错误来计算奖励。策略仍然接收观测并输出动作。环境仍然执行动作。奖励仍然为结果打分。训练器仍然更新模型。这就是环境设计成为如此强大抽象的原因。一旦你能够暴露： reward(prompt, completion) -> float 或者，对于更丰富的设置： step(state, action) -> next_state, reward, done, info 你就可以把这个环境接入一个小型教学训练器、TRL、Unsloth、PRIME-RL、verl，或自定义的分布式训练栈。框架会变。问题的形状不会。 ## 完整的流水线完整的流水线是这样的。首先，定义环境。这包括动作模式、解析器、验证器、模拟器或渲染器，以及奖励函数。这是最重要的部分，因为它定义了模型实际在优化什么。然后生成提示词。从手写的种子提示词开始，再用合成提示词进行扩展。提示词要足够多样，使模型无法靠记忆一个狭窄的模式存活。然后生成教师轨迹。用一个更强的模型采样候选动作。通过环境验证它们。保留通过验证的轨迹，同时也保存被拒绝的轨迹，因为它们对调试模式和奖励很有用。然后对学生模型做 SFT。SFT 的目的是把模型带进有效动作空间。在做 RL 之前，先测量有效动作率和平均奖励。然后运行 RL。采样提示词，为每个提示词生成多个补全，用环境为它们打分，计算组内相对优势，并更新策略。然后评估。使用留出的提示词。跟踪解析率、模式有效率、环境奖励、语义得分、奖励分布，以及对渲染图形的人工审查。对于视觉任务，检查截图。对于工具任务，检查最终状态。然后迭代。改进奖励。添加更难的提示词。刷新教师轨迹。调整 RL 超参数。添加负面测试。重复。一个有用的压缩心智模型是：教师轨迹解决冷启动。 SFT 教会动作语言。 RL 在环境内优化行为。评估告诉你环境是否度量了正确的东西。 ## 结语智能体训练的魔力不在框架之中。框架很重要，尤其是在大规模场景下。分布式 rollout 生成、vLLM 服务、批处理、LoRA 训练、检查点保存、参考模型、日志记录和可复现性都是真实的工程问题。但它们不是概念上的核心。概念核心更小。定义动作空间。定义成功。生成轨迹。模仿好的轨迹。采样新的尝试。奖励好的尝试。更新模型。这就是整个系统的形状。 SFT 和 RL 在这个结构中扮演着不同的角色。SFT 将模型带入动作空间的有效区域。RL 则利用环境重塑策略，使其趋向更高回报的行为。教师轨迹解决了冷启动问题。验证器将执行转化为监督信号。理解了这一点，tldraw 智能体就不再是一个特例。浏览器智能体、编程智能体、电子表格智能体、机器人规划器和数学求解器都拥有相同的骨架：状态 -> 动作 -> 验证 -> 奖励 -> 学习剩下的就是工程问题了。 ## 参考资料 - Gemini 文本生成文档：https://ai.google.dev/gemini-api/docs/text-generation - Gemini 结构化输出文档：https://ai.google.dev/gemini-api/docs/structured-output - TRL 文档：https://huggingface.co/docs/trl/index - TRL GRPOTrainer 文档：https://huggingface.co/docs/trl/grpo_trainer - Unsloth RL 指南：https://docs.unsloth.ai/get-started/reinforcement-learning-rl-guide ## 相关链接 - [anshuman](https://x.com/athleticKoder) - [@athleticKoder](https://x.com/athleticKoder) - [34K](https://x.com/athleticKoder/status/2057091692235481560/analytics) - [https://ai.google.dev/gemini-api/docs/text-generation](https://ai.google.dev/gemini-api/docs/text-generation) - [https://ai.google.dev/gemini-api/docs/structured-output](https://ai.google.dev/gemini-api/docs/structured-output) - [https://huggingface.co/docs/trl/index](https://huggingface.co/docs/trl/index) - [https://huggingface.co/docs/trl/grpo_trainer](https://huggingface.co/docs/trl/grpo_trainer) - [https://docs.unsloth.ai/get-started/reinforcement-learning-rl-guide](https://docs.unsloth.ai/get-started/reinforcement-learning-rl-guide) - [升级到 Premium](https://x.com/i/premium_sign_up) - [2026年5月20日 21:30](https://x.com/athleticKoder/status/2057091692235481560) - [34.3K 次浏览](https://x.com/athleticKoder/status/2057091692235481560/analytics) - [查看引用](https://x.com/athleticKoder/status/2057091692235481560/quotes) --- *导出时间: 2026/5/21 09:54:27*

相关文档