复现 Memdoor：5 个 27B 小模型组队打 LiveCodeBench Hard

这篇能做出什么

按本文走完一遍，你会得到一个可运行的「多小模型组队」系统：5 个 27B 级别的模型实例（同一个权重也可以是不同权重）通过角色分工、并行采样、结果聚合三层结构协作，去解 LiveCodeBench Hard 难度的竞赛编程题。同时你会得到一套对照测试脚本，能用自己的机器、自己的题量，跑出「单模型 vs 团队」在准确率、端到端延迟、token 成本三个维度的实测数字，而不是照抄别人的结论。

具体交付物：

• 一个本地 OpenAI 兼容推理服务，供 5 路并发调用；
• 一套角色提示词：Planner、Solver×5（分工不同）、Tester、Reviewer；
• 一个沙箱执行器，负责跑候选代码、比对输出、超时保护；
• 一层聚合逻辑：行为投票 → 阈值路由 → 复核兜底；
• 一个对照 runner，输出准确率 / 延迟 / 成本三联表。

先说清楚一个反直觉的点：团队方案几乎一定更贵，延迟也不一定更低。它换来的是准确率的提升。如果你的题面简单，团队反而会拉低性价比。所以本文的重点不只是「怎么搭」，还有「怎么用数据判断值不值得搭」。

Memdoor 这一路思路的共同点是：多个中等规模模型各司其职，靠结构而不是靠单个模型的参数量取胜。本文按这个思路做工程复现，具体设计是本文给出的可运行版本，细节以你所参考的原始描述和官方文档为准。

前置条件清单

硬件与模型

• 至少一张能跑 27B 级别模型（4bit 量化或 FP8）的 GPU。5 个实例不必是 5 张卡，用张量并行起一个推理服务、开 5 路并发即可，等价于「5 个成员」。
• 推理服务推荐 vLLM 或同类方案，暴露 OpenAI 兼容接口。安装方式与参数以官方文档当前版本为准。
• 备选方案：直接用云端 API 上的 27B 级模型，按 token 付费。这样省去部署，但成本口径不同，后面对照时要分清楚。

软件

• Python 3.10+，openai、datasets、tqdm、numpy、pandas。
• Docker（强烈建议），用于隔离执行模型生成的代码。

数据集

• LiveCodeBench 官方仓库的题集导出文件。以官方仓库当前说明为准，注意按时间窗口和难度筛选，只保留 difficulty == hard 的题。
• 题目的评测方式可能是 stdin/stdout，也可能是函数签名调用。本文统一按 stdin/stdout 实现，如果你用的版本是函数调用，改一个 adapter 即可。

一个必须先做的检查

确认所选模型的训练截止时间早于题目的发布时间窗口。否则你测到的不是推理能力，是记忆。这一步只花十分钟，省掉后面所有争论。

步骤一：起推理服务并验证并发

用 vLLM 起一个服务（参数以官方文档为准）：

```bash

python -m vllm.entrypoints.openai.api_server \

--model /path/to/your-27b-model \

--served-model-name team-27b \

--max-model-len 16384 \

--gpu-memory-utilization 0.9 \

--port 8000

```

验证：

```bash

curl http://127.0.0.1:8000/v1/models

```

能返回 team-27b 就通了。注意 --max-model-len 要能装下最长题面加最长回答；设太小会静默截断，表现为模型「话说到一半就没了」。

封装一个客户端，后面所有角色都用它：

```python

llm.py

import asyncio, os

from openai import AsyncOpenAI

client = AsyncOpenAI(

base_url=os.getenv("BASE_URL", "http://127.0.0.1:8000/v1"),

api_key=os.getenv("API_KEY", "EMPTY"),

)

MODEL = os.getenv("MODEL", "team-27b")

async def chat(system: str, user: str, temperature: float = 0.2,

max_tokens: int = 4096) -> dict:

"""返回 {'text': str, 'ptok': int, 'ctok': int}"""

resp = await client.chat.completions.create(

model=MODEL,

messages=[{"role": "system", "content": system},

{"role": "user", "content": user}],

temperature=temperature,

max_tokens=max_tokens,

)

return {

"text": resp.choices[0].message.content or "",

"ptok": resp.usage.prompt_tokens,

"ctok": resp.usage.completion_tokens,

}

```

加一个全局并发闸门，避免 5×k 路请求把服务打爆：

```python

SEM = asyncio.Semaphore(8) # 按显存实测调整

async def guarded_chat(*args, **kwargs):

async with SEM:

return await chat(*args, **kwargs)

```

步骤二：准备题库与沙箱执行器

读题、筛 Hard：

```python

data.py

import json

def load_problems(path: str, difficulty: str = "hard") -> list[dict]:

items = []

with open(path, encoding="utf-8") as f:

for line in f:

line = line.strip()

if not line:

continue

ex = json.loads(line)

if ex.get("difficulty", "").lower() != difficulty:

continue

items.append({

"id": ex.get("question_id") or ex.get("id"),

"prompt": ex.get("question_content") or ex.get("problem"),

"public": ex.get("public_test_cases") or [],

"private": ex.get("private_test_cases") or [],

})

return items

```

字段名以你手上的导出文件为准，用 .get() 兜底，跑不通就先打印一条样本看结构。

沙箱执行器（开发期够用，生产请换容器）：

```python

sandbox.py

import os, subprocess, sys, tempfile

def run_python(code: str, stdin_data: str = "", timeout: int = 10) -> dict:

with tempfile.TemporaryDirectory() as d:

path = os.path.join(d, "main.py")

with open(path, "w", encoding="utf-8") as f:

f.write(code)

try:

r = subprocess.run(

[sys.executable, path],

input=stdin_data, capture_output=True, text=True,

timeout=timeout, cwd=d,

)

return {"ok": r.returncode == 0,

"stdout": r.stdout,

"stderr": r.stderr[-2000:]}

except subprocess.TimeoutExpired:

return {"ok": False, "stdout": "", "stderr": "TIMEOUT"}

except Exception as e:

return {"ok": False, "stdout": "", "stderr": repr(e)}

```

输出比对要宽容一点，否则误判会污染统计：

```python

def normalize(s: str) -> str:

return "\n".join(line.rstrip() for line in s.strip().splitlines())

def same(a: str, b: str, is_float: bool = False) -> bool:

if is_float:

try:

return abs(float(a) - float(b)) <= 1e-6

except ValueError:

return False

return normalize(a) == normalize(b)

```

安全提醒：模型会写出 while True、递归爆栈、写文件、os.system 的代码。开发机上也请至少放在容器里跑，并且限制 CPU 时间和内存。

步骤三：第一层——角色分工

5 个 27B 实例不是「同样的模型问五遍」，那样投票收益很小。让它们带不同视角入场。

Planner（规划者，1 次调用）：只出思路不写代码。

```python

PLANNER_SYS = """你是竞赛编程的规划者。读题后输出：

1) 数据规模与时限推算出的目标复杂度；

2) 主解法的算法骨架（伪代码级别，不写完整实现）；

3) 至少两个容易出错的边界情况；

4) 一句提醒：输入输出必须走 stdin/stdout，不要写 if __name__ == '__main__' 里的示例测试。

不要输出完整代码。"""

```

Solver ×5（并行，各自不同倾向）：

```python

SOLVER_STYLES = {

"direct": "直接写出最朴素但正确的解法，优先保证正确性。",

"complex": "先根据数据规模确认复杂度上限，再写实现，注意常数因子。",

"edge": "以边界与反例为第一优先级，写完后在脑中逐条检查空输入、单元素、极值。",

"template": "优先套用成熟的数据结构/算法模板（线段树、并查集、图遍历等），保证实现稳健。",

"alt": "如果规划者的思路有第二条路线，走那条；没有就换一个角度重新建模。",

}

SOLVER_SYS = """你是竞赛编程的实现者。只输出一个 Python 代码块，符合 stdin/stdout 规范，

不包含任何示例测试、不使用 input() 之外的交互、不写文件、不联网。

代码要能直接运行。{style}"""

```

Tester（1 次调用）：不看解法，只造测试。

```python

TESTER_SYS = """你是测试工程师。给定题目，输出一个 Python 脚本：

从 stdin 读入一个随机种子，向 stdout 打印一组该题目的合法输入。

要求覆盖小规模、边界规模与一个随机中等规模。只输出代码块。"""

```

Reviewer（1 次调用）：拿题面 + 若干候选 + 执行结果做最终判断。

```python

REVIEWER_SYS = """你是评审。你会看到题目、若干候选代码，以及每个候选在公开样例和随机对拍上的表现。

你的任务是：

• 若存在明显更可靠的候选，直接返回它的编号；
• 否则指出各候选的共同缺陷，并给出修正后的完整代码块；
• 如果全部候选都错，输出 REWRITE 并给出新的实现思路。

只输出：编号 / 代码块 / REWRITE + 思路。"""

```

步骤四：第二层——并行采样

一层扇出：Planner 出提纲 → 5 个 Solver 并发 → 每个 Solver 采 k 个候选。k 建议从 2 开始，先跑通再调大。

```python

sample.py

import asyncio

from llm import guarded_chat

from sandbox import run_python, same

async def sample_one(problem: dict, plan: str, style_name: str,

style_desc: str, k: int = 2, temperature: float = 0.7):

sys = SOLVER_SYS.format(style=style_desc)

user = f"题目：\n{problem['prompt']}\n\n规划者提纲：\n{plan}"

tasks = [guarded_chat(sys, user, temperature=temperature, max_tokens=4096)

for _ in range(k)]

outs = await asyncio.gather(*tasks, return_exceptions=True)

results = []

for o in outs:

if isinstance(o, Exception):

continue

code = extract_code(o["text"])

results.append({"style": style_name, "code": code,

"ptok": o["ptok"], "ctok": o["ctok"]})

return results

async def sample_team(problem: dict, plan: str, k: int = 2):

tasks = [sample_one(problem, plan, name, desc, k)

for name, desc in SOLVER_STYLES.items()]

batches = await asyncio.gather(*tasks)

return [c for b in batches for c in b]

```

代码块提取取最长的那一段，能避开「先解释后给代码」的坑：

```python

import re

def extract_code(text: str) -> str:

blocks = re.findall(r"``(?:python)?\s*\n(.*?)``", text, re.S)

return max(blocks, key=len).strip() if blocks else text.strip()

```

采样完立刻过沙箱，得到每个候选的「行为指纹」——这是下一步投票的原料：

```python

def evaluate(problem: dict, cand: dict, tester_code: str, n_probe: int = 5):

"""返回 {'pass_public': bool, 'fingerprint': tuple, 'err': str}"""

ok_public = True

for tc in problem["public"]:

r = run_python(cand["code"], tc.get("input", ""))

if not r["ok"] or not same(r["stdout"], tc.get("output", "")):

ok_public = False

break

fp = []

for seed in range(n_probe):

gen = run_python(tester_code, str(seed))

if not gen["ok"]:

fp.append("GEN_FAIL")

continue

r = run_python(cand["code"], gen["stdout"])

fp.append(normalize(r["stdout"])[:200] if r["ok"] else "ERR")

return {"pass_public": ok_public, "fingerprint": tuple(fp), "err": ""}

```

步骤五：第三层——聚合：投票、路由、复核

投票用行为一致性，不用文本相似度。 竞赛编程题的代码写法千差万别，但正确解在同一组随机输入上的输出必须一致。所以把「同一组输入下的输出序列」当签名做聚类，最大簇就是事实上的多数答案。

```python

from collections import Counter

def vote(cands: list[dict]):

pool = [c for c in cands if c["eval"]["pass_public"]]

if not pool:

pool = cands

sigs = Counter(c["eval"]["fingerprint"] for c in pool)

if not sigs:

return None, 0.0

top_sig, top_n = sigs.most_common(1)[0]

winner = next(c for c in pool if c["eval"]["fingerprint"] == top_sig)

return winner, top_n / len(pool)

```

路由定阈值。 一致性足够高就直接交卷，省掉复核的延迟和 token：

```python

CONSENSUS = 0.6 # 最大簇占比阈值，按实测调

def route(winner, ratio, cands):

if winner and ratio >= CONSENSUS and winner["eval"]["pass_public"]:

return "accept", winner

if winner and winner["eval"]["pass_public"]:

return "review", winner # 有候选但意见分裂

return "rewrite", winner # 没人通过公开样例

```

阈值怎么定？在开发集上扫 0.4 / 0.5 / 0.6 / 0.8，画一条「准确率 vs 复核调用率」的曲线，选拐点。别凭感觉。

复核兜底。 只把分裂或全错的题送去 Reviewer，这是整套结构里省成本的关键：

```python

async def aggregate(problem, cands, tester_code):

for c in cands:

c["eval"] = evaluate(problem, c, tester_code)

winner, ratio = vote(cands)

decision, base = route(winner, ratio, cands)

if decision == "accept":

return {"final": base["code"], "path": "vote", "extra_tok": 0}

brief = "\n\n".join(

f"候选{i}（{c['style']}，公开样例{'通过' if c['eval']['pass_public'] else '失败'}）：\n{c['code'][:1500]}"

for i, c in enumerate(cands[:4]))

r = await guarded_chat(REVIEWER_SYS, f"题目：\n{problem['prompt']}\n\n{brief}",

temperature=0.2, max_tokens=4096)

return {"final": extract_code(r["text"]), "path": "review",

"extra_tok": r["ptok"] + r["ctok"]}

```

再加一条保命规则：复核后仍失败，回退到单模型高温度重写一次。团队不是万能的，兜底比死循环好。

步骤六：端到端跑通一小批

先取 20 道 Hard 题，跑通再放大：

```python

run_team.py

import asyncio, json, time

from data import load_problems

from llm import guarded_chat

from sample import sample_team

from aggregate import aggregate

async def solve_team(problem, k=2):

t0 = time.time()

plan = await guarded_chat(PLANNER_SYS, problem["prompt"], 0.3, 2048)

tester = await guarded_chat(TESTER_SYS, problem["prompt"], 0.2, 2048)

cands = await sample_team(problem, plan["text"], k=k)

res = await aggregate(problem, cands, extract_code(tester["text"]))

tok = plan["ptok"] + plan["ctok"] + tester["ptok"] + tester["ctok"] \

+ sum(c["ptok"] + c["ctok"] for c in cands) + res["extra_tok"]

return {"id": problem["id"], "final": res["final"], "path": res["path"],

"wall": time.time() - t0, "tokens": tok}

async def main():

problems = load_problems("lcb_hard.jsonl")[:20]

outs = []

for p in problems:

outs.append(await solve_team(p))

print(p["id"], outs[-1]["path"], round(outs[-1]["wall"], 1), "s")

with open("team_result.json", "w") as f:

json.dump(outs, f, ensure_ascii=False)

asyncio.run(main())

```

跑完后把 final 里的代码拿去跑隐藏用例，得到 pass@1。这一步必须用官方评测口径，自己写的比对逻辑容易和官方判定不一致。

步骤七：与单个前沿模型做对照实测

对照要控制变量：同一批题、同一台机器、同一份评测脚本。三个方案各跑一遍：

• A：单模型单次，temperature=0.2；
• B：单模型 + 一轮自我修复（执行失败就把报错喂回去重写）；
• C：本教程的团队方案。

每题的延迟按端到端 wall-clock 记，成本按 token 记：

```python

def cost_of(ptok, ctok, in_price_per_m, out_price_per_m):

"""单价单位：每百万 token 的价格，以官方页面当前公示为准。"""

return ptok / 1e6 * in_price_per_m + ctok / 1e6 * out_price_per_m

```

自托管的时候没有 token 单价，换成 GPU 小时口径：

```

单题摊销成本 ≈ 单卡小时价 × 占用小时数 ÷ 该时段完成的题数

```

填写这张表，数字全部来自你自己的跑测：

读表时注意三件事：

1. 延迟口径：并行采样压的是每题内部串行等待，但复核和重写会往上加。P90 比 P50 更能反映用户体验，因为 Hard 题的长尾很凶。

2. 成本口径：云 API 按 token 计价，自托管按 GPU 时间摊销，两者不能直接比大小，但可以比「同样预算下的通过题数」。

3. 规模效应：20 道题的数字波动很大，至少跑到 100 道以上再下结论，且看置信区间。

如果 C 相对 A 的准确率提升不明显，先查两件事：k 是不是太小导致候选池不够多样；温度是不是太低让 5 个 Solver 写出了几乎一样的代码。

常见坑与排错

模型输出里带 if __name__ == '__main__': 的示例测试。这会让程序在读取 stdin 之前先打印示例，输出比对全错。提示词里明确禁止，同时在 extract_code 后做一次正则清理，把这一段删掉。

5 路并发把显存打满，延迟反而变长。表现是单请求延迟没变，端到端延迟飙升。降低 SEM 的上限，或者调小 --max-num-seqs，用实测延迟曲线找拐点，不要盲目加大并发。

投票投出「一致地错」。5 个 Solver 用同一份提纲、同一个模型，很容易集体踩同一个坑，最大簇百分百一致但答案是错的。对策：alt 这个角色必须真的换建模角度；温度拉开到 0.7~0.9；投票前必须过公开样例这道硬门槛。

随机对拍数据不合法。Tester 造出的输入可能违反题目约束（比如 n 超范围），导致所有候选都崩，簇全变成 ERR，投票失去意义。让 Tester 的输出先过一遍你自己的校验器，或者干脆只用题目公开样例做指纹。

浮点输出误判。输出是小数时用容差比较；同时注意有些题目要求保留固定小数位，1.0 和 1.00 在字符串比对下不等但实际等价。用 is_float 开关分流处理。

沙箱里死循环。timeout=10 是底线，不是建议。再叠加一层 resource.setrlimit 限制内存，否则一段 [0]*10**9 就能把机器拖垮。

题目时间窗口没对齐。模型训练数据里可能包含早期题目的题解与测试用例。用官方的时间划分筛选，并且明确记录所选窗口，否则复现结果无法解释。

评测口径自己实现。隐藏用例的调用方式、输入格式、多组测试的处理，官方评测脚本都有约定。自己写一套，准确率数字会和别人对不上，而且通常会偏低。

没有记录 trace。跑完一批只留了最终分数，后面想分析「复核到底救回了多少题」就无从下手。每题落盘：走了哪条路径（vote / review / rewrite）、各候选的通过情况、token 明细。

下一步建议

加难度路由。不是所有题都值得上团队。先用单模型跑一遍，只有失败或输出不通过的题才升到团队，成本能降一大截，准确率损失很小。这是投入产出比最高的一步改动。

把复核换成更强的模型。Reviewer 只做判断不做生成，调用次数少，换成更强的模型性价比通常优于把 5 个 Solver 全升级。

加一轮执行反馈修复。把沙箱的 stderr 喂回给选中的候选做一次定点修复，这一轮对语法错误和小边界错的救回率很高，对算法错的题帮助有限。

做候选去重。5×k 个候选里往往有大段重复，先用行为指纹去重再投票，能减少沙箱执行时间，也就是压低延迟。

扩到别的数据集。同一套三层结构换成别的竞赛编程或代码修复数据集，验证它是通用结构而不是对某个数据集过拟合。

沉淀配置与 trace。把温度、k、并发上限、一致性阈值写成一份 YAML，连同每题的 trace 一起归档。团队方案参数多，不记录就没法做消融。

最后提醒一句：这类系统最容易犯的错是先搭结构再想指标。正确顺序是先把对照 runner 和评测口径做扎实，再去调角色和阈值。没有可靠的测量，任何「团队更强」的说法都只是感觉。