DEV Community

llimage
llimage

Posted on

规格驱动 + 自动门控:用 FROST-SOP 写一个 Agent 产出质量守门员

规格驱动 + 自动门控:用 FROST-SOP 写一个 Agent 产出质量守门员

一、你的 Agent 产出,真的"合格"了吗?

如果你在用 Agent 做实际项目,下面这些场景一定不陌生:

  • 写了一篇推广文章,发出去才发现有事实错误
  • 代码改完上线,测试发现老功能悄悄坏掉了
  • Agent 生成的报告看起来很专业,但关键数据对不上
  • 每次交付前都要人工反复检查,效率还不如自己写

问题出在哪?你缺少一道"质量闸门"。

传统开发有单元测试、集成测试、CI/CD 流水线——每一层都是一道闸门,不合格的代码根本进不了生产环境。但到了 Agent 时代,很多人却把"模型输出"直接当"最终产物",中间没有任何校验环节。

结果就是:质量完全靠运气,时好时坏。

这篇文章,我要带你用 FROST-SOP 的工程化思路,从零搭建一个 Spec-Gate(规格门控)系统——让 Agent 的每一份产出,都必须经过自动化的质量检验,不合格就打回重做。

所有代码都可以直接运行,而且是我们在 GOAI 大赛和日常项目中真实在用的方案。


二、核心思路:Spec + Gate = 可验证的质量

在写代码之前,先理解两个核心概念。

2.1 Spec(规格):定义"什么是合格"

Spec 就是一份结构化的合格标准说明书。它不是模糊的"写得好一点",而是精确的、可检查的约束条件。

举个例子,一篇技术文章的 Spec 可能长这样:

检查项 标准 检查方式
字数 1500-3000 字 数值校验
代码示例 至少 2 段 结构校验
事实准确性 无虚构数据 LLM 校验
格式规范 Markdown 合法 语法校验
外链有效性 所有链接可访问 网络校验

Spec 的精髓:把模糊的"质量要求",翻译成可执行的检查规则。

2.2 Gate(门控):执行"合不合格"的判定

Gate 是 Spec 的执行者——它读取规格说明书,然后一条一条去检查产出物,最后给出一个"通过 / 不通过"的判定。

在 FROST-SOP 的体系里,Gate 有三种典型形态:

Gate 类型 特点 适用场景
硬规则 Gate 程序精确判定,零误差 字数、格式、语法、结构
语义 Gate LLM 辅助判断,有一定误差 逻辑完整性、事实准确性
人工 Gate 人类最终兜底 高风险决策、创意评估

三层 Gate 叠加,就是一条完整的质量流水线:硬规则快速过滤 → 语义 Gate 深度检查 → 关键节点人工确认。


三、实战第一步:写一个最简单的硬规则 Gate

让我们从最简单的版本开始——一个检查文章字数和格式的硬规则 Gate。

3.1 定义规格

# spec_gate/specs.py
from dataclasses import dataclass, field
from typing import List, Callable, Optional

@dataclass
class CheckRule:
    """一条检查规则"""
    name: str                    # 规则名称
    description: str             # 规则描述
    check_fn: Callable           # 检查函数
    severity: str = "error"      # 严重程度:error / warning
    weight: float = 1.0          # 权重(用于评分)

@dataclass
class Spec:
    """规格说明书"""
    name: str
    rules: List[CheckRule] = field(default_factory=list)

    def add_rule(self, rule: CheckRule):
        self.rules.append(rule)
        return self
Enter fullscreen mode Exit fullscreen mode

3.2 实现基础检查函数

# spec_gate/checks.py
import re
from typing import Tuple

def check_word_count(text: str, min_words: int = 1500, max_words: int = 3000) -> Tuple[bool, str]:
    """检查字数是否在范围内"""
    # 中文按字符数,英文按单词数,混合估算
    chinese_chars = len(re.findall(r'[\u4e00-\u9fa5]', text))
    english_words = len(re.findall(r'[a-zA-Z]+', text))
    total_estimate = chinese_chars + english_words

    if total_estimate < min_words:
        return False, f"字数不足:{total_estimate}(最低要求 {min_words}"
    elif total_estimate > max_words:
        return False, f"字数超标:{total_estimate}(最高限制 {max_words}"
    return True, f"字数合规:{total_estimate}"

def check_code_blocks(text: str, min_count: int = 2) -> Tuple[bool, str]:
    """检查代码块数量"""
    code_blocks = re.findall(r'```

[\s\S]*?

```', text)
    count = len(code_blocks)
    if count < min_count:
        return False, f"代码块不足:{count} 个(至少 {min_count} 个)"
    return True, f"代码块数量:{count}"

def check_markdown_headings(text: str, min_level: int = 2) -> Tuple[bool, str]:
    """检查是否有合理的标题层级"""
    h1_count = len(re.findall(r'^# ', text, re.MULTILINE))
    h2_count = len(re.findall(r'^## ', text, re.MULTILINE))

    if h1_count != 1:
        return False, f"H1 标题数量异常:{h1_count} 个(应为 1 个)"
    if h2_count < min_level:
        return False, f"二级标题不足:{h2_count} 个(至少 {min_level} 个)"
    return True, f"标题结构正常:{h1_count} 个 H1,{h2_count} 个 H2"

def check_external_links(text: str) -> Tuple[bool, str]:
    """检查外链格式(不验证可访问性,那是网络检查的事)"""
    # 匹配 Markdown 链接格式
    links = re.findall(r'\[([^\]]+)\]\(([^)]+)\)', text)
    bad_links = []
    for text_display, url in links:
        if not url.startswith(('http://', 'https://')):
            bad_links.append(f"{text_display} -> {url}")

    if bad_links:
        return False, f"无效链接格式:{', '.join(bad_links)}"
    return True, f"链接格式正常,共 {len(links)}"
Enter fullscreen mode Exit fullscreen mode

3.3 Gate 执行器

# spec_gate/gate.py
from dataclasses import dataclass
from typing import List, Dict, Any
from .specs import Spec, CheckRule

@dataclass
class CheckResult:
    """单条规则的检查结果"""
    rule_name: str
    passed: bool
    message: str
    severity: str
    weight: float

@dataclass 
class GateResult:
    """Gate 的整体结果"""
    spec_name: str
    total_rules: int
    passed_count: int
    failed_count: int
    score: float            # 0-100 分
    passed: bool            # 是否整体通过
    details: List[CheckResult]

    def summary(self) -> str:
        passed_rules = [r for r in self.details if r.passed]
        failed_rules = [r for r in self.details if not r.passed]
        lines = [
            f"📋 {self.spec_name} 检查报告",
            f"   总分:{self.score:.1f}/100 | 通过:{len(passed_rules)}/{self.total_rules}",
        ]
        if failed_rules:
            lines.append("   ❌ 未通过项:")
            for r in failed_rules:
                icon = "🔴" if r.severity == "error" else "🟡"
                lines.append(f"   {icon} {r.rule_name}: {r.message}")
        else:
            lines.append("   ✅ 全部通过!")
        return "\n".join(lines)


class SpecGate:
    """规格门控执行器"""

    def __init__(self, spec: Spec, pass_score: float = 80.0):
        self.spec = spec
        self.pass_score = pass_score

    def run(self, target: Any) -> GateResult:
        """对目标执行所有检查"""
        results = []
        total_weight = 0
        earned_weight = 0

        for rule in self.spec.rules:
            try:
                passed, message = rule.check_fn(target)
            except Exception as e:
                passed, message = False, f"检查执行出错:{str(e)}"

            total_weight += rule.weight
            if passed:
                earned_weight += rule.weight

            results.append(CheckResult(
                rule_name=rule.name,
                passed=passed,
                message=message,
                severity=rule.severity,
                weight=rule.weight,
            ))

        score = (earned_weight / total_weight * 100) if total_weight > 0 else 0
        has_critical_failure = any(
            not r.passed and r.severity == "error" 
            for r in results
        )

        return GateResult(
            spec_name=self.spec.name,
            total_rules=len(self.spec.rules),
            passed_count=sum(1 for r in results if r.passed),
            failed_count=sum(1 for r in results if not r.passed),
            score=score,
            passed=score >= self.pass_score and not has_critical_failure,
            details=results,
        )
Enter fullscreen mode Exit fullscreen mode

3.4 组装第一个 Gate

# example_article_gate.py
from spec_gate.specs import Spec, CheckRule
from spec_gate.gate import SpecGate
from spec_gate.checks import (
    check_word_count, check_code_blocks,
    check_markdown_headings, check_external_links,
)

# 定义一篇技术文章的质量规格
article_spec = Spec(name="技术文章质量规格")
article_spec.add_rule(CheckRule(
    name="字数检查",
    description="文章字数应在 1500-3000 字之间",
    check_fn=lambda t: check_word_count(t, 1500, 3000),
    severity="error",
    weight=2.0,
))
article_spec.add_rule(CheckRule(
    name="代码示例", 
    description="至少包含 2 段代码示例",
    check_fn=lambda t: check_code_blocks(t, min_count=2),
    severity="error",
    weight=2.0,
))
article_spec.add_rule(CheckRule(
    name="标题结构",
    description="应有合理的 Markdown 标题层级",
    check_fn=check_markdown_headings,
    severity="warning",
    weight=1.0,
))
article_spec.add_rule(CheckRule(
    name="链接格式",
    description="所有外链格式正确",
    check_fn=check_external_links,
    severity="warning",
    weight=1.0,
))

# 创建 Gate 并执行检查
def check_article(markdown_text: str):
    gate = SpecGate(article_spec, pass_score=80.0)
    result = gate.run(markdown_text)
    print(result.summary())
    return result

if __name__ == "__main__":
    # 测试:用本文当输入
    with open(__file__.replace("example_article_gate.py", 
                               "20260820_周四_代码教程_SpecGate质量守门员实战.md"), 
              "r", encoding="utf-8") as f:
        test_text = f.read()
    check_article(test_text)
Enter fullscreen mode Exit fullscreen mode

四、实战第二步:加入 LLM 语义 Gate

硬规则 Gate 能搞定格式和结构,但判断不了"内容质量"。比如一篇文章结构完整、字数达标,但逻辑混乱、事实错误——硬规则查不出来。

这时候就需要语义 Gate:让 LLM 来做内容层面的审查。

4.1 语义检查器

# spec_gate/semantic_checks.py
import json
from typing import Tuple, List

class SemanticChecker:
    """基于 LLM 的语义检查器"""

    def __init__(self, llm_client):
        self.llm = llm_client

    def check_factual_accuracy(self, text: str, context: dict = None) -> Tuple[bool, str]:
        """检查事实准确性:是否有明显的虚构或错误陈述"""
        prompt = f"""
你是一个事实核查员。请检查以下文章中是否有明显的事实错误或虚构数据。

检查要点:
1. 是否有明确给出但无法验证的具体数字?
2. 是否有不符合常识的技术断言?
3. 是否有虚构的产品、公司或人物名称(除非明确标注为示例)?

文章内容:
---
{text[:3000]}
---

请以 JSON 格式返回:
{{
    "passed": true/false,
    "issues": ["问题1", "问题2"...],
    "summary": "简要总结"
}}
"""
        try:
            response = self.llm.chat(prompt)
            result = json.loads(response)
            if result.get("passed"):
                return True, result.get("summary", "事实核查通过")
            else:
                issues = result.get("issues", [])
                return False, f"事实核查发现 {len(issues)} 个问题:{'; '.join(issues[:3])}"
        except Exception as e:
            # 语义检查失败不应该阻塞流程,降级为 warning
            return True, f"语义检查执行异常(已跳过):{str(e)}"

    def check_logical_completeness(self, text: str, structure_requirements: List[str]) -> Tuple[bool, str]:
        """检查逻辑完整性:是否覆盖了要求的所有要点"""
        requirements_str = "\n".join(f"- {r}" for r in structure_requirements)
        prompt = f"""
请检查以下文章是否覆盖了以下所有要点:
{requirements_str}

文章内容(前3000字):
---
{text[:3000]}
---

请以 JSON 格式返回:
{{
    "passed": true/false,
    "covered": ["已覆盖的要点"...],
    "missing": ["缺失的要点"...],
    "summary": "简要总结"
}}
"""
        try:
            response = self.llm.chat(prompt)
            result = json.loads(response)
            if result.get("passed"):
                return True, f"逻辑完整,覆盖全部 {len(structure_requirements)} 个要点"
            else:
                missing = result.get("missing", [])
                return False, f"缺失 {len(missing)} 个要点:{'; '.join(missing)}"
        except Exception as e:
            return True, f"逻辑检查执行异常(已跳过):{str(e)}"
Enter fullscreen mode Exit fullscreen mode

4.2 集成到 Gate 系统

# example_full_gate.py
from spec_gate.specs import Spec, CheckRule
from spec_gate.gate import SpecGate
from spec_gate.checks import check_word_count, check_code_blocks
from spec_gate.semantic_checks import SemanticChecker

# 假设你有一个 LLM 客户端
# from your_llm import llm_client
# semantic = SemanticChecker(llm_client)

# 扩展规格:加入语义检查
full_article_spec = Spec(name="技术文章完整质量规格(含语义)")

# 硬规则层
full_article_spec.add_rule(CheckRule(
    name="字数检查",
    description="1500-3000 字",
    check_fn=lambda t: check_word_count(t, 1500, 3000),
    severity="error", weight=2.0,
))
full_article_spec.add_rule(CheckRule(
    name="代码示例",
    description="至少 2 段代码",
    check_fn=lambda t: check_code_blocks(t, 2),
    severity="error", weight=2.0,
))

# 语义层(如果有 LLM 客户端的话)
# full_article_spec.add_rule(CheckRule(
#     name="事实准确性",
#     description="无明显事实错误",
#     check_fn=semantic.check_factual_accuracy,
#     severity="error", weight=3.0,
# ))
# full_article_spec.add_rule(CheckRule(
#     name="逻辑完整性",
#     description="覆盖核心要点",
#     check_fn=lambda t: semantic.check_logical_completeness(
#         t, ["问题引入", "核心概念", "代码实战", "总结展望"]
#     ),
#     severity="warning", weight=2.0,
# ))
Enter fullscreen mode Exit fullscreen mode

五、实战第三步:接入 FROST-SOP 工作流

单独的 Gate 只是一个检查工具,真正的威力在于把它嵌入到工作流里,让不合格的产出自动回炉重造

这就是 FROST-SOP 的价值:SOP 定义流程,Gate 在流程节点上做质量把关。

5.1 带 Gate 的 SOP 执行器

# spec_gate/gated_sop.py
from typing import List, Callable, Dict, Any
from .gate import SpecGate, GateResult

class GatedStep:
    """带门控的步骤"""
    def __init__(self, name: str, action: Callable, gate: SpecGate = None, 
                 max_retries: int = 3):
        self.name = name
        self.action = action
        self.gate = gate
        self.max_retries = max_retries

class GatedSOP:
    """带门控的 SOP 执行器"""

    def __init__(self, name: str):
        self.name = name
        self.steps: List[GatedStep] = []
        self.execution_log = []

    def add_step(self, step: GatedStep):
        self.steps.append(step)
        return self

    def run(self, context: Dict[str, Any]) -> Dict[str, Any]:
        """执行完整的 SOP,每一步都经过 Gate 检查"""
        self.execution_log = []

        for i, step in enumerate(self.steps):
            step_log = {
                "step": step.name,
                "step_index": i,
                "retries": 0,
                "gate_result": None,
                "status": "pending",
            }

            for attempt in range(step.max_retries):
                # 执行动作
                try:
                    context = step.action(context)
                except Exception as e:
                    step_log["status"] = "action_error"
                    step_log["error"] = str(e)
                    self.execution_log.append(step_log)
                    raise

                # 如果没有 Gate,直接通过
                if step.gate is None:
                    step_log["status"] = "passed_no_gate"
                    step_log["retries"] = attempt
                    break

                # 执行 Gate 检查
                gate_result = step.gate.run(context)
                step_log["gate_result"] = gate_result
                step_log["retries"] = attempt + 1

                if gate_result.passed:
                    step_log["status"] = "passed"
                    break
                else:
                    # 未通过,把检查结果写入 context,供下一步重试参考
                    context["gate_feedback"] = gate_result
                    step_log["status"] = f"retry_{attempt+1}"

            # 重试耗尽仍未通过
            if step.gate is not None and step_log["status"].startswith("retry_"):
                step_log["status"] = "failed_after_retries"
                self.execution_log.append(step_log)
                raise RuntimeError(
                    f"步骤「{step.name}」经过 {step.max_retries} 次重试仍未通过 Gate 检查\n"
                    f"{gate_result.summary() if gate_result else ''}"
                )

            self.execution_log.append(step_log)

        context["sop_execution_log"] = self.execution_log
        return context

    def get_execution_report(self) -> str:
        """生成执行审计报告"""
        lines = [f"📋 SOP 执行报告:{self.name}", "-" * 40]
        for log in self.execution_log:
            status_icon = {
                "passed": "",
                "passed_no_gate": "",
                "failed_after_retries": "",
                "action_error": "💥",
            }.get(log["status"], "🔄")

            lines.append(f"{status_icon} 步骤 {log['step_index']+1}: {log['step']}")
            lines.append(f"   状态:{log['status']} | 重试次数:{log['retries']}")

            if log.get("gate_result"):
                gr = log["gate_result"]
                lines.append(f"   Gate 得分:{gr.score:.1f}/100")
                if not gr.passed:
                    failed = [r for r in gr.details if not r.passed]
                    for f in failed[:3]:
                        lines.append(f"{f.rule_name}: {f.message}")
            lines.append("")

        return "\n".join(lines)
Enter fullscreen mode Exit fullscreen mode

5.2 一个完整的"文章生成 + 质检"流水线

# example_article_pipeline.py
from spec_gate.gated_sop import GatedSOP, GatedStep
from spec_gate.gate import SpecGate
from spec_gate.specs import Spec, CheckRule
from spec_gate.checks import check_word_count, check_code_blocks

# 1. 定义各步骤的动作
def generate_outline(context: dict) -> dict:
    """生成文章大纲"""
    topic = context["topic"]
    context["outline"] = f"{topic}》大纲:\n1. 问题引入\n2. 核心概念\n3. 代码实战\n4. 总结展望"
    print(f"✍️  生成大纲完成")
    return context

def write_draft(context: dict) -> dict:
    """撰写初稿"""
    outline = context["outline"]
    # 实际场景这里会调用 LLM 写文章
    context["draft"] = f"""# {context['topic']}

## 一、问题引入

这是一个很重要的问题...

## 二、核心概念

核心思想是...

Enter fullscreen mode Exit fullscreen mode


python

示例代码

def hello():
print("Hello FROST")


## 三、代码实战

下面我们来实现...

Enter fullscreen mode Exit fullscreen mode


python

核心实现

class Agent:
def run(self, task):
return task


## 四、总结展望

总结一下...
"""
    print(f"✍️  初稿撰写完成")
    return context

def polish_article(context: dict) -> dict:
    """润色文章(根据 Gate 反馈优化)"""
    feedback = context.get("gate_feedback")
    draft = context["draft"]

    if feedback and not feedback.passed:
        # 根据反馈做针对性优化
        print(f"🔧 根据 Gate 反馈润色文章(得分:{feedback.score:.1f})")
        # 实际场景:把反馈 + 原文一起丢给 LLM 让它修改
        context["draft"] = draft + "\n\n> (经过 Gate 反馈优化后的版本)"
    else:
        print("✨ 无需润色,质量已达标")

    return context

# 2. 定义质量 Gate
quality_spec = Spec(name="文章质量规格")
quality_spec.add_rule(CheckRule(
    name="字数检查",
    description="不少于 200 字(演示用低值)",
    check_fn=lambda c: check_word_count(c.get("draft", ""), 200, 5000),
    severity="error", weight=2.0,
))
quality_spec.add_rule(CheckRule(
    name="代码示例",
    description="至少 2 段代码",
    check_fn=lambda c: check_code_blocks(c.get("draft", ""), 2),
    severity="error", weight=2.0,
))

quality_gate = SpecGate(quality_spec, pass_score=80.0)

# 3. 组装带门控的 SOP
pipeline = GatedSOP(name="文章生成流水线")
pipeline.add_step(GatedStep(
    name="生成大纲",
    action=generate_outline,
    max_retries=1,
))
pipeline.add_step(GatedStep(
    name="撰写初稿", 
    action=write_draft,
    gate=quality_gate,
    max_retries=3,
))
pipeline.add_step(GatedStep(
    name="润色优化",
    action=polish_article,
    gate=quality_gate,
    max_retries=2,
))

# 4. 运行!
if __name__ == "__main__":
    context = {"topic": "Spec-Gate 质量守门员实战教程"}
    result = pipeline.run(context)
    print("\n" + pipeline.get_execution_report())
Enter fullscreen mode Exit fullscreen mode


plaintext

运行输出大概长这样:

✍️  生成大纲完成
✍️  初稿撰写完成
🔧 根据 Gate 反馈润色文章(得分:65.0)
✨ 无需润色,质量已达标

📋 SOP 执行报告:文章生成流水线
----------------------------------------
✅ 步骤 1: 生成大纲
   状态:passed_no_gate | 重试次数:0

✅ 步骤 2: 撰写初稿
   状态:passed | 重试次数:1
   Gate 得分:72.5/100
     ✗ 字数检查:字数不足:180(最低要求 200)

✅ 步骤 3: 润色优化
   状态:passed | 重试次数:1
   Gate 得分:100.0/100
Enter fullscreen mode Exit fullscreen mode

六、这套系统的核心价值:从"靠人"到"靠规则"

你可能会说:不就是加了几个 if 判断吗?至于搞得这么复杂?

我想说,结构比功能重要。这套系统的真正价值不是那些检查函数,而是它引入了三个根本性的变化:

6.1 质量标准从"隐性"变成"显性"

以前:"文章质量要高一点"——什么叫高?谁来定义?凭感觉。

现在:所有质量标准都写在 Spec 里,可量化、可讨论、可迭代。团队成员对"什么是好"有了共识。

6.2 检查从"事后抽检"变成"每步必检"

以前:写完了才发现有问题,返工成本极高。

现在:每一个步骤都有 Gate 把守,问题在产生的那一刻就被拦截了。早发现,早修复,成本最低。

6.3 优化从"凭经验"变成"数据驱动"

每一次 Gate 检查的结果都是数据——哪条规则经常失败?哪个步骤重试最多?平均得分多少?

积累一段时间后,你就能精准地看到:瓶颈在哪里,应该优先优化哪里。


七、扩展方向:从文章到一切可验证的产出

这套 Spec-Gate 框架不止能用来检查文章,它可以检查任何可定义规格的产出物

应用场景 Spec 内容 Gate 类型
代码 Review 代码规范、测试覆盖率、安全漏洞 硬规则 + LLM
需求文档 完整性、无歧义性、可测试性 硬规则 + 语义
测试用例 覆盖度、步骤清晰度、预期明确 硬规则 + 语义
设计稿 组件规范、配色规范、布局一致性 视觉 AI
营销文案 合规性、品牌调性、CTA 明确 语义 + 硬规则

本质上,只要你能把"好"定义清楚,Gate 就能帮你守住底线。

而 FROST-SOP 做的事情,就是把这些 Gate 串联成一条完整的流水线——从输入到输出,每一步都有迹可循、有据可查、有闸可守。

这就是为什么我们说 FROST 是思想源头,FROST-SOP 是工程落地:思想告诉你"应该有质量闸门",工程帮你"把闸门真正建起来"。


八、写在最后

回到开头的问题:为什么你的 Agent 产出质量不稳定?

因为你把模型当"员工"用,期待它自觉做好。但正确的做法是把它当"生产线"用——你需要设计好流程,设好闸门,不合格就回炉。

Spec-Gate 的思路不复杂,但它代表了一种思维方式的转变:从"相信模型的能力"到"设计可靠的系统"。

AI 时代的竞争,比的从来不是谁的模型更聪明,而是谁的系统更不犯傻

如果你也在为 Agent 产出质量发愁,不妨从加一道最简单的 Gate 开始。


项目地址:

本周思考题:你现在的项目里,有哪些产出是"凭感觉交付"的?如果给它加一道 Gate,你会检查什么?欢迎在评论区聊聊。

Top comments (0)