562 lines
22 KiB
Markdown
562 lines
22 KiB
Markdown
---
|
||
id: plan.week09
|
||
title: "第 9 周：阶段 2——五个重点 A~E 深挖 + 案例卡"
|
||
type: week
|
||
domain: plan
|
||
stack: []
|
||
week: [9]
|
||
est_hours: 12
|
||
order: 9
|
||
---
|
||
|
||
# 第 9 周：阶段 2——五个重点 A~E 深挖 + 案例卡
|
||
|
||
> 第三段的第 2 周。上周 `week08-inventory-architecture.md`，下周 `week10-agent-m1-tools-mcp.md`。
|
||
|
||
**本周目标**：把五个重点各自做一个小实验，再结合你在公司遇到的真实问题写成**案例卡**。案例卡是简历“项目经历”和面试“讲一个你解决过的难题”的直接素材，也是项目三的设计输入。
|
||
|
||
**每天的固定结构**
|
||
|
||
1. 解释这个重点是什么、为什么难；
|
||
2. 跑一个**可运行的最小实验**（Python 标准库就能跑，不依赖公司代码），亲眼看到问题和解决办法；
|
||
3. 对照公司的实际情况写案例卡（只写机制，私有细节放 `notes/private/`）。
|
||
|
||
**案例卡模板**（每张卡一个文件，放 `notes/cards/`）
|
||
|
||
```text
|
||
标题：
|
||
对应重点：A / B / C / D / E
|
||
现象：
|
||
根因（一层）：
|
||
根因（系统层，可选）：
|
||
当时怎么处理 / 未处理：
|
||
可迁移结论（面试一句）：
|
||
若重做的设计：
|
||
关联代码/路径：（私有，只写在 notes/private/ 里，不进仓库）
|
||
日期：
|
||
```
|
||
|
||
**本周产出**
|
||
|
||
```text
|
||
notes/cards/
|
||
├── A-orchestration.md
|
||
├── B-source-of-truth.md
|
||
├── C-streaming.md
|
||
├── D-human-in-loop.md
|
||
├── E-integration.md
|
||
└── README.md # 汇总“若重做的设计”，作为项目三的设计输入
|
||
notes/experiments/ # 本周的实验脚本
|
||
```
|
||
|
||
---
|
||
|
||
## 第 1 天：A 多 Agent 编排
|
||
|
||
### 学习内容
|
||
|
||
- 说明：多 Agent 编排要回答四个问题：
|
||
1. **谁派活**：固定顺序（链式）、有向无环图（DAG），还是由一个“协调者”模型动态决定？
|
||
2. **什么时候结束**：达成共识？轮数上限？成本上限？**没有上限的循环一定会出事**（模型可能永远不说“同意”）。
|
||
3. **怎么防止重复执行**：用户连点两次“开始”、或者前端重连后又发一次请求，同一个任务不能跑两遍——需要“同一时刻只有一个执行者”的保证（锁 / 租约 / 唯一约束）。
|
||
4. **每个角色的输入输出**：结构化（JSON + 校验），而不是自由文本。
|
||
- 案例：协调者 + 两个角色的最小循环，包含轮数上限、共识判断、防重复执行。
|
||
|
||
```python
|
||
import threading
|
||
from dataclasses import dataclass, field
|
||
|
||
|
||
@dataclass
|
||
class RunState:
|
||
topic: str
|
||
round: int = 0
|
||
transcript: list[str] = field(default_factory=list)
|
||
status: str = "running"
|
||
|
||
|
||
def proposer(state: RunState) -> str:
|
||
return f"方案 v{state.round}：用 Redis 做缓存"
|
||
|
||
|
||
def critic(state: RunState) -> str:
|
||
# 模拟：第 3 轮开始同意
|
||
return "同意" if state.round >= 3 else "反对：没考虑缓存一致性"
|
||
|
||
|
||
def coordinator(state: RunState, max_rounds: int) -> str:
|
||
"""返回下一步：proposer / critic / finish。"""
|
||
if state.transcript and state.transcript[-1].endswith("同意"):
|
||
return "finish"
|
||
if state.round >= max_rounds:
|
||
return "finish" # 轮数上限是必须的“保险丝”
|
||
return "proposer" if len(state.transcript) % 2 == 0 else "critic"
|
||
|
||
|
||
running: set[str] = set()
|
||
lock = threading.Lock()
|
||
|
||
|
||
def start_run(run_id: str, topic: str, max_rounds: int = 5) -> RunState | None:
|
||
with lock: # 防重复执行：同一个 run_id 同一时刻只允许一个执行者
|
||
if run_id in running:
|
||
print(f"{run_id} 已在执行，拒绝重复启动")
|
||
return None
|
||
running.add(run_id)
|
||
try:
|
||
state = RunState(topic)
|
||
while (step := coordinator(state, max_rounds)) != "finish":
|
||
if step == "proposer":
|
||
state.round += 1
|
||
state.transcript.append(f"提议者：{proposer(state)}")
|
||
else:
|
||
state.transcript.append(f"评审者：{critic(state)}")
|
||
state.status = "consensus" if state.transcript[-1].endswith("同意") else "max_rounds"
|
||
return state
|
||
finally:
|
||
with lock:
|
||
running.discard(run_id)
|
||
|
||
|
||
s = start_run("run-1", "缓存方案")
|
||
assert s is not None
|
||
print("\n".join(s.transcript))
|
||
print("结束原因：", s.status, "轮数：", s.round)
|
||
|
||
s2 = start_run("run-2", "缓存方案", max_rounds=2)
|
||
assert s2 is not None and s2.status == "max_rounds"
|
||
print("run-2 结束原因：", s2.status)
|
||
|
||
running.add("run-3") # 模拟 run-3 正在执行
|
||
assert start_run("run-3", "x") is None
|
||
```
|
||
|
||
- 说明（接着讲）：上面的 `running` 集合只在一个进程内有效。多个进程 / 多台机器时要换成数据库唯一约束或 Redis 锁（第 3 周学过 `SET NX PX`），并且要有**过期时间**，否则执行者崩溃后锁永远不释放——第 12 周会实现带租约（lease）的版本。
|
||
|
||
**易错点 / 面试常问（补充）**
|
||
|
||
- 易错点：案例卡只写机制，不写公司的接口路径、表名、字段名；私有细节放 `notes/private/`。
|
||
- 面试常问：多 Agent 什么时候结束（共识条件 + 最大轮数兜底）；怎么防止同一个任务被两个进程同时执行（租约 / 锁 + 带条件的更新）。
|
||
|
||
### 要建的文件
|
||
|
||
- `notes/experiments/a_orchestration.py`（上面的代码）、`notes/cards/A-orchestration.md`
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
python3 notes/experiments/a_orchestration.py
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 案例卡写清：公司产品里“谁派活、什么时候结束、有没有防重复”。示例（**按你的真实情况改写**）：
|
||
|
||
```text
|
||
标题：研讨轮次由前端驱动，缺少服务端的结束条件和防重复保证
|
||
对应重点：A
|
||
现象：同一个研讨在两个标签页里被同时推进，出现重复发言；偶尔出现停不下来的讨论
|
||
根因（一层）：下一轮由前端发起，服务端不知道“当前该谁、是否已在执行”
|
||
根因（系统层）：编排状态没有服务端的权威记录
|
||
当时怎么处理：前端加按钮禁用和轮数上限
|
||
可迁移结论：编排必须有服务端的唯一执行者和明确的结束条件（共识 + 轮数上限 + 成本上限）
|
||
若重做的设计：服务端状态机 + 执行租约 + 结构化的协调者输出
|
||
日期：
|
||
```
|
||
|
||
---
|
||
|
||
## 第 2 天：C 流式与长任务
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- SSE 让用户实时看到生成过程，但**连接是脆弱的**：关页面、切网络、手机锁屏、代理超时都会断。
|
||
- 如果“任务的执行”绑在“这条连接”上（在请求处理函数里循环调用模型），连接一断，任务就停了。
|
||
- 正确做法：**执行和推送分离**。任务在后台独立执行，把每一步的事件写进持久化的事件表；SSE 只是“订阅事件”，断开后用 `Last-Event-ID` 从断点补发（第 4 周 FastAPI 示例已经演示过）。
|
||
- 案例：对比两种写法在“客户端中途断开”时的表现。
|
||
|
||
```python
|
||
import asyncio
|
||
|
||
events: list[str] = []
|
||
|
||
|
||
async def work_step(i: int) -> str:
|
||
await asyncio.sleep(0.02)
|
||
return f"第 {i} 步完成"
|
||
|
||
|
||
async def coupled_handler():
|
||
"""写法一：执行绑在连接上。"""
|
||
for i in range(1, 6):
|
||
yield await work_step(i)
|
||
|
||
|
||
async def background_job(job_events: list[str]):
|
||
"""写法二：后台执行，事件写进“事件表”。"""
|
||
for i in range(1, 6):
|
||
job_events.append(await work_step(i))
|
||
|
||
|
||
async def subscriber(job_events: list[str], last_id: int = 0):
|
||
"""SSE 订阅：只读事件，从 last_id 之后补发。"""
|
||
while True:
|
||
while last_id < len(job_events):
|
||
last_id += 1
|
||
yield last_id, job_events[last_id - 1]
|
||
if len(job_events) == 5:
|
||
return
|
||
await asyncio.sleep(0.01)
|
||
|
||
|
||
async def main():
|
||
# 写法一：客户端收到 2 条后断开
|
||
received = []
|
||
gen = coupled_handler()
|
||
async for e in gen:
|
||
received.append(e)
|
||
if len(received) == 2:
|
||
break
|
||
await gen.aclose() # 连接断开 → 生成器被关闭 → 后面的步骤再也不会执行
|
||
print("写法一：执行到", received[-1], "后就停了")
|
||
|
||
# 写法二：客户端收到 2 条后断开，1 秒后重连
|
||
job_events: list[str] = []
|
||
task = asyncio.create_task(background_job(job_events))
|
||
last = 0
|
||
async for eid, e in subscriber(job_events):
|
||
last = eid
|
||
if eid == 2:
|
||
break # 断开
|
||
await task # 后台任务不受影响，继续执行
|
||
resumed = [e async for _, e in subscriber(job_events, last_id=last)]
|
||
print("写法二：重连后补发", resumed)
|
||
assert len(resumed) == 3
|
||
|
||
|
||
asyncio.run(main())
|
||
```
|
||
|
||
**易错点 / 面试常问（补充）**
|
||
|
||
- 面试常问：用户关掉页面后为什么流程停了（编排循环跑在浏览器里），怎么改（编排放到后台作业，前端只订阅进度，断线后按 `Last-Event-ID` 补齐）。
|
||
|
||
### 要建的文件
|
||
|
||
- `notes/experiments/c_streaming.py`、`notes/cards/C-streaming.md`
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
python3 notes/experiments/c_streaming.py
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 在公司开发环境做一次“断网 / 关页面”实验（只观察，不改代码），记录“单轮继续”和“整个循环停掉”的区别，写进卡片。
|
||
- 可迁移结论示例：“长任务的生命周期不能等于一条 HTTP 连接的生命周期；执行放后台，连接只负责订阅，断线按事件 ID 补发。”
|
||
|
||
---
|
||
|
||
## 第 3 天：B 状态与权威源
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- 同一个事实（比如“任务进行到哪一步”）如果存了多份（前端内存、草稿、作业表、业务任务表），就会出现**不一致**：一处更新了，另一处没更新。
|
||
- 原则：**每类状态只有一个权威来源**，其他地方要么从它读，要么由它的变化事件**派生**；需要同时改两张表时，用**同一个事务**。
|
||
- 跨服务无法用一个事务时，用 **outbox（发件箱）模式**：在同一个事务里“改业务数据 + 写一条待发送事件”，再由后台程序把事件可靠地发出去（至少一次，接收方要幂等）。
|
||
- 案例：重现“作业进行到第 3 步，任务状态还是未开始”，再用“同一事务 + outbox”修好。
|
||
|
||
```python
|
||
import sqlite3
|
||
|
||
db = sqlite3.connect(":memory:")
|
||
db.executescript("""
|
||
CREATE TABLE tasks (id INTEGER PRIMARY KEY, status TEXT NOT NULL);
|
||
CREATE TABLE jobs (id INTEGER PRIMARY KEY, task_id INT, step INT, status TEXT);
|
||
CREATE TABLE outbox (id INTEGER PRIMARY KEY AUTOINCREMENT, topic TEXT, payload TEXT, sent INT DEFAULT 0);
|
||
INSERT INTO tasks VALUES (1, 'todo'), (2, 'todo');
|
||
INSERT INTO jobs VALUES (1, 1, 0, 'running'), (2, 2, 0, 'running');
|
||
""")
|
||
|
||
|
||
def advance_bad(job_id: int, step: int) -> None:
|
||
# 只更新作业表，“顺便”更新任务的代码写在另一个地方，而且忘了调用
|
||
db.execute("UPDATE jobs SET step = ? WHERE id = ?", (step, job_id))
|
||
db.commit()
|
||
|
||
|
||
def advance_good(job_id: int, step: int, total: int = 3) -> None:
|
||
with db: # 同一个事务：作业进度、任务状态、待发事件，要么都成功，要么都不写
|
||
db.execute("UPDATE jobs SET step = ?, status = ? WHERE id = ?",
|
||
(step, "done" if step == total else "running", job_id))
|
||
db.execute("UPDATE tasks SET status = ? WHERE id = (SELECT task_id FROM jobs WHERE id = ?)",
|
||
("done" if step == total else "doing", job_id))
|
||
db.execute("INSERT INTO outbox (topic, payload) VALUES ('job.progress', ?)", (f"{job_id}:{step}",))
|
||
|
||
|
||
for s in (1, 2, 3):
|
||
advance_bad(1, s)
|
||
advance_good(2, s)
|
||
|
||
print("坏写法：", db.execute("SELECT j.step, t.status FROM jobs j JOIN tasks t ON t.id = j.task_id WHERE j.id = 1").fetchone())
|
||
print("好写法：", db.execute("SELECT j.step, t.status FROM jobs j JOIN tasks t ON t.id = j.task_id WHERE j.id = 2").fetchone())
|
||
|
||
# 后台发件程序：把未发送的事件发出去（这里只打印），发完标记
|
||
for oid, topic, payload in db.execute("SELECT id, topic, payload FROM outbox WHERE sent = 0").fetchall():
|
||
print("发送事件", topic, payload)
|
||
db.execute("UPDATE outbox SET sent = 1 WHERE id = ?", (oid,))
|
||
db.commit()
|
||
```
|
||
|
||
**易错点 / 面试常问（补充）**
|
||
|
||
- 易错点：“先写业务表再发消息”和“先发消息再写业务表”都可能在中间崩溃，导致两边不一致；发件箱模式把“要发的消息”和业务数据放在同一个事务里写入，再由后台程序发送。
|
||
- 面试常问：什么是权威数据源（出现分歧时以谁为准）；草稿和正式记录为什么要分开。
|
||
|
||
### 要建的文件
|
||
|
||
- `notes/experiments/b_source_of_truth.py`、`notes/cards/B-source-of-truth.md`
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
python3 notes/experiments/b_source_of_truth.py
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 在公司环境对比草稿数据和任务 / 过程报告，找出哪些状态没有回写，写进卡片（字段名、表名只写在 `notes/private/`）。
|
||
- 能讲清 outbox 模式解决了什么问题，为什么接收方要幂等。
|
||
|
||
---
|
||
|
||
## 第 4 天：D 人机协作
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- Agent 必须停下来问人的几种情况：**需求不清楚**（澄清）、**高风险操作**（删除数据、发邮件、付款前确认）、**连续失败**（换模型或人工接管）、**置信度低**。
|
||
- 关键是“停下来”之后的状态：流程要**持久化地挂起**（状态 = 等待用户），用户可能几小时后才回复，回复后从挂起点继续，而不是从头再来。
|
||
- 第 12 周会用 LangGraph 的 `interrupt()` + `Command(resume=...)` 实现；今天先用最朴素的方式理解。
|
||
- 换模型恢复：失败时记录“失败在哪一步、错误码”，换模型后**只重跑失败的那一步**。
|
||
- 案例：
|
||
|
||
```python
|
||
import json
|
||
import sqlite3
|
||
|
||
db = sqlite3.connect(":memory:")
|
||
db.execute("CREATE TABLE runs (id TEXT PRIMARY KEY, state TEXT, data TEXT)")
|
||
|
||
|
||
def save(run_id: str, state: str, data: dict) -> None:
|
||
db.execute("INSERT OR REPLACE INTO runs VALUES (?, ?, ?)", (run_id, state, json.dumps(data, ensure_ascii=False)))
|
||
db.commit()
|
||
|
||
|
||
def load(run_id: str) -> tuple[str, dict]:
|
||
state, data = db.execute("SELECT state, data FROM runs WHERE id = ?", (run_id,)).fetchone()
|
||
return state, json.loads(data)
|
||
|
||
|
||
def call_model(model: str, prompt: str) -> str:
|
||
if model == "model-a":
|
||
raise RuntimeError("QUOTA_EXCEEDED") # 模拟额度不足
|
||
return f"[{model}] 对“{prompt}”的分析"
|
||
|
||
|
||
def step(run_id: str, user_reply: str | None = None, model: str = "model-a") -> None:
|
||
state, data = load(run_id)
|
||
if state == "new":
|
||
if len(data["request"]) < 6: # 需求太短，需要澄清
|
||
save(run_id, "waiting_human", {**data, "question": "请补充：目标用户是谁？预算多少？"})
|
||
return
|
||
state = "analyzing"
|
||
if state == "waiting_human":
|
||
if user_reply is None:
|
||
return # 还没回复，保持挂起
|
||
data["request"] += f"（补充：{user_reply}）"
|
||
state = "analyzing"
|
||
if state in ("analyzing", "failed"):
|
||
try:
|
||
data["result"] = call_model(model, data["request"])
|
||
save(run_id, "done", data)
|
||
except RuntimeError as e:
|
||
save(run_id, "failed", {**data, "error": str(e), "failed_step": "analyzing"})
|
||
|
||
|
||
save("r1", "new", {"request": "做个网站"})
|
||
step("r1")
|
||
print(load("r1")) # 挂起，等待澄清
|
||
step("r1", user_reply="面向中小企业，预算 5 万")
|
||
print(load("r1")[0], load("r1")[1].get("error")) # 额度不足，失败
|
||
step("r1", model="model-b") # 换模型，只重跑失败的一步
|
||
print(load("r1"))
|
||
```
|
||
|
||
### 要建的文件
|
||
|
||
- `notes/experiments/d_human_in_loop.py`、`notes/cards/D-human-in-loop.md`
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
python3 notes/experiments/d_human_in_loop.py
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 在公司环境走一遍“需要澄清”或“换模型失败 → 恢复”，记录当时的表现：挂起状态存在哪？换模型后是从头开始还是从失败点继续？
|
||
|
||
---
|
||
|
||
## 第 5 天：E 与业务系统集成
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- 模型的输出最终要变成**业务记录**（任务、报告、工单），这一步要求：
|
||
1. **结构化**：用 JSON Schema / pydantic 校验，不合格就重试或报错，不能把自由文本直接塞进业务表；
|
||
2. **幂等**：网络重试、消息重复投递会导致同一结果写两次——用“幂等键”（如 `run_id + step`）加唯一约束；
|
||
3. **可审计**：记录是哪个模型、哪个版本的提示词、哪次运行产生的；
|
||
4. **权限隔离**：Agent 只能写它有权限的业务对象（第 3 周的“查询都带 board_id”是同一个道理）。
|
||
- 案例：
|
||
|
||
```python
|
||
import json
|
||
import sqlite3
|
||
|
||
db = sqlite3.connect(":memory:")
|
||
db.execute("""CREATE TABLE reports (
|
||
id INTEGER PRIMARY KEY, idem_key TEXT UNIQUE NOT NULL,
|
||
title TEXT NOT NULL, risk TEXT NOT NULL CHECK (risk IN ('low','medium','high')),
|
||
model TEXT NOT NULL, prompt_version TEXT NOT NULL, run_id TEXT NOT NULL)""")
|
||
|
||
|
||
def parse_report(raw: str) -> dict:
|
||
data = json.loads(raw)
|
||
if not isinstance(data.get("title"), str) or data.get("risk") not in ("low", "medium", "high"):
|
||
raise ValueError(f"模型输出不符合约定：{raw}")
|
||
return data
|
||
|
||
|
||
def save_report(run_id: str, step: str, raw: str, model: str, prompt_version: str) -> str:
|
||
data = parse_report(raw)
|
||
try:
|
||
with db:
|
||
db.execute(
|
||
"INSERT INTO reports (idem_key, title, risk, model, prompt_version, run_id) VALUES (?, ?, ?, ?, ?, ?)",
|
||
(f"{run_id}:{step}", data["title"], data["risk"], model, prompt_version, run_id),
|
||
)
|
||
return "created"
|
||
except sqlite3.IntegrityError:
|
||
return "duplicate-ignored" # 重复投递：幂等键已存在，直接忽略
|
||
|
||
|
||
raw = '{"title": "缓存方案评审", "risk": "medium"}'
|
||
print(save_report("run-1", "summary", raw, "model-b", "summary@v3"))
|
||
print(save_report("run-1", "summary", raw, "model-b", "summary@v3")) # 重试 / 重复投递
|
||
try:
|
||
save_report("run-2", "summary", '{"title": "x", "risk": "unknown"}', "model-b", "summary@v3")
|
||
except ValueError as e:
|
||
print("拒绝写入：", e)
|
||
print(db.execute("SELECT idem_key, title, risk, model, prompt_version FROM reports").fetchall())
|
||
```
|
||
|
||
### 要建的文件
|
||
|
||
- `notes/experiments/e_integration.py`、`notes/cards/E-integration.md`（可以和 B 合并，但五个重点都要覆盖）
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
python3 notes/experiments/e_integration.py
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 跟一次公司里的过程报告 / 任务状态生成：LLM 结果怎样变成业务记录？有没有校验、幂等、审计字段？写进卡片。
|
||
|
||
---
|
||
|
||
## 第 6 天：补写和打磨案例卡
|
||
|
||
### 学习内容
|
||
|
||
- 说明：好的案例卡要能在面试里用 1 分钟讲完，结构就是 **STAR**（情境、任务、行动、结果）的压缩版。检查每张卡：
|
||
- “现象”是否具体（用户看到了什么）？
|
||
- “根因”有没有分两层（直接原因 + 系统设计原因）？
|
||
- “可迁移结论”是否一句话、离开公司背景也成立？
|
||
- “若重做的设计”是否能在项目三里实现？
|
||
- 有没有任何真实名称、路径、数据？
|
||
- 优先写成卡的题目（来自你已经遇到过的问题）：
|
||
1. 关掉浏览器后，多轮研讨就停了（编排跑在前端）——C / A
|
||
2. 分析已经进行到第三步，但任务状态还是“未开始”（权威源 / 回写缺失）——B / E
|
||
3. 模型额度或鉴权失败，以及“换模型后恢复”——D
|
||
4. 主分支升级后某张表缺字段导致接口 500（应用依赖和数据库迁移）——可以作为第 6 张“工程化”卡（迁移要随代码一起发布，第 2 周学过）
|
||
|
||
### 要建的文件
|
||
|
||
- 修改 5 张卡；如果写得动，补第 6 张迁移卡
|
||
|
||
### 运行方式
|
||
|
||
- 每张卡对着手机讲一遍，计时 ≤ 1 分钟。
|
||
|
||
### 完成标准
|
||
|
||
- 每张卡都能 1 分钟讲完；脱敏检查脚本通过。
|
||
|
||
---
|
||
|
||
## 第 7 天：汇总设计输入
|
||
|
||
### 学习内容
|
||
|
||
- 说明：把五张卡的“若重做的设计”汇总成一张表，这就是项目三（第 10~13 周）的**需求清单**，每一项都要在项目里有对应实现。
|
||
- 案例：
|
||
|
||
```markdown
|
||
| 来源卡片 | 若重做的设计 | 项目三里的实现 | 周 |
|
||
| --- | --- | --- | --- |
|
||
| A | 服务端编排 + 结束条件（共识 / 轮数 / 成本） | LangGraph 条件边 + max_rounds | 11 |
|
||
| A | 同一任务只有一个执行者 | runs 表 + 租约（lease） | 12 |
|
||
| C | 执行与推送分离，断线补发 | 后台作业 + 事件表 + SSE Last-Event-ID | 12 |
|
||
| B | 单一权威源 | 状态只在 checkpointer / runs 表，前端只读 | 12 |
|
||
| D | 持久化挂起，回复后继续 | interrupt() + Command(resume=...) | 12 |
|
||
| D | 换模型只重跑失败步骤 | 从检查点恢复 + 模型配置可切换 | 12 |
|
||
| E | 结构化输出 + 幂等 + 审计 | pydantic 校验 + 幂等键 + 记录模型和提示词版本 | 11、13 |
|
||
```
|
||
|
||
### 要建的文件
|
||
|
||
- `notes/cards/README.md`；在总纲进度表勾掉第 9 周
|
||
|
||
### 完成标准
|
||
|
||
- [ ] A~E 每个重点至少一张案例卡
|
||
- [ ] 每张卡都写了“若重做，我会怎样设计”
|
||
- [ ] 设计输入表完成，每一项都对应到第 10~13 周
|
||
|
||
---
|
||
|
||
<!-- week-topics:start -->
|
||
|
||
## 本周相关专题
|
||
|
||
> 本节由 `tools/gen_week_topics.py` 根据专题文件的 `week` 字段自动生成，请勿手改；想调整就修改对应专题文件的 frontmatter 后重新生成。
|
||
|
||
- **AI 编程练习（每周）**：本周至少一个功能用 AI 编程工具（Cursor / Claude Code / Codex 等）完成，按 [AI 编程：工作流](../topics/ai/ai-coding/01-workflow-rules-skills.md) 写清需求和规则，按 [验证与证据](../topics/ai/ai-coding/02-verify-evidence.md) 保存测试结果和会话记录（脱敏后），第 15 周整理成“AI 编程 + 验证方法”的应聘案例。
|
||
|
||
<!-- week-topics:end -->
|
||
|
||
## 本周自查清单
|
||
|
||
- [ ] 5 个实验脚本都跑过，能讲清每个实验演示了什么
|
||
- [ ] 5 张案例卡（脱敏），每张 1 分钟能讲完
|
||
- [ ] 设计输入表
|
||
- [ ] 能回答：“为什么编排不该只活在浏览器里？”“权威源是什么？”“outbox 模式解决什么问题？”“Agent 什么时候要停下来问人？”“LLM 结果怎么安全地写进业务系统？”
|