4.6 综合案例：从PRD到上线的全自动循环

文档摘要

4.6 综合案例：从 PRD 到上线的全自动循环导读本章将内容生产、DevOps 等单领域循环串联为端到端的全自动交付循环——从 PRD 出发，经过设计、开发、QA、部署，全程由循环引擎编排。这是 Loop Engineering 最有野心的场景，也是六大原语协同工作的终极演练。我们将实现多循环编排框架，并引入 Ralph Loop 模式（prd.json + progress.txt + AGENTS.md）作为工程管理基础设施。

4.6 综合案例：从 PRD 到上线的全自动循环

导读

本章将内容生产、DevOps 等单领域循环串联为端到端的全自动交付循环——从 PRD 出发，经过设计、开发、QA、部署，全程由循环引擎编排。这是 Loop Engineering 最有野心的场景，也是六大原语协同工作的终极演练。我们将实现多循环编排框架，并引入 Ralph Loop 模式（prd.json + progress.txt + AGENTS.md）作为工程管理基础设施。

学习目标

理解端到端交付循环的多循环编排模式
掌握 Ralph Loop 模式的设计与实现
实现从 PRD 到代码到 CI/CD 的全链路自动化
设计多循环之间的状态传递与依赖管理
构建可观测、可中断、可恢复的自动化框架

核心概念

端到端交付循环总览

多循环编排架构

子循环	输入	输出	验证标准	失败动作
设计循环	PRD	架构文档+接口定义	评审通过	回到PRD澄清
开发循环	设计文档	可编译代码+测试	CI全绿	回到开发修复
QA循环	代码	测试报告+覆盖率	通过率≥95%	回到开发修复
部署循环	测试通过代码	线上服务	Canary达标	自动回滚

Ralph Loop 模式

Ralph Loop 是 Loop Engineering 的工程管理基础设施，三文件协作：

prd.json — 结构化需求，机器可解析
progress.txt — 进度追踪，人和机器都能读
AGENTS.md — 循环配置，定义行为规则

环境准备


mkdir -p e2e-loop/state/{design,dev,qa,deploy}
cat > e2e-loop/state/prd.json << 'EOF'
{
  "id": "feat-user-search",
  "title": "用户搜索功能",
  "priority": "P1",
  "description": "新增用户搜索接口，支持按姓名、邮箱、手机号搜索",
  "acceptance_criteria": ["支持模糊搜索", "返回分页结果", "响应时间<200ms"],
  "technical_constraints": {"database": "PostgreSQL", "framework": "FastAPI", "test_coverage_min": 85}
}
EOF

分步实战

步骤一：实现 Ralph Loop 状态管理器


"""e2e_loop/ralph_loop.py — Ralph Loop 三文件状态管理"""
import json, time
from pathlib import Path
from dataclasses import dataclass, field
from enum import Enum

class LoopPhase(Enum):
    PRD_REVIEW = "prd_review"; DESIGN = "design"
    DEVELOPMENT = "development"; QA = "qa"
    DEPLOY = "deploy"; MONITOR = "monitor"; COMPLETE = "complete"

@dataclass
class ProgressEntry:
    phase: LoopPhase; status: str  # pending/in_progress/passed/failed
    message: str = ""; artifacts: list = field(default_factory=list)

@dataclass
class LoopConfig:
    max_design_iterations: int = 3; max_dev_iterations: int = 5
    max_qa_iterations: int = 3; test_coverage_min: float = 85.0

class RalphLoop:
    PHASES = [LoopPhase.PRD_REVIEW, LoopPhase.DESIGN, LoopPhase.DEVELOPMENT,
              LoopPhase.QA, LoopPhase.DEPLOY, LoopPhase.MONITOR, LoopPhase.COMPLETE]

    def __init__(self, state_dir: str = "./state"):
        self.state_dir = Path(state_dir); self.state_dir.mkdir(parents=True, exist_ok=True)
        self.prd_path = self.state_dir / "prd.json"
        self.progress_path = self.state_dir / "progress.txt"
        self.agents_path = self.state_dir / "AGENTS.md"
        self.config = LoopConfig(); self.progress: list[ProgressEntry] = []

    def load_prd(self) -> dict | None:
        return json.loads(self.prd_path.read_text()) if self.prd_path.exists() else None

    def save_prd(self, prd: dict):
        self.prd_path.write_text(json.dumps(prd, ensure_ascii=False, indent=2))

    def update_progress(self, phase: LoopPhase, status: str, msg="", artifacts=None):
        entry = next((e for e in self.progress if e.phase == phase), None)
        if not entry:
            entry = ProgressEntry(phase, status, msg, artifacts or [])
            self.progress.append(entry)
        else:
            entry.status = status; entry.message = msg
            if artifacts: entry.artifacts.extend(artifacts)
        self._persist()

    def _persist(self):
        lines = [f"# {time.strftime('%Y-%m-%d %H:%M:%S')}\n"]
        done = sum(1 for e in self.progress if e.status == "passed")
        total = len(self.PHASES) - 1
        bar = "█" * int(20*done/total) + "░" * (20 - int(20*done/total)) if total else ""
        lines.append(f"进度: [{bar}] {done}/{total}\n")
        icons = {"passed":"✅","in_progress":"🔄","failed":"❌","pending":"⏳"}
        for p in self.PHASES:
            e = next((x for x in self.progress if x.phase == p), None)
            icon = icons.get(e.status, "⏳") if e else "⏳"
            lines.append(f"  {icon} {p.value}: {e.status if e else 'pending'}")
            if e and e.message: lines.append(f"     └─ {e.message}")
            lines.append("")
        self.progress_path.write_text("\n".join(lines))

    def get_current_phase(self) -> LoopPhase:
        for p in self.PHASES:
            e = next((x for x in self.progress if x.phase == p), None)
            if not e or e.status in ("pending","in_progress","failed"): return p
        return LoopPhase.COMPLETE

    def generate_agents_md(self, prd: dict):
        lines = [f"# AGENTS.md — {prd.get('title','')}",
                 f"- 需求ID: {prd.get('id')}", f"- 优先级: {prd.get('priority')}",
                 f"- 最大设计迭代: {self.config.max_design_iterations}",
                 f"- 最大开发迭代: {self.config.max_dev_iterations}",
                 f"- 最低覆盖率: {self.config.test_coverage_min}%"]
        self.agents_path.write_text("\n".join(lines))

步骤二：多循环编排引擎


"""e2e_loop/orchestrator.py — 多循环编排"""
import logging
from dataclasses import dataclass, field
from typing import Callable

@dataclass
class SubLoopResult:
    phase: object; success: bool; message: str = ""
    artifacts: list = field(default_factory=list); should_retry: bool = False

class LoopOrchestrator:
    PHASES = [LoopPhase.DESIGN, LoopPhase.DEVELOPMENT, LoopPhase.QA, LoopPhase.DEPLOY]

    def __init__(self, ralph: RalphLoop):
        self.ralph = ralph; self.loops: dict = {}
        self.max_retries = {LoopPhase.DESIGN: 3, LoopPhase.DEVELOPMENT: 5,
                            LoopPhase.QA: 3, LoopPhase.DEPLOY: 3}

    def register(self, phase, handler): self.loops[phase] = handler

    def run(self) -> dict:
        result = {"success": False, "completed": [], "failed_at": None}
        for phase in self.PHASES:
            handler = self.loops.get(phase)
            if not handler:
                self.ralph.update_progress(phase, "skipped"); continue
            self.ralph.update_progress(phase, "in_progress")
            for attempt in range(1, self.max_retries[phase]+1):
                r = handler(self.ralph)
                if r.success:
                    self.ralph.update_progress(phase, "passed", r.message, r.artifacts)
                    result["completed"].append(phase.value); break
                elif not r.should_retry: break
            else:
                # 超过重试次数，判断回退
                rollback = {LoopPhase.DEVELOPMENT: LoopPhase.DESIGN,
                            LoopPhase.QA: LoopPhase.DEVELOPMENT,
                            LoopPhase.DEPLOY: LoopPhase.QA}.get(phase)
                if rollback:
                    result["failed_at"] = phase.value
                    result["rollback_to"] = rollback.value; break
                else:
                    self.ralph.update_progress(phase, "failed"); break
        else:
            result["success"] = True
            self.ralph.update_progress(LoopPhase.COMPLETE, "passed", "交付完成")
        return result

步骤三：四个子循环实现


"""e2e_loop/sub_loops.py — 四个子循环"""
import json
from pathlib import Path
from ralph_loop import RalphLoop, LoopPhase
from orchestrator import SubLoopResult

def design_loop(ralph: RalphLoop) -> SubLoopResult:
    prd = ralph.load_prd()
    doc = f"# 架构设计: {prd['title']}\n## 概述\n{prd['description']}\n"
    doc += "## 接口设计\n### GET /api/users/search\n- 参数: q, page, size\n"
    doc += "## 数据模型\n- users表: id, name, email, phone\n"
    d = Path(ralph.state_dir)/"design"; d.mkdir(exist_ok=True)
    (d/"architecture.md").write_text(doc)
    spec = {"endpoints": [{"method":"GET","path":"/api/users/search",
            "params":[{"name":"q","type":"string","required":True}]}]}
    (d/"api_spec.json").write_text(json.dumps(spec, indent=2))
    return SubLoopResult(LoopPhase.DESIGN, True, "架构完成",
                        artifacts=["architecture.md","api_spec.json"])

def dev_loop(ralph: RalphLoop) -> SubLoopResult:
    code = 'from fastapi import FastAPI, Query\napp = FastAPI()\n'
    code += '@app.get("/api/users/search")\nasync def search(q: str = Query(...)): return {"users":[]}\n'
    d = Path(ralph.state_dir)/"dev"; d.mkdir(exist_ok=True)
    (d/"main.py").write_text(code)
    (d/"test_main.py").write_text("def test_search(): assert True\n")
    return SubLoopResult(LoopPhase.DEVELOPMENT, True, "代码生成完成",
                        artifacts=["main.py","test_main.py"])

def qa_loop(ralph: RalphLoop) -> SubLoopResult:
    prd = ralph.load_prd()
    min_cov = prd.get("technical_constraints",{}).get("test_coverage_min",80)
    cov = 87.5  # 模拟
    report = f"# 测试报告\n覆盖率: {cov}% (要求:{min_cov}%)\n全部通过 ✅\n"
    d = Path(ralph.state_dir)/"qa"; d.mkdir(exist_ok=True)
    (d/"test_report.md").write_text(report)
    return SubLoopResult(LoopPhase.QA, cov >= min_cov,
                        f"覆盖率{cov}%", ["test_report.md"],
                        should_retry=cov < min_cov)

def deploy_loop(ralph: RalphLoop) -> SubLoopResult:
    report = "# 部署报告\nCanary 10% → 错误率0.8% → 全量发布 ✅\n"
    d = Path(ralph.state_dir)/"deploy"; d.mkdir(exist_ok=True)
    (d/"deploy_report.md").write_text(report)
    return SubLoopResult(LoopPhase.DEPLOY, True, "Canary通过，全量发布",
                        ["deploy_report.md"])

完整示例：端到端框架串联


"""e2e_loop/main.py — 完整的PRD到上线循环"""
from ralph_loop import RalphLoop, LoopConfig, LoopPhase
from orchestrator import LoopOrchestrator
from sub_loops import design_loop, dev_loop, qa_loop, deploy_loop

def main():
    ralph = RalphLoop("./state")
    if not ralph.load_prd():
        ralph.save_prd({"id":"feat-user-search","title":"用户搜索功能",
            "priority":"P1","description":"用户搜索接口",
            "acceptance_criteria":["模糊搜索","分页结果","响应<200ms"],
            "technical_constraints":{"database":"PostgreSQL","framework":"FastAPI","test_coverage_min":85}})
    prd = ralph.load_prd()
    ralph.generate_agents_md(prd)

    orch = LoopOrchestrator(ralph)
    orch.register(LoopPhase.DESIGN, design_loop)
    orch.register(LoopPhase.DEVELOPMENT, dev_loop)
    orch.register(LoopPhase.QA, qa_loop)
    orch.register(LoopPhase.DEPLOY, deploy_loop)

    result = orch.run()
    print(f"成功: {result['success']}")
    print(f"完成: {result['completed']}")
    if result.get("failed_at"): print(f"失败: {result['failed_at']}")
    print(ralph.progress_path.read_text())

if __name__ == "__main__": main()

Claude Code /goal + /loop 集成


# /goal — 一次性目标
GOAL = """
/goal 从PRD到生产上线
目标: feat-user-search 需求全流程自动化交付
验收: 1.架构设计完成 2.代码CI通过 3.Canary通过 4.全量发布
"""

# /loop — 持续循环
LOOP = """
/loop 开发流水线循环
条件: CI不通过时自动修复并重提交
终止: CI通过 或 修复>5次 或 安全漏洞
"""

FAQ

Q1：某阶段卡住怎么办？

三层超时：单阶段超时自动回退；总超时防止资源占用；失败N次自动告警暂停等待人工。progress.txt 实时可查。

Q2：多循环状态如何传递？

通过 Ralph Loop 共享状态层。子循环完成后将产出写入 artifacts，下游通过 ralph.progress[phase].artifacts 读取。物理存储在 state/{phase}/ 目录。

Q3：部分失败如何处理？

将需求拆分为可独立交付的最小单元，每个单元独立循环实例。编排引擎支持并行调度，feature flag 控制上线顺序。

最佳实践与避坑

✅ 最佳实践

PRD结构化先行：JSON Schema约束，确保AI可解析
阶段解耦设计：明确接口契约，避免强耦合
可中断可恢复：progress.txt 支持任意时刻中断恢复
回退策略预设：AGENTS.md预定义回退规则
产出物版本化：Git管理state/目录

❌ 常见避坑

避免一步到位：先自动化单循环，验证后再串联
保留人工审批：PRD审批、安全审查、生产发布留窗口
避免共享可变状态：独立worktree和数据流接口
避免黑盒循环：必须输出可读日志和状态文件
避免忽略成本控制：设置Token预算上限

小结

端到端全自动循环是 Loop Engineering 的终极愿景——软件交付从"人驱动"升级为"循环驱动"。Ralph Loop 提供工程管理基础设施，编排引擎协调四个子循环顺序执行与异常回退。核心理念：结构化需求驱动自动化、可中断可恢复的进度管理、预设回退策略保障稳定、渐进式自动化降低风险。

延伸阅读

《Loop Engineering: The Definitive Guide》— 循环工程理论
Claude Code /goal 和 /loop 文档 — 原生循环指令
OpenClaw TaskFlow — 分布式任务编排
DORA Metrics — 软件交付效率指标

关键词：端到端交付、Ralph Loop、多循环编排、PRD到上线、自动化框架、Loop Engineering、/goal、/loop
难度：⭐⭐⭐⭐⭐ 专家级
阅读时间：约 15 分钟