feat: flow 第 1 部分(骨架、研究开始、第零轮)与第零轮检验

按任务书 12 实现 src/flow:从账本现算下一步的纯函数、主循环、恢复入口、派模型任务和跑正式运行两个共用动作、冻结配置、基线、第零轮三个研究者阶段和入口分配,走到第 1 轮时停下。materials 导出 finalEntries、加 researcherId 和 StudyRecord。

审阅采纳两条:研究编号本机已用过时冻结前报错;docker run 没启动时写完结果就停下,不在本进程里无限起新编号。

eval/ 新增第零轮检验:起研究命令行、回放假模型、检验用 Jev 客户端、六问质量评估、16-real-1 回放、杀进程与两份同时跑。tsc 与 depcruise 也检查 eval/,新增 eval 只经 index.ts 用正式代码的规则。

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-03 10:58:52 -04:00
parent 8491191a2b
commit 75d7cffd27
36 changed files with 2456 additions and 14 deletions
+27
View File
@@ -0,0 +1,27 @@
import { parseArgs } from "node:util";
import { loadConfig } from "../../src/config/index.ts";
import { runStudy } from "../../src/flow/index.ts";
// 起研究的命令行入口:像用户一样给配置、指定研究目录、调 runStudy,按 Ctrl-C(SIGINT)时发"停下"信号。
// 新研究:node eval/lib/cli.ts --config <research.json> --dir <研究目录>
// 接着跑:node eval/lib/cli.ts --resume --dir <研究目录>(配置写 null,以账本里冻结的那份为准)
// 结局打印成一行 "RESULT <json>",检验脚本按这一行读。
const { values } = parseArgs({
options: { config: { type: "string" }, resume: { type: "boolean" }, dir: { type: "string" } },
strict: true,
});
if (values.dir === undefined) throw new Error("要写 --dir <研究目录>");
if ((values.config === undefined) === (values.resume !== true)) {
throw new Error("--config <research.json> 和 --resume 二选一,必须写一个");
}
const config = values.config === undefined ? null : loadConfig(values.config);
const stop = new AbortController();
process.on("SIGINT", () => {
console.error("收到 SIGINT,发停下信号");
stop.abort();
});
const outcome = await runStudy({ studyDir: values.dir, config, stop: stop.signal });
console.log(`RESULT ${JSON.stringify(outcome)}`);
+71
View File
@@ -0,0 +1,71 @@
// 检验自己用的小 Jev 客户端(任务书 12 第四部分第 6 条):不经 src/jev,也不改它。
// 请求格式参照 src/jev/jev.ts 和 RivalBits research-wiki/参考-Jev使用指南.md 第二部分:
// POST {model, state, questions},回答里 answers.<问题名>.noul 是 0 到 1 的概率。
// 检验的问题都没有校准过阈值,一律以概率大于 0.5 判"是",原始概率照存(任务书 12 第 5.2 节)。
const ENDPOINT = "https://api.typesafe.ai/v1/systemone";
/** 检验用的 Jev 模型名,和研究配置的 models.jev 同一个。 */
export const JEV_MODEL = "jev-latest";
export const THRESHOLD = 0.5;
export const THRESHOLD_NOTE = "阈值未校准:一律以概率大于 0.5 判\"是\"";
/** 一条是非题,写法照 src/jev/questions.ts:问题、关注点、判"是"判"否"的说明和反例。 */
export interface Noul {
type: "noul";
instructions: { question: string; focus: string };
criteria: {
true: { what: string; examples: string[] };
false: { what: string; not_for: string; examples: string[] };
};
}
export interface Answer {
question: string;
probability: number;
yes: boolean;
requestId: string;
}
/** 调不通时重试前等多久:只重试连不上、超时、429 和 5xx(原因在我们控制之外),一共最多 4 次。 */
const RETRY_WAITS_MS = [2_000, 10_000, 30_000];
/** 就一份材料问几问,一次请求。 */
export async function ask(state: Record<string, unknown>, questions: Record<string, Noul>): Promise<Answer[]> {
const key = process.env["TYPESAFE_API_KEY"];
if (key === undefined || key.trim() === "") throw new Error("环境变量 TYPESAFE_API_KEY 没有设置");
const body = JSON.stringify({ model: JEV_MODEL, state, questions });
const failures: string[] = [];
for (let attempt = 0; attempt <= RETRY_WAITS_MS.length; attempt++) {
if (attempt > 0) await new Promise((ok) => setTimeout(ok, RETRY_WAITS_MS[attempt - 1]));
let response: Response;
let text: string;
try {
response = await fetch(ENDPOINT, {
method: "POST",
headers: { "content-type": "application/json", authorization: `Bearer ${key}` },
body,
signal: AbortSignal.timeout(60_000),
});
text = await response.text();
} catch (error) {
failures.push((error as Error).message);
continue;
}
const requestId = response.headers.get("x-typesafe-request-id") ?? "无";
if (!response.ok) {
const what = `HTTP ${response.status}(request-id: ${requestId}):${text.slice(0, 300)}`;
if (response.status !== 429 && response.status < 500) throw new Error(`Jev 拒绝了请求:${what}`);
failures.push(what);
continue;
}
const answers = (JSON.parse(text) as { answers?: Record<string, { noul?: unknown }> }).answers;
return Object.keys(questions).map((question) => {
const probability = answers?.[question]?.noul;
if (typeof probability !== "number" || !(probability >= 0 && probability <= 1)) {
throw new Error(`Jev 的回答里问题 ${question} 没有 0 到 1 之间的概率(request-id: ${requestId}):${text.slice(0, 300)}`);
}
return { question, probability, yes: probability > THRESHOLD, requestId };
});
}
throw new Error(`Jev 调不通,发了 ${failures.length} 次:${failures.join(";")}`);
}
+89
View File
@@ -0,0 +1,89 @@
import { existsSync } from "node:fs";
import { join } from "node:path";
import { Ledger, type LedgerRecord } from "../../src/ledger/index.ts";
import { entryId, type DecisionShapes } from "../../src/materials/index.ts";
// 读账本的小工具:只经 src/ledger/index.ts 打开、只读记录,不写。
// 入口编号用 materials 的 entryId,和正式代码同一条编号规则。
/** 研究目录里账本的全部记录。账本还没有时报错:Ledger.open 会新建一个空账本,检验不能替 flow 建。 */
export function readLedger(studyDir: string): LedgerRecord[] {
const path = join(studyDir, "ledger.sqlite");
if (!existsSync(path)) throw new Error(`${path} 还不存在`);
const ledger = Ledger.open(path);
try {
return ledger.records();
} finally {
ledger.close();
}
}
export type Of<K extends LedgerRecord["kind"]> = Extract<LedgerRecord, { kind: K }>;
export function ofKind<K extends LedgerRecord["kind"]>(records: LedgerRecord[], kind: K): Array<Of<K>> {
return records.filter((r): r is Of<K> => r.kind === kind);
}
export interface StageInfo {
stageId: string;
stage: string;
round: number;
roster: string[];
closed: boolean;
}
export function stages(records: LedgerRecord[]): StageInfo[] {
const closed = new Set(ofKind(records, "stageClose").map((r) => r.content.stageId));
return ofKind(records, "stageOpen").map(({ content: c }) => ({ ...c, closed: closed.has(c.stageId) }));
}
/** 每份提交(阶段、作者、种类、对象、编号)的最新一版,按这一版的序号排。 */
export function finalSubmissions(records: LedgerRecord[]): Array<Of<"submission">> {
const latest = new Map<string, Of<"submission">>();
for (const r of ofKind(records, "submission")) {
const c = r.content;
const slot = JSON.stringify([c.stageId, c.author, c.kind, c.target, c.item]);
const seen = latest.get(slot);
if (seen === undefined || seen.content.version < c.version) latest.set(slot, r);
}
return [...latest.values()].sort((a, b) => a.seq - b.seq);
}
export interface Entry {
id: string;
author: string;
name: string;
body: string;
}
/**
* 只看给定阶段里的提交,各人最后一版入口清单里的入口(方法.md 4.2:讨论时整份替换的新清单接着往后编号)。
* 给第零轮全部阶段就是分配时的"入口全体";只给写入口阶段就是闭门写完时的入口全体。
*/
export function entriesIn(records: LedgerRecord[], stageIds: ReadonlySet<string>): Entry[] {
const lists = new Map<string, Entry[]>();
const used = new Map<string, number>();
for (const s of finalSubmissions(records)) {
const c = s.content;
if (!stageIds.has(c.stageId) || (c.kind !== "entry" && c.kind !== "discussion")) continue;
const entries = (c.content as { entries?: Array<{ name: string; body: string }> }).entries;
if (entries === undefined) continue;
const n = used.get(c.author) ?? 0;
lists.set(c.author, entries.map((e, i) => ({ id: entryId(c.author, n + i + 1), author: c.author, ...e })));
used.set(c.author, n + entries.length);
}
return [...lists.values()].flat();
}
export type Assignment = DecisionShapes["entries"]["output"][number];
/** 第 0 轮的入口分配决定:输入(全部志愿和入口)和输出(分配结果);没有就是 undefined。 */
export function entryDecision(records: LedgerRecord[]): { input: { entries: Array<{ id: string; author: string }> }; output: Assignment[] } | undefined {
const d = ofKind(records, "decision").find((r) => r.content.decision === "entries" && r.content.round === 0);
return d === undefined ? undefined : (d.content as unknown as { input: { entries: Array<{ id: string; author: string }> }; output: Assignment[] });
}
/** 两份分配是否相同:每人分到的入口和第几志愿都一样,不看先后和对象里键的顺序。 */
export function sameAssignment(a: Assignment[], b: Assignment[]): boolean {
const key = (xs: Assignment[]) => xs.map((x) => `${x.researcher} ${x.entry} ${x.rank}`).sort().join("\n");
return key(a) === key(b);
}
+100
View File
@@ -0,0 +1,100 @@
import { createServer, type ServerResponse } from "node:http";
import type { AddressInfo } from "node:net";
import { mkdirSync, writeFileSync } from "node:fs";
import { join } from "node:path";
// 回放用的假模型服务(任务书 12 第四部分第 3 条):本机起一个兼容 OpenAI 的流式接口,
// 用 Pi 自己的 PI_CODING_AGENT_DIR 指一份 models.json 过来,正式代码不知道对面是假模型。
// 每个模型名一份按顺序交出的提交清单:每次被派任务(会话的第一次请求),交出下一份;之后的请求回一句话结束。
// 交完了就不再交,只回一句话:不为凑结果改内容。被退回时模型没有别的可交,会话就以"没交"结束。
/** 一份要交的提交:research_submit 的参数原样(payload,以及可能的 replaces),from 写它从哪来,只用于报告。 */
export interface ReplayItem {
from: string;
arguments: { payload: unknown; replaces?: string };
}
/** 回放脚本:模型名 → 按顺序交出的提交。没有要交的模型(实验员、核验员)写空数组。 */
export interface ReplayScript {
source: string;
models: Record<string, ReplayItem[]>;
}
export interface Replay {
/** 起研究的子进程要带的环境变量。 */
env: Record<string, string>;
/** 每个模型名已经交出的提交(按交出先后)。 */
delivered: Record<string, string[]>;
/** 每个模型名还没交出的提交。 */
remaining: () => Record<string, string[]>;
close: () => Promise<void>;
}
export const REPLAY_PROVIDER = "replay";
/**
* 起服务并写好 Pi 的 models.json、auth.json。delayMs 是每次回复前等多久(杀进程检验要一段"正在写"的窗口,0 就是不等)。
* 一份提交只有在回复完整写出后才算交出:回复前对面断开(控制进程被杀),这一份放回清单最前面,下次被派任务时还交它。
*/
export async function startReplay(script: ReplayScript, agentDir: string, delayMs: number): Promise<Replay> {
const queues = new Map(Object.entries(script.models).map(([model, items]) => [model, [...items]]));
const delivered: Record<string, string[]> = Object.fromEntries([...queues.keys()].map((m) => [m, []]));
const server = createServer((req, res) => {
const chunks: Buffer[] = [];
req.on("data", (c: Buffer) => chunks.push(c));
req.on("end", () => void answer(JSON.parse(Buffer.concat(chunks).toString()), res));
});
async function answer(body: { model: string; messages: Array<{ role: string }> }, res: ServerResponse): Promise<void> {
const queue = queues.get(body.model);
if (queue === undefined) throw new Error(`回放脚本里没有模型 ${body.model}`);
// 会话的第一次请求里还没有助手的回复:这是新派的任务。
const item = body.messages.some((m) => m.role === "assistant") ? undefined : queue.shift();
res.on("close", () => {
if (!res.writableFinished && item !== undefined) queue.unshift(item);
});
await new Promise((ok) => setTimeout(ok, delayMs));
if (res.destroyed) return;
const send = (delta: unknown, finish: string | null) =>
res.write(`data: ${JSON.stringify({ id: "r", object: "chat.completion.chunk", created: 1, model: body.model, choices: [{ index: 0, delta, finish_reason: finish }] })}\n\n`);
res.writeHead(200, { "content-type": "text/event-stream" });
if (item === undefined) {
send({ role: "assistant", content: "好的,我这次没有别的要交了。" }, null);
send({}, "stop");
} else {
const args = JSON.stringify(item.arguments);
send({ role: "assistant", tool_calls: [{ index: 0, id: "call-1", type: "function", function: { name: "research_submit", arguments: args } }] }, null);
send({}, "tool_calls");
}
res.write(`data: ${JSON.stringify({ id: "r", object: "chat.completion.chunk", created: 1, model: body.model, choices: [], usage: { prompt_tokens: 100, completion_tokens: 10 } })}\n\n`);
res.end("data: [DONE]\n\n");
if (item !== undefined) delivered[body.model]!.push(item.from);
}
await new Promise<void>((ok) => server.listen(0, "127.0.0.1", ok));
const baseUrl = `http://127.0.0.1:${(server.address() as AddressInfo).port}/v1`;
const models = [...queues.keys()].map((id) => ({
id,
reasoning: false,
input: ["text"],
contextWindow: 1_000_000,
maxTokens: 32_000,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
}));
mkdirSync(agentDir, { recursive: true });
writeFileSync(join(agentDir, "models.json"), JSON.stringify({ providers: { [REPLAY_PROVIDER]: { baseUrl, api: "openai-completions", apiKey: "k", models } } }, null, 2));
writeFileSync(join(agentDir, "auth.json"), "{}");
return {
env: { PI_CODING_AGENT_DIR: agentDir },
delivered,
remaining: () => Object.fromEntries([...queues].map(([m, q]) => [m, q.map((i) => i.from)])),
close: () => new Promise<void>((ok) => server.close(() => ok())),
};
}
/** 回放用的各角色模型:研究者按脚本里的模型名,实验员、核验员各一个空清单的名字。 */
export function replayModels(researchers: string[], jev: string): unknown {
const m = (model: string) => ({ provider: REPLAY_PROVIDER, model, thinking: "off" });
return { researchers: researchers.map(m), experimenter: m("experimenter"), verifier: m("verifier"), jev };
}
+82
View File
@@ -0,0 +1,82 @@
import { execFileSync, spawn, type ChildProcess } from "node:child_process";
import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs";
import { join, resolve } from "node:path";
// 检验脚本共用的起研究工具:检验目录放在仓库外,配置照用户的写法写一份,研究在子进程里跑(要能被 SIGKILL)。
/** 检验输出一律放在仓库外(任务书 12 第四部分第 4 条),不提交。 */
export const EVAL_ROOT = "/home/iomgaa/inforsi-eval";
/** 新建一个检验目录 <日期-名字>;已经有了就报错,不把两次检验的产物混在一起。 */
export function newEvalDir(name: string): string {
const dir = join(EVAL_ROOT, `${new Date().toISOString().slice(0, 10)}-${name}`);
if (existsSync(dir)) throw new Error(`检验目录 ${dir} 已经有了,换个名字或先挪走`);
mkdirSync(dir, { recursive: true });
return dir;
}
/**
* 照一份底稿写这次检验用的 research.json:换研究编号(容器名、镜像标签都带它),需要时换各角色模型;
* 任务目录、提示词目录写成绝对路径(底稿里相对仓库根目录),这样配置文件放在仓库外也指得对。
*/
export function writeConfig(base: string, path: string, studyId: string, models: unknown): string {
const config = JSON.parse(readFileSync(base, "utf8")) as Record<string, unknown>;
config["studyId"] = studyId;
config["task"] = resolve(config["task"] as string);
config["promptDir"] = resolve(config["promptDir"] as string);
if (models !== null) config["models"] = models;
writeFileSync(path, `${JSON.stringify(config, null, 2)}\n`);
return path;
}
export interface StudyProcess {
child: ChildProcess;
/** 子进程退出:退出码、信号,以及它打印的结局(被杀掉时没有,是 null)。 */
done: Promise<{ code: number | null; signal: NodeJS.Signals | null; outcome: unknown }>;
}
/** 在子进程里起 eval/lib/cli.ts。config 为 null 是接着跑。子进程的输出原样转到本进程,并记进 <研究目录>.log。 */
export function launch(config: string | null, studyDir: string, env: Record<string, string>): StudyProcess {
const args = ["eval/lib/cli.ts", ...(config === null ? ["--resume"] : ["--config", config]), "--dir", studyDir];
const child = spawn(process.execPath, args, { env: { ...process.env, ...env }, stdio: ["ignore", "pipe", "pipe"] });
const log: string[] = [];
let stdout = "";
child.stdout!.on("data", (chunk: Buffer) => {
stdout += chunk.toString();
log.push(chunk.toString());
process.stdout.write(chunk);
});
child.stderr!.on("data", (chunk: Buffer) => {
log.push(chunk.toString());
process.stderr.write(chunk);
});
const done = new Promise<{ code: number | null; signal: NodeJS.Signals | null; outcome: unknown }>((ok) =>
child.on("close", (code, signal) => {
writeFileSync(`${studyDir}.log`, log.join(""), { flag: "a" });
const line = stdout.split("\n").find((l) => l.startsWith("RESULT "));
ok({ code, signal, outcome: line === undefined ? null : JSON.parse(line.slice("RESULT ".length)) });
}),
);
return { child, done };
}
export function sleep(ms: number): Promise<void> {
return new Promise((ok) => setTimeout(ok, ms));
}
/** 本机名字以 prefix 开头的容器(含已停止的)。检验直接问 Docker,不经正式代码。 */
export function containers(prefix: string): string[] {
const names = execFileSync("docker", ["ps", "-a", "--format", "{{.Names}}"], { encoding: "utf8" });
return names.split("\n").filter((n) => n.startsWith(prefix));
}
/**
* 等 flow 把账本的表建好,检验才第一次打开账本:Ledger.open 遇到空文件会自己建表,和 flow 同时建会冲突。
* 信号是研究目录里出现 repo.git:flow 先 Ledger.open(建表),冻结配置时才建裸仓库(src/flow/loop.ts、study.ts 的 freezeConfig)。
*/
export async function waitForLedger(studyDir: string, child: ChildProcess): Promise<void> {
while (!existsSync(join(studyDir, "repo.git"))) {
if (child.exitCode !== null) throw new Error(`研究进程没建出 repo.git 就退出了,退出码 ${child.exitCode}`);
await sleep(50);
}
}