Spaces:
Sleeping
交易系统重构手册(Refactor Handbook)
定位:这不是一次性重构方案,而是长期操作手册。任何未来的代码改动(bug 修复 / 加新策略 / 调参 / 换模型 / 新对手模式)都应该沿本手册操作,产出符合规范的代码 + 自测 + 部署 + 观察闭环。
读者:你自己、本地 agent、任何未来接手的人。
配套文档:
20260713_strategy_inventory.md— 现存 136 条策略的资产台账谁是卧底-大厂操盘手竞赛规则.md— 游戏规则(唯一权威)config/strategy_<profile>.yaml— 所有可调参数(本手册产出的目录)版本约定:本手册自身也用 semver:本次为
v1.0.0。修改架构规则算 major,加分类/规范算 minor,勘误算 patch。改本手册必须同时改CHANGELOG段。
目录
- 第 1 章 — 使用方式
- 第 2 章 — 目标架构:五层与数据流
- 第 3 章 — 每层职责边界(Rules of Layer)
- 第 4 章 — 变更分类决策树
- 第 5 章 — 各类变更的操作规范
- 第 6 章 — 通用编码规范
- 第 7 章 — 自测规范
- 第 8 章 — 部署规范
- 第 9 章 — 观察规范
- 第 10 章 — 常见坑与冲突协议
- 第 11 章 — 术语与命名约定
- 第 12 章 — PR 变更 Checklist(贴在每次 PR 里)
- CHANGELOG
第 1 章 — 使用方式
1.1 什么时候读本手册
必读:
- 拿到"加一个策略""改一个策略""改一个参数""修一个 bug"的任何需求时
- 部署一版新代码之前
- 部署之后要看"改动到底生效了没"时
- 出现线上事故要回滚时
建议读:
- 加入项目后,第 2、3 章一次读通
- 每次改本手册后(跟随 CHANGELOG)
1.2 怎么用
标准流程(黄金四步):
需求 ──► 第 4 章分类决策树,判定属于 5.1–5.8 哪一类
──► 打开对应小节,照规范改
──► 走第 7 章自测(unit + feature + replay 三层)
──► 走第 8 章部署(canary → 观察 → 全量或回滚)
不要跳步:手册里每一条规范都是有历史事故垫底的(大部分能追溯到 STRATEGY_V*.md 里一次具体的翻车)。跳步 = 重复历史事故。
1.3 手册与代码的关系
- 代码是执行体,手册是契约。冲突时以手册为准,代码要改
- 除非改的是 bug 修正,否则不允许"代码这么写因为历史原因"这类理由
- 手册规定"必须"的事情,本地 agent 必须做;"建议"的事情可以延后但要 TODO 标注
1.4 兼容原则(重要)
本地代码持续被 agent 修改,本手册可能落后于代码状态。**手册规定的是"结构和流程",不是"具体行数"**。
- 手册里的目录结构和文件名是契约,代码里必须存在
- 手册里的行号从不精确引用——只用"某函数""某模块"等语义定位
- 新加的策略不需要改手册;新的策略类型(比如"元策略")才需要
- 遇到手册说的东西代码里找不到,第一动作:向用户/上游确认是"代码欠改"还是"手册需更新"
第 2 章 — 目标架构:五层与数据流
2.1 五层总图
┌─────────────────────────────────────────────────────────────────┐
│ L5. Transport 层 (app.py + agent_shell.py) │
│ HTTP 入口 / 竞速网关 / FastAPI 端点 │
├─────────────────────────────────────────────────────────────────┤
│ L4. State 层 (state/) │
│ 唯一 Source of Truth (SoT) │
│ - GameState: dataclass, 不可变 │
│ - Reducer: (state, event) → new_state 纯函数, event sourcing│
│ - PendingLedger / Replay 持久化 │
├─────────────────────────────────────────────────────────────────┤
│ L3. Feature 层 (features/) │
│ 唯一负责"数值计算"的地方 │
│ - cluster_gain, opp_needs, card_value, pool_left… │
│ - 全是纯函数, 输入 GameState, 输出 dataclass, 100% 可测试 │
├─────────────────────────────────────────────────────────────────┤
│ L2. Strategy 层 (strategies/) │
│ 唯一负责"决策"的地方 │
│ - class Strategy: match(ctx)/act(ctx) 协议 │
│ - 每个策略一个文件, ≤120 行, 单一职责 │
│ - StrategyRouter 按 priority 串起来 │
├─────────────────────────────────────────────────────────────────┤
│ L1. LLM Adapter 层 (llm/) │
│ 唯一调用外部模型的地方 │
│ - LLMArbitrator.pick(): 从多个 HIT/UNSURE 里选一个策略 │
│ - SpeechGen.render(): 生成话术 message │
│ - VisionParser.parse(): 解析图片棋盘 │
└─────────────────────────────────────────────────────────────────┘
核心不变量(Invariants):
- 上层依赖下层,下层永远不知道上层。L2 引用 L3、L4;L3 引用 L4;L4 不引用任何上层
- L5 只做 IO;L4 只做状态管理;L3 只做计算;L2 只做决策;L1 只做外部调用
- State 只在 perceive 时变更;interact 期间 State 冻结(快照)
- 每一层可独立替换:换 LLM provider 只改 L1;换定价公式只改 L3;加策略只加 L2
2.2 目录结构(契约)
werewolftown/
├── __init__.py
├── agent.py # WerewolfTownLLMAgent 主类 (L5)
├── transport/ # L5
│ ├── endpoints.py # FastAPI 端点定义
│ ├── race_gate.py # 并发竞速网关(async 版)
│ └── debug_views.py # /debug/* HTML 视图
├── state/ # L4
│ ├── game_state.py # GameState dataclass
│ ├── events.py # Event 类型 + 枚举
│ ├── reducer.py # 纯函数 (state, event) → state
│ ├── pending_ledger.py # pending/sold/cancelled 统一账本
│ └── persistence.py # replay 落盘 + 恢复
├── features/ # L3
│ ├── clusters.py # 集群 / 边际增益 / 满额分析
│ ├── card_value.py # card_value(state, card, purpose)
│ ├── plot_value.py # plot_value(state, plot, viewer)
│ ├── opponents.py # opp_needs / leader / afk / strong
│ ├── pool.py # pool_left / scarcity
│ └── pricing.py # snap_price / dead_zone / lambda
├── strategies/ # L2
│ ├── base.py # Strategy Protocol + DecisionContext + MatchResult
│ ├── router.py # StrategyRouter
│ ├── select_plots/ # 选地策略族
│ │ ├── enumerate_combos.py
│ │ └── ...
│ ├── plan_build/ # 建设策略族
│ │ └── greedy_plus_local_search.py
│ ├── propose_trade/ # 12+ 个发起策略
│ │ ├── _order.py # priority 声明
│ │ ├── r4_leader_skip.py
│ │ ├── buy_card_for_full_cluster.py
│ │ ├── buy_adjacent_plot.py
│ │ ├── buy_plot_combo.py
│ │ ├── ...
│ │ └── skip.py
│ └── evaluate_trade/ # 7+ 个评估策略
│ ├── _order.py
│ ├── multi_deal_pick_best.py
│ ├── reject_no_empty.py
│ └── ...
├── llm/ # L1
│ ├── client.py # OpenAI 兼容客户端
│ ├── arbitrator.py # 从多候选选一个
│ ├── speech.py # 话术生成
│ └── vision.py # 视觉解析
├── constants/ # 静态数据
│ ├── board_grid.py # 52 工位邻接表(不可变)
│ ├── shop_rules.py # FULL_COUNTS / CARD_POOL / REVENUE_TABLE
│ └── phases.py # GamePhase 枚举
├── sanitize.py # 玩家名/message/JSON 过滤
config/
├── strategy_stable.yaml # 生产参数(默认)
├── strategy_canary.yaml # 灰度参数
└── strategy_dev.yaml # 本地实验
replays/
└── <game_id>/
├── events.jsonl # perceive 事件流
├── decisions.jsonl # interact 决策日志
└── meta.json # 对局元信息
tests/
├── unit/ # 纯函数单元测试
├── features/ # Feature 层测试
├── strategies/ # 策略层测试
└── replay/ # 完整对局回放测试
2.3 数据流总图
perceive(req) interact(req)
│ │
▼ ▼
┌─────────────┐ ┌───────────────┐
│ L5 shell │ │ L5 shell │
│ - dedupe │ │ - race_gate │
│ - persist │ │ - snapshot │
└──────┬──────┘ └───────┬───────┘
│ Event │ ctx
▼ │
┌─────────────┐ │
│ L4 reducer │ │
│ 纯函数 │ │
│ state < │◄─── (state, event) → new_state ──┤
│ append │ │
└──────┬──────┘ │
│ new_state │
▼ ▼
┌────────────────────────────────────────────────┐
│ L3 features (纯函数, on demand) │
│ 从 state 派生 cluster / opp_needs / pricing… │
└────────────────────┬───────────────────────────┘
│ DecisionContext (frozen)
▼
┌─────────────┐
│ L2 router │ 按 priority 遍历策略
│ │ ┌── all MISS → default
│ strategy │──┤── 1 HIT → strategy.act()
│ match/act │ └── multi UNSURE → LLM 仲裁
└──────┬──────┘
│ Action
▼
┌────────────┐
│ L1 speech │ 只填 message
│ (可选) │
└──────┬─────┘
│ AgentResp
▼
HTTP response
关键性质:
- perceive 路径只写 state,不做决策
- interact 路径只读 state,state 保持冻结
- Feature 层是惰性计算——DecisionContext 只暴露 property/method,策略用到才算
- 一次 interact 内,同一个 ctx 被所有策略共享,保证一致性视图
第 3 章 — 每层职责边界(Rules of Layer)
每层都有 DO / DON'T / 判断标准。违反标准的 PR 直接打回。
3.1 L5 Transport 层
DO:
- 处理 HTTP 请求解析、响应封装
- 并发竞速(同一 cache_key 只跑一次真实逻辑,其它请求 await future)
- Replay 事件流落盘(perceive 事件一到就 append)
- 装配子系统:从 config 装载 State/Feature/Strategy/LLM,注入依赖
- 提供
/debug/*HTML 视图
DON'T:
- 做任何决策计算(去 L2)
- 直接读写 memory 变量(去 L4)
- 调 LLM(去 L1)
判断标准:
- 如果这个文件里出现"if round == X" 或 "if my_cash > Y" → 违规
- 如果这个文件里 import 了 L2/L1,是合法的(装配);但不能反过来
- 单文件应 ≤ 200 行
3.2 L4 State 层
DO:
- 定义
GameStatedataclass(不可变,frozen=True或用 pyrsistent) - 定义
Event类型枚举 + 每个事件的字段 - 实现
reduce(state: GameState, event: Event) -> GameState纯函数 - 实现
PendingLedger(合并旧pending_proposals/recent_sold_plots/sold_this_round/ft_cancel_*) - 实现
persistence.py:save_event(game_id, event)/load_events(game_id) -> list[Event]/rebuild_state(events) -> GameState - 提供
apply(event)便捷方法:包装 reducer + 落盘
DON'T:
- 引用任何 L3/L2/L1 模块
- 有任何"策略""建议""定价"字样的函数
- 直接调用
random、time.time()、os.getenv()(这些是副作用) - 允许"允许 memory 被外部改写"的接口(如
set_variable)
判断标准:
- 每个函数必须能这样测试:
assert reduce(state_before, event) == state_after - state 一定是 dataclass,字段命名明确(
cash: int、plots: frozenset[int]) - 不允许
int和str混用作为 key - 服务器重启后
rebuild_state(load_events(game_id))必须能 100% 还原
关键契约:
GameState必须包含:player_id、round、cash、plots、shops、built_on: dict[int, str]、other_players: dict[str, PlayerView]、pending: PendingLedger、initial_cashGameState.replace(**kwargs)返回新实例,不改原实例- 所有集合用
frozenset/tuple,杜绝共享可变引用
3.3 L3 Feature 层
DO:
- 纯函数:输入
GameState+ 参数,输出 dataclass 或原始类型 - 计算集群 / 满额进度 / 边际增益 / 稀缺度 / 领先者 / opp_needs
- 提供缓存装饰器
@feature_cache(同一 GameState 实例内缓存) - 命名规范:动词 + 名词,明确返回值语义
- ✅
calc_cluster_revenue(...),card_value_for_selfuse(...),pool_remaining(...) - ❌
analyze(),get_stuff(),compute()
- ✅
DON'T:
- 有任何 if/else 输出"接受""拒绝""跳过"(决策去 L2)
- 引用 L2/L1
- 调用
random(策略层需要随机化就让 ctx 传入 rng) - 修改传入的 state(必须返回新对象)
判断标准:
- 每个函数写完先想:**能不能一句话说清"输入 X 输出 Y"**?说不清就是没抽象好
- 每个函数必须能这样测试:
assert calc_cluster_gain(state1, plot=5, shop="Happy") == 30000 - 大部分函数应 ≤ 30 行;≥ 50 行说明需要拆
关键契约:
card_value(state, card, purpose: Literal["selfuse","sell","combo"]) -> int是统一入口,杜绝多头维护plot_value(state, plot, viewer: str) -> int从viewer的视角算价值Offerdataclass 用to_me/from_me命名字段,禁止传offer_*/demand_*4 元组
3.4 L2 Strategy 层
DO:
- 每个策略一个文件,实现
Strategy协议 - 通过
match(ctx) -> MatchResult声明"我要不要处理这个局面" - 通过
act(ctx) -> Action产出决策 _order.py里显式声明本 phase 下所有策略的优先级- 复杂策略可拆多个
_helper.py,但 helper 必须在同一策略目录下、且不跨策略引用
DON'T:
- 引用其他策略模块(跨策略共享逻辑 → 提到 L3 features 或 helper)
- 直接调用 LLM(走
ctx.request_llm_arbitration(...)) - 修改 ctx / state
- 有
import random(用ctx.rng) - 单文件 > 120 行(超了就拆)
判断标准:
- 一个策略的
match()应能一句话说清"什么时候触发" - 一个策略的
act()应能一句话说清"触发后返回什么" - 如果你在一个策略里写"如果 round==4 走 A、否则走 B"——那是两个策略,拆开
- 策略之间不允许知道彼此的存在(除了
_order.py声明优先级)
关键契约:
match()返回MatchResult.HIT/MatchResult.MISS/MatchResult.UNSUREact()只返回Actiondataclass;Action→AgentResp的转换在 L5- 每个策略必须暴露
NAME: str、PRIORITY: int、PHASE: GamePhase
3.5 L1 LLM Adapter 层
DO:
- 唯一的 OpenAI 客户端封装(文本 + 视觉分开)
LLMArbitrator.pick(ctx, candidates: list[Strategy]) -> StrategySpeechGen.render(action: Action) -> str(生成 message,禁止改 action 其它字段)VisionParser.parse(snapshot) -> BoardView- Prompt 模板集中在
llm/prompts/目录,一个用途一个文件
DON'T:
- 做任何数值计算 / 决策 / 定价
- 直接输出
AgentResp - Prompt 里塞对手 message(永远走 sanitize 层,且 L1 不主动读 message)
- 有超过 10 个 Prompt 模板(Prompt 爆炸 = 抽象没做好)
判断标准:
- LLM 的输出永远是 枚举(策略名 / 话术字符串 / JSON dict),不是自由文本
- LLM 输出必须走 pydantic 校验;失败 fallback 到确定性策略
- 单次 LLM 调用 ≤ 4s;超时兜底
关键契约:
LLMArbitrator.pick()返回值必须是candidates里的一员,否则 raise- LLM 失败不能让决策失败——所有调用点都有代码兜底
3.6 依赖方向速查表
| 层 | 可以 import 的 | 不可以 import 的 |
|---|---|---|
| L5 | 所有下层 | — |
| L4 | 只有 constants/ + sanitize.py |
L5, L3, L2, L1 |
| L3 | L4 + constants/ |
L5, L2, L1 |
| L2 | L3 + L4 + constants/ |
L5, L1(除 ctx.request_llm_*) |
| L1 | 只有 sanitize.py + constants/ |
L5, L4, L3, L2 |
跨层通信只能通过 DecisionContext / Action / Event 三种数据契约。
第 4 章 — 变更分类决策树
拿到一个需求(对局观察、bug 报告、想加的新策略),先走这棵树,落到 5.1–5.8 的一个小节。
需求进来
│
├─ 是"决策错了"(agent 做了/没做某件事,我觉得应该反过来)
│ │
│ ├─ 触发条件本身就没有对应策略 ────► 5.1 加一条新策略
│ ├─ 策略触发了但产出错 ─────────────► 5.2 修改已有策略
│ ├─ 策略逻辑对,但阈值/价格不对 ────► 5.3 调整数值参数
│ └─ 策略逻辑对,但传入的数值就是错的
│ │
│ ├─ 状态错(cash/plots/built_on 不对)─► 5.8 修改基建 / 状态层
│ └─ 派生指标错(cluster/lead/opp_needs)─► 5.5 加/修 Feature
│
├─ 是"代码崩了 / 超时 / 数据丢失" ─────────► 5.4 修 bug
│
├─ 是"游戏规则变了 / 后端消息类型变了" ────► 5.6 加一个新的 Event 类型
│
├─ 是"想让 LLM 参与更多 / 更少" ───────────► 5.7 修改 LLM 参与方式
│
└─ 是"想改架构本身 / 加新的一层" ──────────► 走本手册的第 2、3 章升级流程
4.1 分类判断口诀
- 加什么 → 5.1(策略)/ 5.5(Feature)/ 5.6(Event)
- 改什么 → 5.2(策略行为)/ 5.3(参数)/ 5.7(LLM 用法)
- 修什么 → 5.4(bug)/ 5.8(基建)
4.2 何时需要跨类
一次改动可能跨类:
| 场景 | 落地路径 |
|---|---|
| 新策略需要一个之前没算过的指标 | 先 5.5 加 Feature,再 5.1 加策略 |
| 修策略时发现下面的估值有 bug | 拆两个 PR:先 5.4 修 bug(含回归测试),再 5.2 改策略 |
| 加 Event 后要让某策略响应它 | 先 5.6 加 Event + reducer,再 5.5/5.1 更新 Feature/策略 |
原则:一个 PR 只做一件事。跨类改动分多个 PR,每个 PR 都过 checklist。
第 5 章 — 各类变更的操作规范
5.1 加一条新策略
5.1.1 什么时候用这条
- 观察到某个局面 agent 应该做 X 但代码里没有对应逻辑
- 想尝试一种全新的博弈套路(如"卡领先者""废卡换地")
- 冠军对局分析出的新战术
不适用:现有策略触发但产出错 → 走 5.2;现有策略触发条件太宽/太窄 → 走 5.2 而不是 5.1
5.1.2 判断放哪层
新策略永远在 L2 strategies/ 下。判断哪个子目录:
| Phase | 目录 | 触发时机 |
|---|---|---|
| SELECT_PLOTS | strategies/select_plots/ |
每回合选地 |
| PLAN_BUILD | strategies/plan_build/ |
每回合建设 |
| PROPOSE_TRADE | strategies/propose_trade/ |
每次 agent 主动发起交易 |
| EVALUATE_TRADE | strategies/evaluate_trade/ |
每次收到别人交易要约 |
5.1.3 代码规范
文件模板 strategies/propose_trade/buy_adjacent_plot.py:
"""BuyAdjacentPlot — R1-R3 无 R4 领先者时,主动买邻接工位扩集群。
来源:L4-11, L4-12, L4-14, L4-18, L4-20(inventory 里的编号,方便追溯)
"""
from __future__ import annotations
from ..base import Strategy, MatchResult, Action
from ...features.plot_value import find_best_buy_candidate
from ...features.opponents import is_me_leader
from ...features.pricing import buy_discount, max_buy, snap_price
NAME = "buy_adjacent_plot"
PHASE = "propose_trade"
PRIORITY = 30 # 见 _order.py 中的编号规则
def match(ctx) -> MatchResult:
if ctx.round == 4 and is_me_leader(ctx.state):
return MatchResult.MISS
if not ctx.is_directed:
return MatchResult.MISS
cand = find_best_buy_candidate(ctx.state)
if cand is None:
return MatchResult.MISS
if cand.net < 0:
return MatchResult.MISS
return MatchResult.HIT
def act(ctx) -> Action:
cand = find_best_buy_candidate(ctx.state) # feature 层缓存, 不重算
price = snap_price(min(
int(cand.gain * ctx.cfg.buy_discount[ctx.round]),
ctx.cfg.max_buy[ctx.round],
))
return Action.initiate(
target=cand.owner,
offer_cash=price,
demand_plots=[cand.plot_id],
speech_hint="buy_plot", # SpeechGen 会用此 hint 生成 message
)
必须:
NAME/PHASE/PRIORITY三个模块级常量match(ctx)只读 ctx,不改act(ctx)只调 features 和 config,不直接算复杂数值- 所有随机化用
ctx.rng - 所有数值参数从
ctx.cfg.<name>读,禁止硬编码 - 顶部注释写清"来源"(inventory 编号或者对局出处),便于追溯
禁止:
- 在策略里 import 其它策略
- 在策略里
time.time()/random.random()/os.getenv() - 单文件超过 120 行(拆 helper 到同目录)
5.1.4 注册到 Router
打开 strategies/propose_trade/_order.py:
from . import (
r4_leader_skip,
buy_card_for_full_cluster,
buy_adjacent_plot, # ← 新加的
buy_plot_combo,
...
)
# 优先级从小到大,第一个 HIT 就返回
STRATEGIES_IN_ORDER = [
r4_leader_skip, # priority 10
buy_card_for_full_cluster, # priority 20
buy_adjacent_plot, # priority 30 ← 插到这里
buy_plot_combo, # priority 40
...
]
优先级编号规则:
- 每个策略的
PRIORITY是它在_order.py里的相对位置 × 10 - 插入新策略时,
PRIORITY也要更新 - 优先级冲突(同 phase 两个策略 PRIORITY 相同)→ CI 会报错
5.1.5 必配的三件套
必须同时提交:
- 策略代码(本节主内容)
- 参数(如果引入了新数字,配到
config/strategy_stable.yaml) - 测试:
tests/strategies/test_buy_adjacent_plot.py:至少 3 个用例(HIT / MISS 边界 / act 输出)tests/replay/test_buy_adjacent_plot_replay.py:至少 1 个真实对局回放证明这条策略被走到
5.1.6 自测
# unit
pytest tests/strategies/test_buy_adjacent_plot.py -v
# 全策略层
pytest tests/strategies/ -v
# 从一场历史 replay 回放,检查这条策略被触发过
python -m werewolftown.tools.replay replays/<game_id> \
--assert-strategy buy_adjacent_plot
5.1.7 PR checklist(本节专属,通用见第 12 章)
- 单文件 ≤ 120 行
-
NAME/PHASE/PRIORITY三常量齐全 -
_order.py已注册 - 硬编码数字全部挪到
strategy_stable.yaml -
tests/strategies/新增了针对性单测(HIT/MISS/act) -
tests/replay/至少一个真实对局回放证明触发过 - 顶部注释写清"来源"(inventory 编号或对局 game_id)
5.2 修改已有策略
5.2.1 什么时候用这条
- 策略被触发了,但产出的 Action 不对(价格错、目标错、方向错)
- 策略触发条件太宽(不该触发时触发)或太窄(该触发时没触发)
- 策略里某段逻辑与 Feature 层重复,可以下沉
5.2.2 判断影响面
第一步:读该策略的顶部注释。找到"来源"里的 inventory 编号,看历史上这条策略是为什么加进来的。
第二步:查 _order.py,看这条策略的邻居(前一个和后一个 priority)。改这条策略可能影响:
- 更高优先级的策略 → 通常不影响(更高优先级的先 return 了)
- 更低优先级的策略 → 可能影响(本策略从 MISS 改 HIT 会抢走它们的机会;反之亦然)
第三步:如果条件收紧(HIT → MISS 变多),下一个策略要能接住;如果条件放宽(MISS → HIT 变多),要确认没抢占后续更合适的策略。
5.2.3 代码规范
只允许 3 种改动:
| 改动 | 允许 | 判断 |
|---|---|---|
A. 收紧/放宽 match() 条件 |
✅ | 只改 match,不改 act |
B. 改 act() 产出 |
✅ | 只改 act,不改 match |
| C. 拆分/合并策略 | ✅ | 必须新加/删文件,不允许原地扩容 |
禁止:
- 在原有策略里加"如果 X 走 A、否则走 B" → 拆成两个策略
- 让 match 有副作用(改 state、调 LLM、动 rng)
- 把参数从 config 挪回硬编码
5.2.4 兼容性检查
改一条策略前必须做:
# 1. 找出所有回放里被这条策略处理过的局面
python -m werewolftown.tools.replay_scan \
--strategy buy_adjacent_plot \
--output before.csv
# 2. 改完策略后再跑一次
python -m werewolftown.tools.replay_scan \
--strategy buy_adjacent_plot \
--output after.csv
# 3. diff 两份结果
python -m werewolftown.tools.replay_diff before.csv after.csv
Diff 里会显示:
- 仍然 HIT 的局面(占多数是好事,说明改动没颠覆本策略)
- HIT → MISS(本次改动让哪些局面不再触发)
- MISS → HIT(哪些新增局面被这条策略接管)
- HIT 后 Action 变化(价格从 X 变 Y)
diff 结果需要作为 PR 描述的一部分,并解释每类变化是否符合预期。
5.2.5 参数与逻辑的边界(关键)
改策略前先问:这个改动真的是逻辑变化,还是只是参数变化?
- "R1 买地折扣从 0.5 改到 0.55" → 是 5.3 调整数值参数,不是 5.2
- "R1 除了买邻接地还要考虑跨街区候选" → 是 5.2(改 match/act 逻辑)
- "把 R4 领先者不买地改成也可以买地" → 拆策略,走 5.1(新加一个
R4LeaderConservativeBuy)
如果 60% 以上的改动是数字,先走 5.3。
5.2.6 自测
# 该策略单测
pytest tests/strategies/test_<strategy_name>.py -v
# 若改动可能影响相邻策略,跑该 phase 全套
pytest tests/strategies/propose_trade/ -v
# 回放差异分析
python -m werewolftown.tools.replay_diff before.csv after.csv --min-diff 5%
5.2.7 PR checklist
- 顶部注释里补一条"改动记录:
<date> - <理由>" - 若加了新
match分支,_order.py里检查过邻居 - 若引入新数字,参数已配到
strategy_stable.yaml - 回放差异分析(before.csv vs after.csv)附在 PR 描述中
- 单测覆盖新增分支
- 单文件仍 ≤ 120 行
5.3 调整数值参数
5.3.1 什么时候用这条
- 策略逻辑对,但阈值/价格偏高偏低
- 想微调"更保守"或"更激进"
- 冠军对局分析出的具体数字更合理
不适用:如果调数字会导致策略"是否触发"发生变化(例如把 accept_threshold 从 5000 改到 -5000)——这算逻辑变化,走 5.2
5.3.2 参数目录结构
所有可调数字都在 config/ 下:
# config/strategy_stable.yaml — 生产环境
version: "20260713"
description: "V18-Fix5 之后的稳定版"
pricing:
buy_discount:
R1: 0.50
R2: 0.45
R3: 0.35
R4: 0.35
max_buy:
R1: 25000
R2: 20000
R3: 15000
R4: 10000
price_snap_step: 500
card_value:
base_price_by_fc:
2: 30000
3: 20000
4: 12000
5: 8000
market_val_by_fc:
2: 15000
3: 15000
4: 12000
5: 10000
scarcity_lambda:
pool_le_1: 0.55
pool_le_2: 0.40
R4: 0.50
fc_le_2: 0.35
default: 0.25
threshold:
accept:
R1: [12000, 18000] # uniform range
R2: [12000, 18000]
R3: [8000, 12000]
R4: 0 # 定值
reject:
R1: [-10000, -6000]
R2: [-10000, -6000]
R3: [-8000, -4000]
R4: 0
# ... 见附录 A 数值参数索引
必须:
- 每个参数都要有语义化名字,不能是
param1x - 分组按功能:
pricing//card_value//threshold//build//select/ - 每个大分组顶部一句话说明用途
5.3.3 修改流程
先改 canary,不改 stable:
# 1. copy stable → canary(如果 canary 落后了)
cp config/strategy_stable.yaml config/strategy_canary.yaml
# 2. 只改 canary 的对应参数
vim config/strategy_canary.yaml
# 3. 加一条 CHANGELOG 到 canary 顶部
参数改动的 CHANGELOG 格式:
changelog:
- date: "20260714"
field: "pricing.buy_discount.R1"
from: 0.50
to: 0.55
reason: "对局 358890 中 R1 买地净值 20k 但我们只出 15k 被拒,测试 0.55 提升成交率"
author: "user"
改完 canary,走部署(见第 8 章)灰度 → 观察 → 全量。
5.3.4 判断能不能只改参数
试探:把新参数值填进现有测试:
def test_accept_threshold_r1():
ctx = build_ctx(round=1, cash=50000, offer_net=10000)
# 用 canary 参数
with load_config("strategy_canary.yaml"):
assert evaluate(ctx).action == "ACCEPT"
# 用 stable 参数
with load_config("strategy_stable.yaml"):
assert evaluate(ctx).action == "REJECT"
如果这样的对照测试写得出来 → 说明确实只是参数变化,走 5.3; 如果对照测试要求"新代码分支"才能通过 → 说明是逻辑变化,走 5.2
5.3.5 自测
# 用 canary 参数跑一遍策略层测试
STRATEGY_PROFILE=canary pytest tests/strategies/ -v
# 用 canary 跑历史回放,看总收益趋势
python -m werewolftown.tools.replay_batch \
--profile canary \
--games replays/last_20/ \
--report canary_vs_stable.md
canary_vs_stable.md 会输出:
- 每局改动前后 agent 总现金差
- 平均差值
- 分位数(P25 / P50 / P75)
只有平均差 > 0 且 P25 也不劣化,才能走灰度部署。
5.3.6 PR checklist
- 只改了
config/strategy_canary.yaml,没动代码 -
changelog:段有条目:字段/from/to/reason/author - 20 场历史 replay 的 canary vs stable 报告附在 PR 描述里
- 平均总现金差 > 0,且 P25 分位数不劣化
5.4 修 bug(正确性问题)
5.4.1 什么时候用这条
- 代码抛异常 / 超时
- 数据错(state 里的 cash/plots/built_on 与实际不符)
- 策略应该 HIT 但没 HIT(match 里有笔误、import 错)
- 明显的算术错(净值 -245000 那种量级偏差)
5.4.2 修 bug 的黄金流程
第一步:先复现,不要先改
# 1. 找到出问题的 game_id(HF 日志、用户报告、监控告警都能给)
GAME_ID=358890
# 2. 从 replays/<game_id>/ 拉出事件流和决策日志
python -m werewolftown.tools.replay replays/$GAME_ID --verbose
# 3. 定位到出错的那次 interact 决策
# 输出会打印: [round=2 phase=propose_trade] strategy=buy_adjacent_plot
# elapsed=45s, action=SKIP, exception=NameError('remaining')
如果 replay 无法复现,先补 replay 采集:这是基建 bug(去 5.8),不是策略 bug。
第二步:写失败测试
# tests/strategies/test_buy_adjacent_plot.py
def test_bug_20260713_undefined_remaining():
"""Bug: 2026-07-13 room-358890 R2 propose_trade 抛 NameError('remaining')."""
ctx = load_from_replay("replays/358890/", round=2, phase="propose_trade")
# 期望:至少不能崩,返回 SKIP 或有效 Action
result = strategy.act(ctx)
assert result is not None
assert result.action in {"initiate", "skip"}
这条测试必须先失败,才能证明 bug 存在。
第三步:修复
修复只做最小改动——只让上面那条测试通过,不做重构、不做优化。
第四步:证明修复不引入回归
pytest tests/strategies/ tests/features/ tests/unit/ -v
python -m werewolftown.tools.replay_batch replays/last_20/ --profile stable_with_fix
replay batch 里所有 game 的最终 agent 现金都应 >= 修前值(允许微调)。
5.4.3 常见 bug 分类与定位
| 症状 | 常见原因 | 定位起点 |
|---|---|---|
| NameError / UnboundLocalError | 函数内 import 或变量未定义 | 该函数模块级 import |
| 数据源不对 | State 层重放丢事件 | state/reducer.py, state/persistence.py |
| 决策超时 | LLM 兜底 + LLM 超时 | 检查是否走到了 L1;策略 match 是否覆盖到该场景 |
| 视角反转(net 反了号) | Offer dataclass 用错方向 | 该策略的 act() 里的 Offer 构造 |
| 循环 import / lazy import 藏 bug | 函数体内 import | 移到模块级;跑 mypy/ruff 应能扫出 |
| int/str key 不一致 | State 里 built_on 序列化问题 | state/game_state.py 的类型注解 |
5.4.4 修 bug 的红线
- 禁止通过
try/except: return SKIP掩盖 bug(旧代码的巨坑,见 inventory 第 2.2 节 P1-5) - 禁止通过"再加一个 if"打补丁到多层——如果某个 bug 需要跨层修,先谈重构(见 5.8)
- 必须在 PR 描述里写清"这个 bug 是被隐藏了多久(前一次涉及此代码的改动日期到发现日期)",帮助我们判断监控是否有盲区
5.4.5 PR checklist
-
tests/有一条以 game_id 命名的回归测试(test_bug_YYYYMMDD_<short_desc>) - 该测试在修前失败、修后通过
- 修改范围最小化(不夹带重构)
- 全量测试通过 + last 20 replay batch 无回归
- PR 描述里包含:"bug 首次可能出现时间 / 被发现时间 / 潜伏原因"
5.5 加一个 Feature(衍生指标)
5.5.1 什么时候用这条
- 你正要写的一个策略需要一个之前没算过的指标
- 有两个以上策略都要用到同一个数字,且这个数字目前散落在各处(临时算或复制粘贴)
- L2 策略层里出现"import 另一个策略的 helper"(禁止 → 抽 L3)
不适用:如果这个指标已经在 features/ 里存在,不要重写;如果只在一处用到,也可以先留在 helper 里,等第二次用到再抽
5.5.2 判断放哪个模块
现有六个功能包:
| 模块 | 职责 | 示例 |
|---|---|---|
features/clusters.py |
集群相关 | calc_cluster_gain, find_typed_clusters, achievable_size |
features/card_value.py |
卡估值 | card_value(state, card, purpose), card_market_val |
features/plot_value.py |
工位估值 | plot_value(state, plot, viewer), find_best_buy_candidate |
features/opponents.py |
对手分析 | opp_needs, is_leader, detect_afk, detect_strong |
features/pool.py |
牌池 | pool_remaining, scarcity_lambda |
features/pricing.py |
定价工具 | snap_price, dead_zone_adjust, buy_discount, max_buy |
判断口诀:新 feature 的输入主体是什么?
- 输入是 (state, plot) →
plot_value.py - 输入是 (state, card) →
card_value.py - 输入是 (state, opponent_pid) →
opponents.py - 输入是 (state,) 且返回集群统计 →
clusters.py - 输入是 (state,) 且返回稀缺度 →
pool.py - 是数值变换工具 →
pricing.py
不满足以上,可新开模块,但必须在本手册第 2.2 节目录树里追加。
5.5.3 代码规范
# features/plot_value.py
from dataclasses import dataclass
from typing import Literal
from ..state.game_state import GameState
from ..constants.shop_rules import FULL_COUNTS
Viewer = Literal["me", "opponent"]
@dataclass(frozen=True)
class BuyCandidate:
"""一个可买入的工位候选,附带估值细节。"""
plot_id: int
owner: str
gain: int # 净收益(未减价)
best_shop: str # 若买入应该建哪种业务
is_adjacent: bool # 是否邻接我方工位
net: int # gain - suggested_price
def plot_value(state: GameState, plot_id: int, viewer: Viewer = "me") -> int:
"""从 viewer 视角计算一个工位的估值。
- viewer="me": 我方买入该地能获得的最大边际增益
- viewer="opponent": 对手买入该地能获得的最大边际增益 (卖地时用)
"""
...
必须:
- 每个函数第一行 docstring 说清"输入 / 输出 / 语义"
- 返回值是 dataclass(
frozen=True)或原始类型,不返回 dict - 输入 state 不能被修改(拿 frozen dataclass 就免这条)
- 加类型注解,
mypy --strict能过 - 遵循命名规范:
calc_*计算类,find_*查找类,is_*/has_*布尔,<name>_value估值
禁止:
- 有
if round == 4: skip_this()之类的决策逻辑(决策去 L2) - 依赖
random - 修改全局变量
- 单函数 > 50 行
5.5.4 缓存
大量 feature 会被同一个决策周期里多次调用,避免重算:
from ..features._cache import feature_cache
@feature_cache
def find_best_buy_candidate(state: GameState) -> BuyCandidate | None:
"""在 (state, {}) 上缓存: 同一 state 实例内不重算。"""
...
feature_cache 用 id(state) 做 key,state 一变缓存失效(state 不可变,一旦有事件到达就产生新 state 实例)。
5.5.5 自测
pytest tests/features/test_plot_value.py -v
测试规范:
- 每个 feature 至少 3 个 case(正常 / 边界 / 空输入)
- 用
tests/features/fixtures.py里的make_state(...)便捷构造 state - 断言必须给具体数字(不是
> 0这种模糊断言),因为策略层依赖具体值
def test_plot_value_adjacent_full_cluster():
state = make_state(
my_plots=[4, 5], built_on={4: "Happy", 5: "Happy"}, shops=["Happy"],
)
# plot=6 邻接 4,5 (Happy 2 连) + 手牌 Happy → 满额 3 连
v = plot_value(state, 6, viewer="me")
assert v == 70000 - 30000 # 3 连满额 70k - 之前 2 连 30k
5.5.6 PR checklist
- Feature 放在合适的模块(按 5.5.2 判断)
- 输入 state,输出 dataclass/原始类型,无副作用
- docstring 说清输入输出语义
- 类型注解完整
-
tests/features/test_<module>.py覆盖 ≥ 3 个 case - 若加了
@feature_cache,验证过 state 变更后缓存失效
5.6 加一个新的 Event 类型
5.6.1 什么时候用这条
- 游戏引擎新增了 perceive 消息类型(如新的 phase 事件)
- 某种局面 State 层需要新的字段来追踪(如新增 opp_needs 缓存)
- 后端字段格式变化(比如
offer_shops从字符串改成数组)
不适用:如果只是把已有事件更完整地记录到 state,这是 5.8 的活;如果只是让某策略响应已有事件,这是 5.1/5.2
5.6.2 五步走
Step 1: 在 state/events.py 定义事件类型
from dataclasses import dataclass
from typing import Literal
@dataclass(frozen=True)
class DealCountered:
"""引擎推送的还价事件。"""
kind: Literal["deal_countered"] = "deal_countered"
game_id: str
round: int
deal_id: str
source: str # 还价方
target: str # 被还价方
offer_cash: int
offer_plots: tuple[int, ...]
offer_shops: tuple[str, ...]
demand_cash: int
demand_plots: tuple[int, ...]
demand_shops: tuple[str, ...]
ts: float # 事件到达时间戳
Step 2: 在 state/reducer.py 加 case
def reduce(state: GameState, event: Event) -> GameState:
match event:
case DealProposed(...): return _reduce_deal_proposed(state, event)
case DealCountered(...): return _reduce_deal_countered(state, event) # ← 新加
...
Step 3: 在 transport/endpoints.py 加解析
def _req_to_event(req: AgentReq) -> Event:
match req.status.value:
case "deal_proposed": return _parse_deal_proposed(req)
case "deal_countered": return _parse_deal_countered(req) # ← 新加
...
Step 4: 加 replay 兼容
Replay 落盘的 JSON 里加新事件类型,load_events 要能反序列化:
# state/persistence.py
_EVENT_TYPES = {
"deal_proposed": DealProposed,
"deal_countered": DealCountered, # ← 新加
...
}
Step 5: 更新受影响的 features
如果新事件带来的 state 变化会影响某些 feature 的计算,那些 feature 的缓存要考虑到(通常 state 变化自动使缓存失效,但要显式测试)。
5.6.3 事件的兼容性红线
- 旧 replay 必须能加载:新加事件类型不能让旧的 events.jsonl 里的记录反序列化失败
- 事件字段只加不删:如果要重命名字段,通过 property 别名兼容
- event schema 版本化:
state/events.py顶部维护SCHEMA_VERSION,改动时递增 - replay 里的事件带 schema_version:
load_events遇到高版本要提示升级
5.6.4 自测
# 1. 单测 reducer
pytest tests/state/test_reducer_deal_countered.py -v
# 2. 用带新事件的模拟 replay 验证 rebuild_state 无回归
pytest tests/state/test_persistence.py -v
# 3. 用旧的 replay 验证兼容性
pytest tests/state/test_replay_backward_compat.py -v
5.6.5 PR checklist
-
state/events.py定义了新事件类 -
state/reducer.py加了对应 case,纯函数 -
transport/endpoints.py加了 req → event 解析 -
state/persistence.py的_EVENT_TYPES注册了新类型 -
tests/state/有新事件的 reducer 测试 + replay 序列化/反序列化测试 - 旧 replay 数据仍能正常加载(backward-compat 测试)
- 若字段命名有变,写了别名保持兼容
5.7 修改 LLM 参与方式
5.7.1 什么时候用这条
- 想让某个策略的某个参数由 LLM 填("模式 3:LLM 填参数")
- 想让 LLM 在多个候选策略间仲裁("模式 2:LLM 仲裁")
- 想把某个已经稳定的 LLM 调用点固化为纯代码("从模式 2/3 降到模式 1")
- 想改话术生成规则
5.7.2 四种 LLM 参与模式(关键概念)
| 模式 | 描述 | 用途 | 代码入口 |
|---|---|---|---|
| M1 纯代码 | LLM 完全不参与决策 | 成熟策略 | 策略 act() 直接返回 Action |
| M2 LLM 仲裁 | 多个策略 HIT/UNSURE,LLM 从候选里挑一个 | 策略打架时 | ctx.request_llm_arbitration(candidates) |
| M3 LLM 填参 | 策略返回 partial Action,LLM 填空 | 参数不好定 | Action.partial(price=None, ...) + SpeechGen.fill() |
| M4 LLM 全接管 | 所有策略 MISS,LLM 自由决策 | 未知场景兜底 | LLMFullDecide.decide(ctx) |
演化方向:M4 → M3 → M2 → M1,越往右越确定,越可控。
5.7.3 何时升/降模式
从 M4 降到 M3(LLM 全接管 → LLM 只填参数):
- 观察到某场景 LLM 决策大方向稳定,但价格上下 30%
- 加一条策略:
match就是 M4 的触发条件,act返回Action.partial(price=None) - LLM 只填 price,逻辑固化在策略里
从 M3 降到 M2(LLM 填参 → LLM 只仲裁):
- 观察到某参数在两三个离散值之间跳(比如 "10k 或 15k")
- 拆成 2-3 个子策略(分别对应不同参数值),全部 UNSURE
- LLM 从这几个里挑一个
从 M2 降到 M1(LLM 仲裁 → 纯代码):
- 观察到 LLM 仲裁结果稳定倾向某一策略
- 把仲裁改为硬性优先级
- 或者把两个候选策略合并为一个,删掉冲突
M1 → M2/M3 升级(新场景无法完全代码化时):
- 加一个策略,match 返回 UNSURE 而不是 HIT
- 让 LLM 仲裁参与
5.7.4 加/改 LLM 调用点的规范
只有 3 个合法调用点(全部在 L1):
LLMArbitrator.pick(ctx, candidates)— 从多候选选一个策略名SpeechGen.render(action, hint)— 生成 messageVisionParser.parse(snapshot)— 解析图片
新增第 4 个调用点必须先改本手册第 3.5 节。
调用点必须满足:
# llm/arbitrator.py
class LLMArbitrator:
def pick(self, ctx: DecisionContext, candidates: list[Strategy]) -> Strategy:
# 1. 输出必须是候选之一
# 2. 输出用 pydantic 模型强约束
# 3. 超时/失败必须有代码 fallback
try:
resp = self._call_with_timeout(...)
picked = self._parse(resp, candidates)
return picked
except (TimeoutError, ValidationError):
# fallback: 用 priority 最高的候选
return sorted(candidates, key=lambda s: s.PRIORITY)[0]
Prompt 必须:
- 不含对手 message(sanitize 层已经过滤,但 L1 也不主动读)
- 用结构化格式(system 是稳定契约,user 是当前局面)
- 用 pydantic 定义输出 schema:
class ArbitrationResp(BaseModel): picked: str
5.7.5 话术(SpeechGen)规范
话术是当前 LLM 最重要的用途。规则:
- 每个 Action 携带
speech_hint(枚举:"buy_plot","sell_card","counter_up","counter_down","reject","skip", ...) SpeechGen.render(action, hint)根据 hint 选 prompt 模板生成 message- 生成失败 fallback 到固定短话术(见 inventory L4-33, L6-09)
- 话术生成不影响 Action 的其它字段——只填
message字段
5.7.6 自测
# LLM 层单测(用 mocked client)
pytest tests/llm/ -v
# 端到端:某场景应该走 M3
pytest tests/replay/test_llm_participation.py::test_M3_price_filled -v
关键测试:LLM 超时 / 返回 gibberish / 返回不在候选里的值 → 都要能兜底不崩
5.7.7 PR checklist
- 明确本次改动是模式升级还是降级(M2↔M3 等),写在 PR 描述里
- 若新增调用点,本手册第 3.5 节同步更新
- Prompt 通过 pydantic schema 约束输出
- LLM 失败 fallback 是纯代码
- 每次 LLM 调用有超时保护(≤ 4s)
- 单测覆盖:正常返回、超时、格式错误、无效值 四种
5.8 修改基建 / 状态层
5.8.1 什么时候用这条
- Reducer 有 bug(应用某个事件后 state 与实际不符)
- Replay 无法完整恢复(服务器重启后状态残缺)
- 并发竞速网关行为异常
- 想改 State 的数据结构(比如把
plots: list改成frozenset)
警告:这层改动风险最大。任何 L4 的改动都可能牵一发动全身——上面所有 L3 features 都依赖 State 的字段名和语义。动 L4 前必须先谈(本手册也算"上级",需要更新第 3.2 节和第 2.2 节)。
5.8.2 判断改动范围
三个层次,越往下越危险:
| 层次 | 例子 | 危险度 |
|---|---|---|
| A. 只改 reducer 里某个 case 的实现 | 修 deal_accepted 的资产转移 bug |
中 |
| B. 加/删 State 字段 | 加一个 pending: PendingLedger |
高 |
| C. 改 State 序列化格式 | built_on 从 dict 改成 tuple |
极高 |
A 直接走本节流程;B 要更新第 3.2 节;C 要更新第 2.2 节 + backward-compat 迁移
5.8.3 A 类改动流程(修 reducer bug)
- 找到那个 case(比如
_reduce_deal_accepted) - 写失败测试:用 replay 里的一段
(state_before, event, expected_state_after) - 改 reducer,只让这个测试通过
- 跑所有 reducer 单测 + 全 replay 重放:任何一场 replay 重放出来的 state 与录制时不一致 → 回滚
5.8.4 B 类改动流程(加/删 State 字段)
加字段是安全的(默认值兼容旧数据):
@dataclass(frozen=True)
class GameState:
# ... 现有字段
afk_scores: dict[str, int] = field(default_factory=dict) # 新加
Reducer 里在相应事件更新这个字段。旧 replay 加载后该字段是空 dict,不影响策略——策略应该能处理"空字段"作为初始状态。
删字段绝对禁止:改为标注 @deprecated,保留字段但不再写入。等所有依赖它的 feature/策略都不用了,才能真正删。删除的 PR 单独提。
5.8.5 C 类改动流程(改序列化格式)
这是最麻烦的场景。必须走:
- 加新字段并双写:新旧格式同时保留(比如
built_on_dict和built_on_tuple) - 等一个上线周期:观察新格式所有 case 都正常
- 切换 features/策略到新字段:一个 PR 切一个消费方
- 观察一个上线周期:老字段没人读了
- 删老字段:单独 PR
旧 replay 的兼容:persistence.py 的 load_events 需要探测 schema_version,用旧解析器读,然后 reducer 无缝升级为新 state。
5.8.6 竞速网关(transport/race_gate.py)改动
因为你的补充说明第 5 点:后端不发显式重发标记,我们只能靠 cache_key 推断。任何这方面改动必须:
- 保留 cache_key 计算逻辑(
message字段仍要排除,以防对手在附言里做小改动) - 竞速用 async future(不再 threading.Event)
- 保留兜底:如果 2 个请求都超时了怎么办(返回上一次 known-good decision 或 SKIP)
测试用例必须包含:
- 同一 key 的两个请求并发到达
- 第一个请求超时、第二个请求正常返回
- 两个请求都异常
- 请求 key 相同但 message 字段不同 → 仍然合并(因为 cache_key 排除了 message)
5.8.7 自测
# 单测 reducer
pytest tests/state/ -v
# 完整回放:所有 replays/ 里的对局都能无损重放
python -m werewolftown.tools.replay_all_check --strict
# 并发压测(如果改了 race_gate)
python -m werewolftown.tools.race_stress --n 100 --key-dup-rate 0.3
5.8.8 PR checklist
- 明确改动是 A/B/C 哪类(PR 描述前置声明)
- A 类:新增回归测试,全 replay 通过
- B 类:新字段有默认值,本手册第 3.2 节同步更新
- C 类:分成"加字段"→"切消费方"→"删旧字段"三个 PR
- 若涉及 race_gate:并发压测通过
- 若涉及 persistence:backward-compat 测试通过
第 6 章 — 通用编码规范
跨层、跨变更类型的共同底线。任何 PR 都要过。
6.1 Python 版本 & 依赖
- Python **3.11+**(
Dockerfile已升级;list[int]/str | None等语法可用) - 全
from __future__ import annotations - 类型注解 100%(
mypy --strict能过);不放任# type: ignore通过 CI
6.2 命名规范
| 类型 | 规则 | 例 |
|---|---|---|
| 模块 | snake_case | plot_value.py |
| 类 | PascalCase | BuyCandidate, Strategy |
| 函数 | snake_case,动词开头 | calc_cluster_gain, find_best_buy_candidate |
| 常量 | UPPER_SNAKE_CASE | FULL_COUNTS, CARD_POOL |
| 私有函数 | _snake_case |
_reduce_deal_proposed |
| 类型别名 | PascalCase | Viewer = Literal["me", "opponent"] |
语义前缀:
calc_*→ 计算数值find_*→ 查找候选is_*/has_*→ 返回 boolapply_*→ 有副作用(只允许在 L4 L5)render_*→ 生成字符串
6.3 import 规范
必须:
- 所有 import 放模块顶部,不允许函数内 import(历史巨坑,见第 10 章 P10-3)
- 顺序:标准库 → 第三方 → 本项目
- 一行一个 import
唯一例外:
if TYPE_CHECKING:块内的循环 import 破解- 明确的性能优化(罕见,且必须有注释解释)
禁止:
from ..foo import *- 循环 import(用
if TYPE_CHECKING:破除)
6.4 数据类规范
- 用
@dataclass(frozen=True)除非明确要求可变 - 用
frozenset/tuple代替set/list作为字段类型(避免共享可变引用) - 用
Enum/Literal代替字符串字面量
6.5 副作用管理
禁止在 L2/L3 出现:
import random/random.random()→ 用ctx.rngtime.time()→ 用ctx.nowos.getenv(...)→ 用ctx.cfg.<field>logger.info(...)用于记录决策 → 用ctx.decision_log
原因:这些副作用让代码无法用同一 state 稳定重放。
6.6 日志规范
必须结构化:
ctx.decision_log.emit(
strategy_name=NAME,
stage="match",
result="HIT",
features={"gain": cand.gain, "net": cand.net},
)
不再用 logger.warning("[eval] 代码兜底 ACCEPT: net=X") 这种字符串日志。观察平台需要结构化数据。
6.7 错误处理
只允许两种 except:
- 业务边界处(L5 endpoints):捕获所有异常,落成结构化 error_log,返回 SKIP
- 明确的可恢复异常(LLM 超时、JSON 解析失败):catch 指定异常类型,fallback 到确定性代码
禁止:
except Exception: pass/except Exception: return None(历史巨坑)- 在业务函数里裸
try/except(应该让异常冒泡到 L5)
6.8 单文件行数上限
| 层 | 单文件建议 | 硬上限 |
|---|---|---|
| L2 策略 | ≤ 80 | 120 |
| L3 features 单模块 | ≤ 200 | 300 |
| L4 reducer | ≤ 300 | 400 |
| L5 endpoints | ≤ 150 | 200 |
| L1 LLM 模块 | ≤ 150 | 200 |
超硬上限:CI 报错,PR 打回。
6.9 注释规范
- 每个策略 / feature / event 类型的顶部必须有 docstring 说明 "做什么、什么时候用、来源"
- 复杂算法段前加一行注释说明"为什么这样算"
- 禁止用注释残留旧代码("# 老版本本来是这样的...")——直接删
6.10 config / 参数使用
- 所有可调数字必须通过
ctx.cfg访问 - 禁止 hardcode magic number
- 参数默认值放在
config/strategy_stable.yaml,代码里只做ctx.cfg.foo消费
6.11 时区 / 时间戳
- 事件时间戳一律 UTC(
time.time()) - 展示时才转本地时区
- 不用
datetime.now()(无时区意识)
6.12 前置检查工具
必装:
ruff(linter + formatter,替代 flake8 + black)mypy(类型检查,--strict)pytest(测试)pytest-cov(覆盖率,目标 ≥ 80%)
CI 里跑:
ruff check .
ruff format --check .
mypy --strict werewolftown/
pytest --cov=werewolftown --cov-fail-under=80
第 7 章 — 自测规范
三层测试金字塔:unit(快、多)→ features(中、覆盖数值正确性)→ replay(慢、覆盖真实场景)
7.1 测试目录结构
tests/
├── conftest.py # 通用 fixture:make_state, make_ctx, load_config
├── unit/ # 纯函数单测
│ ├── test_board_grid.py # 邻接表
│ ├── test_shop_rules.py # 收益表
│ ├── test_sanitize.py # 消毒
│ └── test_race_gate.py # 并发网关
├── features/ # Feature 层
│ ├── test_clusters.py
│ ├── test_card_value.py
│ ├── test_plot_value.py
│ ├── test_opponents.py
│ ├── test_pool.py
│ └── test_pricing.py
├── strategies/ # 策略层
│ ├── test_select_plots/
│ ├── test_plan_build/
│ ├── test_propose_trade/
│ │ ├── test_r4_leader_skip.py
│ │ ├── test_buy_adjacent_plot.py
│ │ └── ...
│ └── test_evaluate_trade/
├── state/ # State 层
│ ├── test_reducer.py
│ ├── test_persistence.py
│ └── test_replay_backward_compat.py
├── llm/ # LLM 层(用 mock client)
│ ├── test_arbitrator.py
│ ├── test_speech.py
│ └── test_vision.py
└── replay/ # 完整对局回放
├── fixtures/ # game_id 命名的 replay 数据
├── test_replay_smoke.py # 20 场 replay smoke test
├── test_replay_regression.py # 特定 game_id 的历史 bug 回归
└── test_strategy_coverage.py # 每条 KEEP 策略至少一场触发
7.2 三层测试的目的
| 层 | 目的 | 频率 | 时间目标 |
|---|---|---|---|
| unit | 验证纯函数正确性 | 每次改 L4/L1 | < 1s |
| features | 验证数值计算与游戏规则一致 | 每次改 L3 | < 10s |
| strategies | 验证策略触发条件与产出 | 每次改 L2 | < 30s |
| state | 验证 reducer 与 replay 兼容性 | 每次改 L4 | < 20s |
| llm | 验证 LLM 兜底与超时 | 每次改 L1 | < 15s |
| replay | 验证端到端 + 无回归 | 每次 PR 合入前 | < 5min |
红线:所有 pytest 加起来 < 6 分钟;超过就要抽样跑或并行化
7.3 关键 fixture
tests/conftest.py 提供 4 个核心 fixture:
@pytest.fixture
def make_state():
"""构造 GameState 的便捷函数:make_state(my_plots=[1,2], round=2, ...)."""
...
@pytest.fixture
def make_ctx(make_state):
"""构造 DecisionContext:make_ctx(state=..., phase=..., rng_seed=42)."""
...
@pytest.fixture
def load_config():
"""context manager:with load_config('canary') as cfg: ctx = make_ctx(cfg=cfg)."""
...
@pytest.fixture
def replay_at():
"""从 replay 拉出某局某回合某 phase 的 ctx:replay_at('358890', round=2, phase='propose_trade')."""
...
必须用 fixture,禁止在测试里手工构造 state——保持一致性。
7.4 unit 测试规范
def test_calc_cluster_revenue_full():
"""Happy(fc=4) 4 连满额 = 10 万。"""
assert calc_cluster_revenue(cluster_size=4, full_count=4) == 100_000
def test_calc_cluster_revenue_multi_full():
"""8 个连通 fc=2 卡 = 4 个满额 = 16 万。"""
assert calc_cluster_revenue(cluster_size=8, full_count=2) == 160_000
规范:
- 一个测试测一个断言
- 用具体数字,不用
> 0这种模糊断言 - 名字说明测的什么条件(
test_<函数>_<场景>)
7.5 features 测试规范
def test_plot_value_adjacent_full_cluster(make_state):
"""邻接现有 Happy 2 连的 plot,我方买入可扩到满额 3 连。"""
state = make_state(
my_plots=[4, 5],
built_on={4: "Happy", 5: "Happy"},
shops=["Happy"],
round=2,
)
# plot 6 邻接 4,5,手牌有 Happy
v = plot_value(state, plot_id=6, viewer="me")
# 3 连满额 70k - 之前 2 连 30k = 40k,× 剩余 3 轮
assert v == 40_000 * 3
规范:
- fixture 构造 state,一行一场景
- 断言写明"预期收益如何算出来的"(注释)
- 覆盖:正常 / 边界 / 空输入 / 极端值
7.6 strategies 测试规范
每个策略至少 4 类测试:
class TestBuyAdjacentPlot:
def test_hit_normal(self, make_ctx):
"""R2 有邻接可买候选 → HIT。"""
ctx = make_ctx(round=2, my_plots=[4,5], has_adjacent_buy=True)
assert buy_adjacent_plot.match(ctx) == MatchResult.HIT
def test_miss_r4_leader(self, make_ctx):
"""R4 我方是领先者 → MISS,让给 R4LeaderSkip。"""
ctx = make_ctx(round=4, is_me_leader=True)
assert buy_adjacent_plot.match(ctx) == MatchResult.MISS
def test_miss_no_candidate(self, make_ctx):
"""没有邻接候选 → MISS。"""
ctx = make_ctx(round=2, my_plots=[])
assert buy_adjacent_plot.match(ctx) == MatchResult.MISS
def test_act_price_snap(self, make_ctx):
"""价格取整到 500。"""
ctx = make_ctx(round=2, buy_candidate_gain=20_100)
act = buy_adjacent_plot.act(ctx)
assert act.offer_cash % 500 == 0
7.7 replay 测试规范
Replay smoke:
@pytest.mark.parametrize("game_id", RECENT_20_GAMES)
def test_replay_no_crash(game_id):
"""任意场 replay 完整跑一遍不能崩。"""
events = load_events(f"replays/{game_id}/events.jsonl")
state = GameState.initial(player_id="子涵")
for event in events:
state = reduce(state, event)
# 最后一个 event 是 phase_result,state 里应有 4 个玩家现金记录
assert len(state.other_players) + 1 == 4
Strategy coverage:
@pytest.mark.parametrize("strategy_name", KEEP_STRATEGIES_LIST)
def test_strategy_touched_in_history(strategy_name):
"""每条 KEEP 策略在最近 50 场里至少被触发过 1 次。"""
hits = replay_scan(strategies=[strategy_name], games=recent_50_games())
assert hits > 0, f"{strategy_name} 从未被触发,可能是死代码"
Regression:
def test_bug_20260713_undefined_remaining():
"""Bug: 2026-07-13 room-358890 R2 抛 NameError('remaining')."""
ctx = replay_at("358890", round=2, phase="propose_trade")
resp = decide(ctx)
assert resp is not None
assert resp.action in {"initiate", "skip"}
7.8 覆盖率要求
- unit + features + state + llm 加权覆盖率 ≥ 80%(
pytest --cov-fail-under=80) - 每条 KEEP 策略必须有至少 1 个专属测试
- 每次修 bug 必须有对应回归测试
- 覆盖率降低的 PR 打回
7.9 测试环境隔离
- 测试永远用
strategy_dev.yaml(不影响 stable/canary) - LLM 用 mock:
tests/llm/mocks.py提供MockLLMClient,返回预设 response - 视觉:所有测试
DISABLE_VISION=true - 每个测试独立 rng:
ctx = make_ctx(rng_seed=42)
7.10 自测流程速查
改动后本地跑:
# 快速验证(约 30 秒)
pytest tests/unit tests/features tests/strategies -x --ff
# 完整验证(约 5 分钟)
pytest --cov=werewolftown --cov-fail-under=80
# 若改了策略/参数,加跑 replay
pytest tests/replay/ -v
PR 前必须 通过完整验证。
第 8 章 — 部署规范
HF Space 单进程环境 + 后端 60s 超时约束下的最小可行 canary。
8.1 部署环境认知
- 平台:Hugging Face Space (Docker)
- 限制:
- 单容器单进程;HF Space 一般给 2G 内存/2 CPU
- 重启会清空
/tmp和进程内存(因此 replay 必须持久化) - HF 提供
/data目录(挂载持久磁盘),重启不丢
- 能力:
- Space 有 API 秘钥(
API_KEY/VISION_API_KEY) - 通过
Dockerfile环境变量控制模型 - HF Space 每次 push 会自动 rebuild → 存在 build 期间旧版本仍在服务的窗口
- Space 有 API 秘钥(
- 不能:
- 多实例并行(想 A/B 只能进程内路由)
- 蓝绿部署(HF Space 单实例)
8.2 三档配置
# config/strategy_stable.yaml — 生产
# config/strategy_canary.yaml — 灰度(1 场里概率路由 X%)
# config/strategy_dev.yaml — 本地实验
运行时选择:
# transport/endpoints.py
def _pick_profile(game_id: str) -> str:
canary_ratio = float(os.getenv("CANARY_RATIO", "0.0"))
if canary_ratio == 0.0:
return "stable"
# 用 game_id 哈希决定本局用哪个 profile;同一局全程一致
h = int(hashlib.md5(game_id.encode()).hexdigest(), 16) / 2**128
return "canary" if h < canary_ratio else "stable"
关键性质:canary 路由按 game_id 粒度,一局内绝不换 profile(否则 State 层参数中途改变,行为诡异)。
8.3 部署流程
步骤:
- 代码合入主分支 → git push → HF Space 自动 rebuild
- rebuild 期间(约 3-5 分钟):老版本仍在服务
- rebuild 完成后:新代码上线,
CANARY_RATIO默认0.0(所有对局走 stable) - 打开灰度:在 HF Space Settings 里改环境变量
CANARY_RATIO=0.1(10% 对局走 canary) - 观察 24 小时:看 canary_vs_stable 报告
- 决策:
- 若 canary 有正收益 → 提升
CANARY_RATIO到 0.5、1.0 - 若 canary 有负收益 → 关闭
CANARY_RATIO=0.0,回滚代码
- 若 canary 有正收益 → 提升
8.4 灰度策略
参数改动(走 5.3):
- 只改
strategy_canary.yaml,不动代码 CANARY_RATIO=0.1起步- 观察 20+ 场对局
- 平均总现金差 > 0 且 P25 不劣化 → 参数从 canary 提升到 stable(改 stable.yaml)
代码改动(走 5.1/5.2/5.4/5.5/5.6/5.7/5.8):
- 用
PROFILE_STRATEGIES环境变量分流:# transport/endpoints.py def _register_strategies(profile: str): if profile == "canary": return canary_strategies # 含新策略 return stable_strategies # 老策略 - 参数与代码分流独立控制
- 观察窗口至少 20 场
8.5 回滚
参数回滚(分钟级):
- 直接把 canary.yaml 里的字段还原
- 或
CANARY_RATIO=0.0
代码回滚(半小时级):
- Git revert 到上一个 stable commit
- Push → HF rebuild
- rebuild 期间 canary 仍在跑(这是隐患,必要时同时
CANARY_RATIO=0.0)
紧急回滚(若 canary 引发系统性错误):
- 立即
CANARY_RATIO=0.0 - 观察 stable 是否恢复正常
- 排查 canary 问题
- 修好后重新灰度
8.6 部署前 checklist
- 所有测试通过(含 replay batch)
- canary_vs_stable 报告:平均总现金差 > 0 且 P25 不劣化
- 已在本地
STRATEGY_PROFILE=canary跑过 10+ 场 replay - PR 描述里有"预期胜率变化"和"关键观察指标"
- 老 replay 兼容性测试通过(如果动过 L4)
- Git tag 加一版号(如
v2026.07.14.canary1),便于回滚
8.7 部署后 checklist
- HF Space rebuild 成功(看构建日志)
- 前 5 局对局无 crash(看第 9 章观察规范)
- canary_vs_stable 数据在观测面板可看
- 若 24h 内出现严重问题 → 走"紧急回滚"
8.8 特殊场景
Push 完发现有 bug 想快速回滚:
- 立即
CANARY_RATIO=0.0 - 若 stable 也受影响(比如状态层 bug):立即 git revert push
HF Space 冷启动:
- 第一场对局会因 pip install 或 Docker 拉取慢
- 冷启动期出错的对局记录清晰,不算 canary 事故
Space 反复重启:
- 检查
/data/replays/有没有累计过多(每局约 200KB) - 定期清理
/data/replays/中 30 天前的对局
第 9 章 — 观察规范
你之前的 pain point:改了策略但不知道有没有生效。这一章的目标是任何改动上线后 24 小时内可以拿到量化答案。
9.1 观察的三个层次
| 层 | 目的 | 输出 |
|---|---|---|
| L9.1 事件流原始记录 | 复现 / 回放 | replays/<game_id>/events.jsonl + decisions.jsonl |
| L9.2 结构化指标 | 快速判断"有没有生效"、"有没有坏" | metrics/<date>/*.jsonl |
| L9.3 对局报告 | 一局结束的自动复盘 | reports/<game_id>.md |
9.2 事件流持久化(L9.1)
目录结构:
/data/
└── replays/
└── <game_id>/
├── meta.json # 局 meta:player_id, players, start_ts, end_ts, profile
├── events.jsonl # perceive 事件流(一行一 event)
├── decisions.jsonl # interact 决策日志(一行一决策)
└── llm_calls.jsonl # LLM 调用记录(可选,脱敏后)
写入时机:
events.jsonl:每次 perceive 事件到达,L4 reducer apply 前 appenddecisions.jsonl:每次 interact 返回前 appendmeta.json:phase_start时创建,phase_result时补 end_ts
格式:
// events.jsonl 一行
{"kind":"deal_countered","ts":1720839200.12,"round":2,"deal_id":"deal_x1","source":"心怡",...}
// decisions.jsonl 一行
{"ts":1720839200.5,"round":2,"phase":"propose_trade","game_id":"358890",
"profile":"canary","strategy":"buy_adjacent_plot",
"match_result":"HIT","action":{"kind":"initiate","target":"心怡","offer_cash":15000,"demand_plots":[6]},
"features":{"gain":40000,"net":25000,"is_leader":false},
"elapsed_ms":250,"llm_used":false}
必须:
strategy字段是选中策略的NAMEfeatures是决策依据的关键数字(够重构复现)profile记录了本局走 stable 还是 canaryllm_used记录了本次决策是否用到 LLM
9.3 结构化指标(L9.2)
每小时 cron 从 replays/ 里聚合出:
/data/metrics/
└── 20260714/
├── strategy_hits.jsonl # 每策略每小时命中次数
├── strategy_action.jsonl # 每策略每小时产出的 Action 汇总(价格分布)
├── llm_calls.jsonl # LLM 调用次数 / 成功率 / 延迟
├── errors.jsonl # 错误 / 超时统计
└── canary_vs_stable.jsonl # canary/stable 分组的收益对比
canary_vs_stable.jsonl 是最重要的一份:
{"hour":"2026-07-14T10","profile":"stable","games":5,"avg_cash":425000,"p25":380000,"p50":420000,"p75":465000}
{"hour":"2026-07-14T10","profile":"canary","games":3,"avg_cash":448000,"p25":410000,"p50":445000,"p75":490000}
9.4 对局报告(L9.3)
每局 phase_result 后自动生成 markdown 报告:
# Game 358890 报告
- 玩家: 子涵 (第 2 名, $445k)
- Profile: canary
- 时长: 12min
- 最大集群: Happy 3 连(工位 4-5-6)
## 每回合决策
| R | Phase | 策略 | 结果 | 关键指标 |
|---|---|---|---|---|
| 1 | select_plots | enumerate_combos | 选 [4,5,7] | score=95000 |
| 1 | propose_trade | buy_adjacent_plot | 出 15k 买浩宇 6 号 | net=25000 |
| 1 | evaluate_trade | accept_on_net | ACCEPT 心怡卖 5 号 | net=8000 |
| ... |
## 异常
- R2 propose_trade: LLM 兜底触发 (2.3s)
- R3 evaluate_trade: 卡估值 net=-245000 (可能 bug)
报告用来做什么:
- 复盘本局所有决策
- 一眼看出策略是否被触发
- 异常段自动醒目
- 可以对着 PR 描述里的"预期决策"逐条验证
9.5 观察面板(/debug/*)
Web 面板(HF Space 上通过 https://<space>/debug/ 访问)提供:
/debug/games # 最近 50 局列表
/debug/games/<game_id> # 单局报告(自动生成的 markdown)
/debug/strategies # 每策略近 7 日命中数柱状图
/debug/canary_report # 最近 24h canary vs stable 汇总
/debug/errors # 最近 100 条错误 + 超时
/debug/prompts # 最近 100 次 LLM 调用(现有的,增强分组)
9.6 一次改动的观察路径
你改了什么 → 打开哪个视图 → 期望看到什么:
| 改动类型 | 看这个视图 | 期望信号 |
|---|---|---|
| 5.1 加新策略 | /debug/strategies |
新策略 24h 内命中次数 > 0 |
| 5.2 改策略 | /debug/strategies/<name> |
命中次数变化 + Action 分布变化,符合 PR 预期 |
| 5.3 调参数 | /debug/canary_report |
canary 平均现金 > stable |
| 5.4 修 bug | /debug/errors |
相关错误从计数消失 |
| 5.5 加 Feature | /debug/games/<game_id> |
相关策略的 features 字段里出现新指标 |
| 5.6 加 Event | /debug/games/<game_id> |
events.jsonl 里出现新事件类型 |
| 5.7 改 LLM 参与 | /debug/prompts + llm_calls.jsonl |
调用次数 / 成功率符合预期 |
| 5.8 改基建 | /debug/errors + replay all check |
无新增 error;全 replay 无损重放 |
9.7 SLO(服务质量目标)
这些红线,一旦触碰立即回滚:
| 指标 | 阈值 | 数据源 |
|---|---|---|
| 决策异常率 | ≤ 1% | errors.jsonl |
| 决策超时率(>50s) | ≤ 0.5% | decisions.jsonl |
| LLM 兜底率 | ≤ 5% | decisions.jsonl 里 llm_used=true |
| 单局决策失败次数 | ≤ 5(后端会下线) | decisions.jsonl |
| 死代码策略(0 hit / 24h) | 0 | strategy_hits.jsonl |
| 上线 canary 后胜率下降 | ≤ 5% | canary_vs_stable.jsonl |
9.8 长时间趋势观察
周报:reports/weekly/YYYY-Www.md
# 2026-W28 周报
- 总局数: 87
- 胜率: 34% (stable) / 41% (canary)
- ELO 变化: +23
- 触发最多的策略: buy_adjacent_plot (156 次)
- 从未触发的策略: sell_cheap_non_adjacent (0 次) → 需要考虑删除
- LLM 兜底次数: 12 / 87*7 = 2% 决策
- 新增 bug: 2 (已修)
- 最大遗憾: R2 358950 - 应该买卡但走了 SKIP
从周报判断:
- 死代码策略(0 触发)应该删或改条件
- 高 LLM 兜底率的场景应该抽取为策略
- 胜率没提升的策略应该重新审视
9.9 观察工具的实现
不需要重写监控平台——用最简单的方式:
metrics/*.jsonl用 cron/scheduled_tasks 每小时聚合reports/*.md由phase_result事件触发生成/debug/*用 FastAPI + Jinja 模板生成静态 HTML- 数据不入数据库,都是 jsonl 文件
保留手段:老的 hf_monitor_log.txt cron 依然跑,但用途从"tail HF log"变成"tail decision_log 并跑异常检测"
第 10 章 — 常见坑与冲突协议
这一章是"历史事故档案",每一条都能追溯到 STRATEGY_V*.md 里一次具体翻车。每次改动前扫一眼,避免踩同样的坑。
P10-1 状态多源头(agent 拿不到最新数据的根源)
症状:agent 决策时依据的 cash/plots/built_on 与实际不符
历史事故:V16-Fix7、V18-Fix4#1、Fix4#3、V17 §D6 全部在打补丁
根因:memory 变量、事件流重放、req 覆盖、纠正函数四个地方都能改状态
规范:只 L4 State 层管状态;L2/L3 只读;perceive 事件是唯一写入源
判断:如果你在策略里想写 agent.memory.set_variable(...) → 停下来,改本手册
P10-2 视角混淆(净值反号)
症状:_evaluate_deal_value 算出 net = -245000(应为 +30000 之类的合理值)
历史事故:V18 §2 全部
根因:_evaluate_deal_value 期望"我方视角",但 Path A/B/C/F20 传参视角不一致
规范:Offer dataclass 用 to_me / from_me 命名,禁止再传 offer_*/demand_* 4 元组
判断:写策略时如果需要 "swap",说明抽象有问题
P10-3 函数内 lazy import 隐藏 bug
症状:模块级 import 错了 3 周没人发现(from ..tools.counter_deal import _validate_offer_against_assets — 实际在 initiate_trade.py)
历史事故:V17 §D1
根因:_try_code_only_eval 函数体内 import 才会执行,import 错误被 except Exception: return None 静默吞
规范:所有 import 放模块顶部;ruff 应能扫出未使用 import;CI 强制 mypy --strict
判断:任何函数体内 from ... import ... → PR 打回,除非明确 comment 解释
P10-4 except Exception 掩盖 bug
症状:某个策略实际从未生效,但日志只有 WARNING [xxx] 代码兜底异常,降级LLM
历史事故:V17 §D3(Fix4/Fix5/Fix6 定价改动从未执行过)
规范:除 L5 边界 + 明确可恢复异常,禁止 catch Exception
判断:try/except: return None 是重大 code smell
P10-5 状态"降级模式"(黑天鹅制造机)
症状:某场对局 agent 从头到尾 SKIP 所有交易,全 REJECT 所有要约
历史事故:V14-Fix F6 引入 → V17 §D4 阉割 → 待修改文档要求彻底删
根因:一次误判(比如 R1 事件断层)触发 degraded_mode=True,整局静默失败
规范:任何"整局静默改行为"的开关都禁止;数据不全时应该明确暴露,让上层决定
判断:不设 kill switch;能算就算,算不了报错
P10-6 事件去重键太宽
症状:某场 deal_accepted 事件到达两次但 settle_assets 只跑一次,第一次 crash 时资产未落账
历史事故:V14-Fix F7 引入的 _seen_perceive_events
规范:用幂等 settle(每个事件带唯一 id,reducer 里做幂等判断),不用去重集合
判断:如果需要"跳过已见事件",说明 reducer 不幂等
P10-7 R4 决策清仓打架
症状:R4 一次决策同时被两个"清仓策略"处理
历史事故:V16-Fix2 F10 + V16 R12 步骤 4 未清理老代码
规范:每个 phase 里同一 return path 只允许一个策略 HIT;重叠策略必须合并
判断:定期扫 _order.py,寻找"逻辑重叠但没显式互斥"的策略对
P10-8 fc>=5 卡的估值坑
症状:花 45k 买 3 张 Happy 卡,最后建了 1 张
历史事故:V18-Fix2 (_compute_card_cluster_gain 里没考虑 fc>=5)
规范:**所有卡估值路径统一走 card_value(state, card, purpose)**,一处判断 fc>=5
判断:如果代码里出现"if fc>=5",检查 6 处买卡路径是否都判了
P10-9 built_on int/str key
症状:state["built_on"][6] KeyError,实际存的是 "6"
历史事故:JSON 序列化后 str key,运算需要 int key
规范:**State 层字段用 dict[int, str]**,所有输入立即转 int
判断:get_built_on() 这种"每次都转"的模式 = state 层没做对
P10-10 阻塞的 asyncio + threading 混用
症状:interact() 在 uvicorn worker 里被阻塞
历史事故:_async_interact 里的 loop.is_running() 分支
规范:全 async;用 async future 做竞速,不用 threading.Event
判断:import threading 出现在 L5 之外的地方 → 打回
P10-11 参数散落 200+ 处
症状:改一个 buy_discount 要 grep 全项目
历史事故:全项目通病
规范:**所有 magic number 挪到 config/strategy_*.yaml**;代码里只有 ctx.cfg.xxx
判断:if round == 4: xxx = 12000 出现在策略里 → 参数化
P10-12 死代码策略
症状:某条策略从未被走到,但仍占用维护成本
历史事故:L4-31 F22 倒卖(已被 Fix3#5 移除但代码可能残留)
规范:周报会自动扫出"0 hit / 7 天"的策略,必须删除或明确废弃
判断:test_strategy_touched_in_history 会强制这条
冲突协议:两个改动同时命中
场景:本地 agent 正在写策略 A,PR 未合入;你要修 bug B,B 影响 A 依赖的 feature 协议:
- B 优先:bug 修复不等策略
- A 的 PR 需要 rebase 到 B 之后
- A 的作者(本地 agent)需要重新跑 canary vs stable,并附新的报告到 PR
场景:你要加新 Feature X,本地 agent 已经在另一 PR 加了功能重叠的 Feature Y 协议:
- 先合入其中一个
- 另一个 rebase 后,把重叠部分改为直接调用已合入的 Feature
- 禁止 X 和 Y 都保留——两个 feature 做同一件事必然会漂移
第 11 章 — 术语与命名约定
11.1 领域术语
| 术语 | 定义 | 备注 |
|---|---|---|
| State | 一局对局的完整数据快照 | L4 内的 GameState |
| Event | perceive 收到的一条消息 | 引擎推送 → 落到 events.jsonl |
| Action | interact 要返回的一次决策 | L2 策略的输出 |
| Reducer | (state, event) → new_state 纯函数 |
唯一改状态的地方 |
| Feature | 从 state 派生的数值指标 | L3 纯函数 |
| Strategy | 决策单元 | L2 一个文件 |
| DecisionContext | interact 一次决策的只读上下文 | 含 state + features + rng + cfg |
| MatchResult | 策略是否要接管本次决策 | HIT / MISS / UNSURE |
| Profile | 一份完整参数配置 | stable / canary / dev |
| Offer | 一次交易报价的双方视角 | to_me / from_me |
| PendingLedger | 追踪本回合已卖/挂起/取消 | 合并旧 pending_proposals 等 |
| Speech Hint | 话术生成模板名 | "buy_plot", "sell_card", ... |
11.2 游戏规则术语(不改,直接沿用引擎)
plot= 工位(1-52)shop= 业务("淘小宝"/"猫天天"等 8 种)built_on= {plot_id: shop_type},物理事实cluster= 同类连通集群full_count(fc) = 满额数 (2/3/4/5)round= 回合 (1-4)quota= 配额(选地保留数 / 交易次数 / 还价次数)deal_id= 交易 ID (deal_xxx或free_xxx)phase= 阶段(SELECT_PLOTS / PROPOSE_TRADE 等)
11.3 视角约定
在代码里永远:
me/opponent而不是self/other/ownerto_me/from_me而不是offer_*/demand_*viewer=me或viewer=opponent参数化
在 Prompt 里(跟引擎对齐):
- 保留
offer_*/demand_*,但只在 L1 内使用;L2/L3 一律 to_me/from_me
11.4 文件命名约定
| 类型 | 规则 |
|---|---|
| 策略文件 | snake_case.py,动词/结果描述 (buy_adjacent_plot.py, reject_bad_swap.py) |
| Feature 模块 | snake_case.py,主体名词 (plot_value.py, opponents.py) |
| Reducer case | _reduce_<event_name>() |
| Test 文件 | test_<被测模块>.py |
| Bug 回归测试 | test_bug_YYYYMMDD_<short_desc>() |
| Config 文件 | strategy_<profile_name>.yaml |
| Replay 目录 | replays/<game_id>/ |
| Report 文件 | reports/games/<game_id>.md, reports/weekly/YYYY-Www.md |
第 12 章 — PR 变更 Checklist(贴在每次 PR 里)
贴到 PR 描述最上方,逐项打勾。未打勾的项必须解释理由。
12.1 通用(所有 PR 必过)
- 明确变更类型(对应 5.1–5.8 哪一节)
- 顶部 docstring 说清"这次改了什么,为什么"
-
ruff check .通过 -
ruff format --check .通过 -
mypy --strict werewolftown/通过 -
pytest --cov=werewolftown --cov-fail-under=80通过 - 无函数内
import(P10-3) - 无
except Exception: pass/return None(P10-4) - 无硬编码 magic number(P10-11)
- 单文件行数在硬上限之内(第 6.8 节)
- Git commit 有清晰 message + Signed-off-by
12.2 加/改策略专用(5.1、5.2)
- 策略文件 ≤ 120 行
-
NAME/PHASE/PRIORITY三常量齐全 -
_order.py已注册 + priority 与邻居不冲突 - 顶部注释写清"来源"(inventory 编号或 game_id)
-
tests/strategies/有针对性单测(HIT/MISS/act 覆盖) -
tests/replay/至少 1 场真实对局证明触发 - (改策略时)
replay_diff报告附在 PR 描述 - (改策略时)"改动记录" 更新到顶部注释
12.3 参数调整专用(5.3)
- 只改
config/strategy_canary.yaml(不动 stable,不动代码) -
changelog:段有条目(字段/from/to/reason/author) - 20 场 replay 的 canary vs stable 报告附 PR
- 平均总现金差 > 0 且 P25 不劣化
- 有对照单测证明"仅参数变化,非逻辑变化"
12.4 修 bug 专用(5.4)
-
tests/有以 game_id 命名的回归测试test_bug_YYYYMMDD_... - 测试在修前 fail、修后 pass
- 修改范围最小(无夹带重构)
- Last 20 replay batch 无回归
- PR 描述:"bug 潜伏时间 / 发现时间 / 潜伏原因"
12.5 加 Feature 专用(5.5)
- Feature 放在合适的模块(按 5.5.2 判断)
- 输入 state,输出 dataclass/原始类型,无副作用
- docstring 说清输入输出语义
- 类型注解完整
-
tests/features/test_<module>.py覆盖 ≥ 3 case - 若用
@feature_cache,验证过缓存失效
12.6 加 Event 专用(5.6)
-
state/events.py定义新事件类,frozen=True -
state/reducer.py加对应 case -
transport/endpoints.py加 req → event 解析 -
state/persistence.py:_EVENT_TYPES注册新类型 -
tests/state/有 reducer 测试 + backward-compat 测试 - 旧 replay 仍能加载
12.7 LLM 参与改动专用(5.7)
- 明确模式升降(M2↔M3 等)
- Prompt 通过 pydantic schema 约束输出
- LLM 失败 fallback 是纯代码
- 超时保护 ≤ 4s
- 单测:正常返回、超时、格式错误、无效值 四种
12.8 基建 / 状态层改动专用(5.8)
- 明确 A/B/C 类
- A 类:回归测试 + 全 replay 通过
- B 类:新字段有默认值 + 本手册 3.2 节更新
- C 类:分三个 PR(加字段 / 切消费方 / 删旧字段)
- race_gate 改动:并发压测通过
- persistence 改动:backward-compat 通过
12.9 部署前必查
- canary_vs_stable 报告:均值 > 0,P25 不劣化
- 本地 STRATEGY_PROFILE=canary 跑过 10+ 场 replay
- Git tag 加版号
- PR 描述:"预期胜率变化" + "关键观察指标"
12.10 部署后必看(24h 内)
-
/debug/errors无新增错误 -
/debug/strategies新策略命中 > 0 -
/debug/canary_report数据符合预期 - SLO 未触碰(第 9.7 节)
CHANGELOG
v1.0.0 — 2026-07-13
- 首版发布
- 建立五层架构(L1–L5)与 8 类变更规范(5.1–5.8)
- 覆盖:编码规范 / 自测 / 部署 / 观察 / 常见坑 / PR checklist
- 配套文档:
20260713_strategy_inventory.md(136 条策略台账)
未来版本规则
- patch(1.0.x):勘误、示例更新、trivia
- minor(1.x.0):新增变更类型、新观察规范、新工具集成
- major(x.0.0):架构变化(如加新层)、契约破坏性变更
加条 CHANGELOG 的格式
### v1.1.0 — 2026-08-01
- 加入第 5.9 节:多语言 message 支持
- 更新 6.11 节:时区处理示例
- 修正 5.3.2 里参数结构示意的错别字(fc4 → fc_4)