MBE 12-Factor Agents 生产化实施计划

版本: v2.7
日期: 2026-02-12
基线: MBE v3.2.0
参考: 12-Factor Agents
进度: Phase 1~4 ✅ + Factor 4 全覆盖 ✅ + Line 13 统一评分 ✅ + Phase 5 运维自动化 ✅

一、背景与目标

12-Factor Agents 是面向生产环境 LLM 应用的 12 条工程原则。经过与 MBE 现有架构的逐条对照分析，MBE 在 上下文工程 (Factor 3)、提示词管理 (Factor 2)、控制流所有权 (Factor 8)、小而专注的 Agent (Factor 10) 方面已经做得很好。

本计划聚焦 MBE 尚有改进空间的 5 个 Factor，目标是：

二、现状评估

三、实施路线图

Phase 1 (W1-W2)     Phase 2 (W3-W4)     Phase 3 (W5-W6)     Phase 4 (W7-W8)
┌──────────────┐    ┌──────────────┐    ┌──────────────┐    ┌──────────────┐
│ 决策审计链路  │    │ 暂停/恢复    │    │ Self-Critique│    │ 人工审核     │
│ (Factor 5+12)│───▸│ (Factor 6)   │───▸│ 自愈回路     │───▸│ 标准化       │
│              │    │              │    │ (Factor 9)   │    │ (Factor 7)   │
│ 结构化输出   │    │              │    │              │    │              │
│ (Factor 4)   │    │              │    │              │    │              │
└──────────────┘    └──────────────┘    └──────────────┘    └──────────────┘
   基础层               能力层              质量层              协作层

Phase 1：决策审计链路 + 结构化输出（第 1–2 周）

1.1 目标

• 每次请求的完整决策链路可追溯（专家路由 → 检索 → 生成 → Self-Critique）
• 所有 LLM 决策点输出统一的 Pydantic schema
• 核心引擎重构为无状态函数签名

1.2 数据库迁移：决策审计表

文件: mises-behavior-engine/migrations/013_add_decision_audit.sql

-- 决策审计表：记录每次请求的完整决策链路
CREATE TABLE IF NOT EXISTS decision_audit (
    id              BIGSERIAL PRIMARY KEY,
    session_id      VARCHAR(64) NOT NULL,
    request_id      VARCHAR(64) NOT NULL,       -- 单次请求唯一 ID
    step_name       VARCHAR(64) NOT NULL,       -- 决策步骤名
    step_order      INTEGER NOT NULL,           -- 步骤序号
    input_snapshot  JSONB NOT NULL DEFAULT '{}', -- 输入快照
    output_snapshot JSONB NOT NULL DEFAULT '{}', -- 输出快照（结构化决策结果）
    duration_ms     INTEGER,                    -- 耗时
    status          VARCHAR(16) NOT NULL DEFAULT 'ok', -- ok / error / retry / human_review
    error_detail    TEXT,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_decision_audit_session ON decision_audit(session_id);
CREATE INDEX idx_decision_audit_request ON decision_audit(request_id);
CREATE INDEX idx_decision_audit_created ON decision_audit(created_at);

-- 会话检查点表：支持暂停/恢复（Phase 2 使用）
CREATE TABLE IF NOT EXISTS session_checkpoints (
    id              BIGSERIAL PRIMARY KEY,
    session_id      VARCHAR(64) NOT NULL,
    stage           VARCHAR(32) NOT NULL,       -- uneasiness/desires/paths/assessment/action
    stage_data      JSONB NOT NULL DEFAULT '{}', -- 该阶段的完整状态
    context_window  JSONB NOT NULL DEFAULT '[]', -- 压缩后的上下文窗口
    is_active       BOOLEAN NOT NULL DEFAULT TRUE,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    expires_at      TIMESTAMPTZ
);

CREATE INDEX idx_checkpoint_session ON session_checkpoints(session_id, is_active);

1.3 结构化决策输出 Schema

文件: mises-behavior-engine/src/schemas/decisions.py（新建）

"""
12-Factor Agent Factor 4: Tools are just structured outputs
所有 LLM 决策点的统一输出 Schema
"""
from pydantic import BaseModel, Field
from typing import Optional, List
from enum import Enum
from datetime import datetime


class DecisionType(str, Enum):
    EXPERT_ROUTE = "expert_route"
    KNOWLEDGE_RETRIEVE = "knowledge_retrieve"
    RESPONSE_GENERATE = "response_generate"
    SELF_CRITIQUE = "self_critique"
    HUMAN_REVIEW = "human_review"          # Phase 4
    MISES_STAGE = "mises_stage"


class ExpertRouteDecision(BaseModel):
    """专家路由决策"""
    matched_expert_id: str
    confidence: float = Field(ge=0.0, le=1.0)
    fallback_experts: List[str] = []
    routing_method: str                     # "miras" | "graph" | "keyword"
    scene_detected: Optional[str] = None
    intent_analysis: Optional[str] = None


class KnowledgeRetrieveDecision(BaseModel):
    """知识检索决策"""
    chunks_retrieved: int
    retrieval_method: str                   # "local" | "context" | "global"
    top_chunk_score: float
    graph_enhanced: bool = False
    titans_memory_used: bool = False


class SelfCritiqueDecision(BaseModel):
    """Self-Critique 评估结果"""
    is_valid: bool
    overall_score: float = Field(ge=0.0, le=1.0)
    module_scores: dict = {}               # {module_name: score}
    issues_found: List[str] = []
    retry_suggested: bool = False
    retry_guidance: Optional[str] = None   # 给 LLM 的重试指导


class MisesStageDecision(BaseModel):
    """五步行为分析阶段决策"""
    stage: str
    stage_output: dict = {}
    next_stage: Optional[str] = None
    is_complete: bool = False
    pause_requested: bool = False          # Phase 2


class DecisionRecord(BaseModel):
    """统一决策记录（写入 decision_audit 表）"""
    request_id: str
    session_id: str
    step_name: str
    step_order: int
    decision_type: DecisionType
    input_snapshot: dict = {}
    output_snapshot: dict = {}
    duration_ms: Optional[int] = None
    status: str = "ok"
    error_detail: Optional[str] = None
    created_at: datetime = Field(default_factory=datetime.utcnow)

1.4 核心引擎重构：无状态签名

文件: mises-behavior-engine/src/core/engine.py

重构 MisesEngine.process() 为纯函数签名，所有状态通过参数传入/传出：

# === 重构前 (当前) ===
class MisesEngine:
    async def process(self, question: str, session_id: str, ...) -> str:
        # 内部从 Redis 读取 session state
        # 内部修改 self 上的状态
        # 返回字符串

# === 重构后 ===
class MisesEngine:
    async def process(
        self,
        question: str,
        session_state: SessionState,           # 显式传入
        context: Optional[dict] = None,        # 预取的上下文
    ) -> EngineResult:
        """
        无状态归约器：new_state = engine(old_state, event)
        引擎本身不读写外部状态，所有 I/O 由调用方负责
        """
        pass


class EngineResult(BaseModel):
    """引擎处理结果"""
    response: str
    updated_session: SessionState
    decisions: List[DecisionRecord]            # 完整决策链路
    side_effects: List[SideEffect] = []        # 需要调用方执行的副作用

改造策略：渐进式，不一步到位

1. 第 1 周：先加 DecisionRecord 记录层，不改现有流程

   • 在 engine.py 各决策点插入审计记录
   • 在 expert_router.py 返回 ExpertRouteDecision
   • 在 self_critique.py 返回 SelfCritiqueDecision

2. 第 2 周：提取 EngineResult 返回值，将副作用显式化

   • TITANS 编码、HOPE 学习更新等作为 SideEffect 返回
   • 调用方（mcp/server.py）负责执行副作用

1.5 具体改动清单

1.6 验收标准

• 每次 /mcp/analyze 请求在 decision_audit 表产生 3–6 条记录
• 可通过 session_id 查询完整决策链路
• 决策记录包含耗时、状态、输入输出快照
• 现有功能无退化（所有现有测试通过）

Phase 2：暂停/恢复机制（第 3–4 周）

2.1 目标

• 五步行为分析支持跨会话暂停/恢复
• 用户可以在任意阶段暂停，数天后继续
• 法律咨询等长周期场景可用

2.2 设计

用户提问 → 不舒适识别 → [检查点1]
                          ↓
         愿望分析 → [检查点2]
                      ↓
         路径生成 → [检查点3] ← 用户可在此暂停，3天后恢复
                      ↓
         可行性评估 → [检查点4]
                       ↓
         行动跟踪 → [检查点5] → 完成

2.3 核心改动

文件: mises-behavior-engine/src/core/checkpoint.py（新建）

"""
12-Factor Agent Factor 6: Launch/Pause/Resume
五步行为分析检查点管理
"""
from typing import Optional
from datetime import datetime, timedelta

from src.schemas.models import SessionState
from src.schemas.decisions import DecisionRecord


class CheckpointManager:
    """会话检查点管理器"""

    def __init__(self, db_pool):
        self.db = db_pool

    async def save_checkpoint(
        self,
        session_id: str,
        stage: str,
        stage_data: dict,
        context_window: list,
        ttl_days: int = 30
    ) -> str:
        """保存检查点，返回 checkpoint_id"""
        pass

    async def load_checkpoint(
        self,
        session_id: str,
        stage: Optional[str] = None
    ) -> Optional[dict]:
        """
        加载最新检查点
        - stage=None: 加载最新活跃检查点
        - stage="desires": 加载指定阶段检查点
        """
        pass

    async def resume_session(
        self,
        session_id: str,
        new_input: Optional[str] = None
    ) -> SessionState:
        """
        从检查点恢复会话
        1. 加载检查点
        2. 重建 SessionState
        3. 压缩历史上下文（避免 token 超限）
        4. 注入新输入（如有）
        """
        pass

    async def pause_session(
        self,
        session_id: str,
        reason: str = "user_requested"
    ) -> dict:
        """
        暂停会话
        1. 保存当前阶段检查点
        2. 标记会话为 paused
        3. 返回恢复信息（session_id, stage, resume_hint）
        """
        pass

文件: mises-behavior-engine/src/core/engine.py

# MisesEngine.process() 每完成一个阶段，自动保存检查点
async def _run_stage(self, stage: str, session_state: SessionState, ...):
    result = await self._stage_handlers[stage](...)

    # 自动保存检查点
    await self.checkpoint_mgr.save_checkpoint(
        session_id=session_state.session_id,
        stage=stage,
        stage_data=result,
        context_window=self._compress_context(session_state)
    )

    return result

2.4 API 端点

文件: mises-behavior-engine/src/mcp/server.py

# 暂停会话
@app.post("/mcp/session/{session_id}/pause")
async def pause_session(session_id: str, reason: str = "user_requested"):
    """暂停行为分析会话"""
    result = await checkpoint_mgr.pause_session(session_id, reason)
    return {"status": "paused", "resume_info": result}

# 恢复会话
@app.post("/mcp/session/{session_id}/resume")
async def resume_session(session_id: str, new_input: Optional[str] = None):
    """从检查点恢复会话"""
    session_state = await checkpoint_mgr.resume_session(session_id, new_input)
    result = await engine.process(question=new_input, session_state=session_state)
    return result

2.5 具体改动清单

2.6 验收标准

• 用户在路径生成阶段暂停，3 天后恢复，上下文完整
• 恢复时上下文经过压缩，token 不超限
• 检查点数据存储在 PostgreSQL，不依赖 Redis TTL
• API /mcp/session/{id}/pause 和 /resume 正常工作

Phase 3：Self-Critique 自愈回路（第 5–6 周）

3.1 目标

• Self-Critique 检测到问题后，将具体错误信息反馈给 LLM 重新生成
• 从"门卫模式"（静默放行/拒绝）升级为"教练模式"（指出问题 + 引导修正）
• 最多重试 2 次，超过则降级

3.2 当前问题

# 当前 self_critique.py 第 283-292 行：fail-open
# 检测失败 → 直接返回 is_valid=True → 问题被静默放行

3.3 设计

LLM 生成初始回答
       ↓
Self-Critique 评估 ──→ 通过 → 返回用户
       ↓ (不通过)
  构建修正上下文：
  - 原始回答
  - 具体问题列表
  - 修正指导（retry_guidance）
       ↓
LLM 重新生成（带修正上下文）
       ↓
Self-Critique 再评估 ──→ 通过 → 返回用户
       ↓ (仍不通过)
  降级策略：
  - 标记为 low_confidence
  - 附加免责声明
  - 记录到 decision_audit (status=retry_exhausted)

3.4 核心改动

文件: mises-behavior-engine/src/core/self_critique.py

class SelfCritiqueEngine:

    async def critique_with_retry(
        self,
        response: str,
        context: dict,
        max_retries: int = 2
    ) -> CritiqueWithRetryResult:
        """
        12-Factor Agent Factor 9: Compact Errors into Context Window
        检测到问题后，将错误压缩进上下文，引导 LLM 重试
        """
        for attempt in range(max_retries + 1):
            critique = await self.evaluate(response, context)

            if critique.is_valid and critique.overall_score >= 0.7:
                return CritiqueWithRetryResult(
                    final_response=response,
                    attempts=attempt + 1,
                    final_critique=critique
                )

            if attempt < max_retries:
                # 构建修正上下文
                retry_context = self._build_retry_context(
                    original_response=response,
                    critique=critique,
                    attempt=attempt + 1
                )
                # LLM 带修正上下文重新生成
                response = await self.llm.regenerate(retry_context)
            else:
                # 重试耗尽：降级
                return CritiqueWithRetryResult(
                    final_response=self._add_disclaimer(response, critique),
                    attempts=attempt + 1,
                    final_critique=critique,
                    degraded=True
                )

    def _build_retry_context(
        self, original_response: str, critique, attempt: int
    ) -> str:
        """
        将 Self-Critique 结果压缩为修正指导
        关键：具体、可操作、不超过 500 tokens
        """
        issues = "\n".join(f"- {issue}" for issue in critique.issues_found)
        return f"""你之前的回答存在以下问题（第 {attempt} 次修正）：

{issues}

修正要求：
{critique.retry_guidance}

请重新生成回答，确保解决以上所有问题。保持回答的专业性和完整性。"""

3.5 各模块的修正指导模板

3.6 具体改动清单

3.7 验收标准

• 幻觉检测触发时，LLM 自动重新生成无幻觉的回答
• 重试次数 ≤2，不影响响应延迟超过 50%
• 重试耗尽时附加免责声明，不静默放行
• 所有重试过程记录到 decision_audit (status=retry)

3.8 ✅ 实施记录（v1.2, 2026-02-11）

实际架构决策：创建独立的 CritiqueRetryEngine（src/core/critique_retry.py）， 而非修改现有 self_critique.py。原因：

1. 现有 self_critique.py 专注路径级批评，改动风险高
2. critique_integration.py 是后台 fire-and-forget，不适合改为同步阻塞
3. 独立模块可以安全地包装两者，不侵入现有架构

已完成文件清单：

核心流程：

engine._process_stage() 生成 result.speech_output
    → CritiqueRetryEngine.critique_and_retry()
        → 并行执行: 安全检查 + 一致性检查 + 内容质量启发式
        → 通过 → 直接返回
        → 未通过 → _build_retry_guidance() 构建修正上下文
                 → _regenerate_response() LLM 重新生成
                 → 再检查（最多 2 次重试）
        → 重试耗尽 → 降级 + 免责声明
        → 安全阻断 → 安全回退响应
    → 记录到 decision_audit (step_name=self_critique_retry)
    → 失败降级为旧的 schedule_background_critique

降级安全网：

• 自愈引擎失败 → fallback 到旧的后台批评（fire-and-forget）
• 自愈引擎未启用 → 保留旧逻辑
• 通过 MBE_SELF_HEALING_ENABLED=false 环境变量可随时关闭

Phase 4：人工审核标准化（第 7–8 周）

4.1 目标

• 将"请求人工审核"建模为标准 tool call
• Self-Critique 高风险内容自动触发人工审核流程
• 审核完成后自动恢复处理流程（利用 Phase 2 的暂停/恢复机制）

4.2 设计

LLM 生成回答
    ↓
Self-Critique 评估
    ↓ (高风险: score < 0.4 或安全模块触发)
  输出 tool call: request_human_review
    ↓
CheckpointManager 暂停会话
    ↓
通知审核员（Celery task → webhook/邮件/Admin UI）
    ↓
审核员在 Admin UI 审核
    ↓ (approve / reject / modify)
Webhook 回调 → 恢复会话
    ↓
返回审核后的回答给用户

4.3 核心改动

文件: mises-behavior-engine/src/core/human_review.py（新建）

"""
12-Factor Agent Factor 7: Contact humans with tool calls
人工审核作为标准工具调用
"""

class HumanReviewTool:
    """
    统一的人工审核工具
    对引擎而言，这就是一个普通的 tool call
    """

    async def request_review(
        self,
        session_id: str,
        content: str,
        reason: str,
        urgency: str = "normal",        # normal / high / critical
        critique_result: dict = None,
        metadata: dict = None
    ) -> HumanReviewRequest:
        """
        1. 创建审核请求记录 (PostgreSQL)
        2. 暂停会话 (CheckpointManager)
        3. 分发通知 (Celery task)
        4. 返回审核请求 ID
        """
        pass

    async def on_review_complete(
        self,
        review_id: str,
        action: str,                    # "approve" / "reject" / "modify"
        reviewer_id: str,
        modified_content: Optional[str] = None,
        notes: Optional[str] = None
    ) -> HumanReviewResult:
        """
        审核完成回调
        1. 更新审核记录
        2. 恢复会话 (CheckpointManager.resume_session)
        3. 继续处理流程
        """
        pass

文件: mises-behavior-engine/src/core/self_critique.py

# 在 critique_with_retry() 中增加人工审核触发
if critique.overall_score < 0.4 or critique.has_safety_issue:
    # 触发人工审核 tool call
    review_request = await self.human_review.request_review(
        session_id=session_id,
        content=response,
        reason=f"Self-Critique score {critique.overall_score:.2f}",
        urgency="high" if critique.has_safety_issue else "normal",
        critique_result=critique.dict()
    )
    return CritiqueWithRetryResult(
        final_response=None,
        status="pending_human_review",
        review_id=review_request.id
    )

4.4 API 端点

# 审核完成回调
@app.post("/admin/review/{review_id}/complete")
async def complete_review(review_id: str, body: ReviewCompleteRequest):
    result = await human_review.on_review_complete(
        review_id=review_id,
        action=body.action,
        reviewer_id=body.reviewer_id,
        modified_content=body.modified_content
    )
    return result

# 获取待审核列表
@app.get("/admin/review/pending")
async def list_pending_reviews(urgency: Optional[str] = None):
    return await human_review.list_pending(urgency=urgency)

4.5 具体改动清单

4.6 验收标准

• Self-Critique score < 0.4 自动触发人工审核
• 审核员在 Admin UI 可看到待审核内容及 Self-Critique 报告
• 审核完成后会话自动恢复，用户收到审核后回答
• 审核全过程记录到 decision_audit (status=human_review)

4.7 ✅ 实施记录（v2.0, 2026-02-11）

架构决策：HumanReviewTool 作为独立模块（src/core/human_review.py）， 通过 CritiqueRetryEngine._trigger_human_review() 自动触发，复用 Phase 2 的 CheckpointManager 实现暂停/恢复。

已完成文件清单：

核心流程：

Self-Critique 自愈回路
    ├─ 安全阻断 → _trigger_human_review(urgency=critical)
    └─ 降级 + 分数 < 0.35 → _trigger_human_review(urgency=auto)
                ↓
    HumanReviewTool.request_review()
        → 写入 human_review_requests 表
        → CheckpointManager.pause_session() 暂停会话
        → Celery: send_content_review_notification 通知审核员
                ↓
    审核员操作 (Admin API)
        → POST /api/human-review/{id}/complete
        → HumanReviewTool.on_review_complete()
            → 更新审核记录
            → CheckpointManager.resume_session() 恢复会话
            → Redis 会话状态更新
                ↓
    用户收到审核后的回答

SLA 策略：

API 端点：

四、非功能性要求

4.1 向后兼容

• 所有改动不破坏现有 API 契约
• 新增字段使用 Optional 默认值
• 数据库迁移为 ADD COLUMN ... DEFAULT，不锁表
• Phase 1 的审计记录为旁路写入，不影响主流程延迟

4.2 性能影响预算

4.3 监控指标

# Prometheus 新增指标
mbe_decision_audit_total{step_name, status}       # 决策审计计数
mbe_checkpoint_save_total{stage}                   # 检查点保存计数
mbe_checkpoint_resume_total{stage}                 # 检查点恢复计数
mbe_self_critique_retry_total{module, attempt}     # Self-Critique 重试计数
mbe_human_review_total{urgency, action}            # 人工审核计数
mbe_human_review_latency_seconds{urgency}          # 审核响应时间

五、风险与缓解

六、里程碑与交付

七、与现有计划的关系

八、W9 收尾：集成测试 + 架构文档 + 监控指标

8.1 ✅ 端到端集成测试

文件: tests/integration/test_12factor_e2e.py

总计: 53 项测试（16 Phase3 + 23 Phase4 + 14 E2E）全部通过。

8.2 ✅ 架构文档更新

• docs/product/MBE_SYSTEM_CARD.md 新增 §2.1.1 "12-Factor Agents 生产化架构"
• 含四层架构图和核心设计原则

8.3 ✅ Prometheus 监控指标

文件: src/monitoring/prometheus_metrics.py + src/core/critique_retry.py

8.4 Bug 修复

E2E 测试发现并修复 1 个真实 bug：

• critique_retry.py: _run_critique_checks 抛异常时未被捕获，导致整个自愈引擎崩溃
• 修复: 在循环内加 try/except，异常时默认通过（不阻塞主流程）

九、Factor 4 缺口补全：结构化输出全覆盖

日期: 2026-02-11 | 版本: v2.1

9.1 缺口盘点

审计发现 Factor 4 "Tools are just structured outputs" 存在三处缺口：

9.2 实施方案

缺口 1: ContextVar 跨模块审计传播

问题: KnowledgeRetrieveDecision 和 ResponseGenerateDecision 已在 schemas/decisions.py 中定义，但知识检索和 LLM 调用发生在 dynamic_expert.py 深层，trail 对象无法传递。

方案: 使用 ContextVar 实现零侵入式审计链路传播。

decisions.py:  _current_trail = ContextVar(...)  ← 新增
                set_current_trail() / get_current_trail()
mcp/server.py: rest_analyze → set_current_trail(trail)
engine.py:     process()    → set_current_trail(trail)
dynamic_expert.py: general_qa → get_current_trail() → trail.add(...)

变更文件:

• src/schemas/decisions.py — 添加 ContextVar、set_current_trail、get_current_trail
• src/mcp/server.py — 入口设置 trail；知识库工具 + 专家成功路径记录审计
• src/core/engine.py — process() 设置 trail；事实性问题用 ResponseGenerateDecision
• src/knowledge/dynamic_expert.py — general_qa 中埋入 KnowledgeRetrieveDecision + ResponseGenerateDecision

覆盖情况: 6/6 Decision 模型已全部在审计链路中实际使用。

缺口 2: module_scores 填充

问题: SelfCritiqueDecision.module_scores 字典始终为空，丢失各模块（safety/consistency/content_quality）的独立分数。

方案: 从 CritiqueCheckResult.module_results 提取 score 字段汇总到 module_scores。

变更: src/core/engine.py 中记录 SelfCritiqueDecision 时增加提取逻辑：

for mod_name, mod_data in check_result.module_results.items():
    if isinstance(mod_data, dict) and "score" in mod_data:
        module_scores[mod_name] = mod_data["score"]

缺口 3: structured_llm_call 结构化 LLM 调用

问题: LLM 输出为自由文本，代码侧用正则/字符串后处理提取信息，缺乏 schema 约束。

方案: 创建 src/llm/structured.py 工具模块，封装：

1. 将 Pydantic model 自动转为 JSON schema 提示注入 system_prompt
2. 使用 response_format="json" 调用 LLM
3. 解析 JSON → Pydantic 校验 → 返回结构化实例（失败返回 None）

Pilot: expert_router._rewrite_query 已迁移为结构化调用，返回 QueryRewriteResult(rewritten_query, matched_domain, reasoning)，并保留纯文本 fallback。

后续迁移路线（按优先级）:

9.3 测试

tests/integration/test_12factor_e2e.py::TestFactor4GapCompletion — 11 个测试项：

• ContextVar 跨模块传递验证
• KnowledgeRetrieveDecision / ResponseGenerateDecision 审计写入
• 全部 6 种 Decision 模型可序列化
• module_scores 从 CritiqueCheckResult 提取
• structured_llm_call JSON 解析（正常/markdown包裹/混合文本/非法JSON）
• schema hint 生成
• mock LLM 端到端调用验证

全部 25 个 E2E 测试通过，无回归。

十、Line 13 补全：Self-Critique 15 模块统一评分 schema

日期: 2026-02-11 | 版本: v2.2

10.1 问题

原始分析 Line 13 要求："Self-Critique 的 15 个模块统一为结构化的评分 schema，而不是各自独立的处理逻辑"。

审计发现 SelfCritiqueDecision.module_scores 仅有 3 个模块填充（safety 聚合、consistency、content_quality），覆盖率 20%。

10.2 实施方案

a) SC-12~15 安全子维度拆分

改造: _check_safety 从调用 critique_safety（只返回 bool）改为直接调用 SafetyCritiqueEngine.full_check，获取 result.scores 中的 4 个子分数。

_check_safety()
  ↓
  SafetyCritiqueEngine.full_check()
    → SC-12 harmful_content: 0.0~1.0
    → SC-13 privacy:         0.0~1.0
    → SC-14 bias:            0.0~1.0
    → SC-15 emotional_safety: 0.0~1.0
  ↓
  缓存到 self._last_safety_sub_scores
  ↓
  _run_critique_checks 汇总时展开为独立 module_results

b) SC-06 幻觉检测

新增: _check_hallucination() 启发式方法（毫秒级，不调 LLM），检测：

• 编造的精确统计数据（研究表明 73.6%...）
• 虚构的法规/论文引用（根据《XX法》第X条、哈佛研究发现）
• 编造的地址/联系方式
• 可疑品牌推荐

c) 不适合接入 _run_critique_checks 的模块

10.3 变更文件

10.4 统一评分矩阵（最终状态）

覆盖率: 8/15 模块已纳入统一 module_scores（53%）。其余 7 个模块有独立实现路径或待未来扩展。

10.5 测试

TestLine13UnifiedCritiqueSchema — 8 个测试项：

• SC-12~15 子维度拆分验证
• SC-06 幻觉检测（精确统计/虚构法规/正常文本）
• module_results 包含全部核心维度
• module_scores 8 维度提取
• 安全违规时子维度分数反映
• 端到端 CritiqueRetryEngine → SelfCritiqueDecision 映射

全部 49 个测试通过（33 E2E + 16 单元测试），无回归。

十一、Factor 5 缺口补全：HOPE 学习信号审计

11.1 问题

Factor 5 要求 "执行状态与业务状态统一"。审计发现：decision_audit 表覆盖了 ExpertRoute / KnowledgeRetrieve / ResponseGenerate / SelfCritique / MisesStage / HumanReview 六类决策，但 HOPE 在线学习（process_interaction 返回的 learning_signal / surprise / preference_updated）仅输出到 logger.debug，未进入审计表。

11.2 方案

11.3 HopeLearningDecision 字段

class HopeLearningDecision(BaseModel):
    user_id: str           # 用户标识
    learning_signal: float # HOPE 学习信号
    surprise: float        # 惊奇度
    preference_updated: bool # 是否触发偏好更新
    context_source: str    # 调用来源: engine / dynamic_expert / mcp
    extra: Dict[str, Any]  # 附加上下文（如 sources_count）

11.4 审计覆盖总结

实施后 decision_audit 表覆盖 全部 7 种 决策类型：

11.5 测试

TestFactor5HopeLearningAudit — 7 个测试项：

• HopeLearningDecision 序列化/反序列化
• HopeLearningDecision 默认值
• DecisionType 枚举包含 HOPE_LEARNING
• trail.add 记录 HOPE_LEARNING（engine 上下文）
• trail.add 记录 HOPE_LEARNING（dynamic_expert 上下文）
• DecisionType 总计 7 种验证
• HOPE_LEARNING 与其他决策类型混合 trail 记录

全部 56 个测试通过（40 E2E + 16 单元测试），无回归。

十二、Factor 6 缺口补全：检查点完整性 + TITANS 记忆重建

12.1 问题

Factor 6 (Launch/Pause/Resume) 的两个剩余缺口：

1. stage_data 过于简略 — checkpoint 只保存 {"current_stage": "desires"}，丢失了该阶段的 结构化输出（如 uneasiness 分析结果、desires 列表、paths 选项），恢复精度不足。
2. TITANS 短期记忆断档 — 短期 TTL 5 分钟、中期 24 小时，用户暂停数天后 TITANS 对话记忆 全部过期，恢复后 LLM 缺少对话语境。

12.2 缺口 1：stage_data 完整性增强

文件: src/core/engine.py

checkpoint 保存时，stage_data 从仅含 current_stage 扩展为：

_stage_data = {
    "current_stage": session.current_stage,
    "stages_completed": list(session.stages_completed),
    "stage_output": session.context[stage_key],  # 该阶段的完整结构化输出
    "last_user_input": user_input[:500],
    "last_response_preview": result.speech_output[:500],
}

阶段→输出键映射：

12.3 缺口 2：TITANS 记忆重建

文件: src/core/engine.py — 新增 _rebuild_titans_memory_from_checkpoint()

恢复流程扩展：

session.is_paused=True
  → checkpoint_mgr.resume_session()    # 恢复 stage + context
  → _rebuild_titans_memory_from_checkpoint()  # ← 新增
      → 从 checkpoint context 提取关键信息
      → 合成 "恢复摘要" 作为虚拟对话
      → 注入 TITANS encode_interaction()
      → TITANS 短期缓存重新拥有对话语境

提取的信息片段：

• uneasiness.surface + root_cause → "用户的困扰"
• desires.primary_desire → "用户的愿望"
• paths.paths[].name → "可选方案"
• chosen_path → "用户选择"

注入为一条带 context={"resumed": True} 标记的记忆， 让后续 LLM 调用能通过 retrieve_relevant_memories() 访问恢复的语境。

12.4 测试

TestFactor6CheckpointEnhancements — 8 个测试项：

• stage_data 包含各阶段结构化输出（uneasiness / paths / action）
• init 阶段不保存 checkpoint 验证
• TITANS 重建：完整上下文注入验证
• TITANS 重建：空上下文不触发
• TITANS 重建：禁用时不触发
• TITANS 重建：失败时不抛异常（旁路原则）

全部 64 个测试通过（48 E2E + 16 单元测试），无回归。

十三、Factor 7 缺口补全：人工审核 MCP Tool Call

13.1 问题

Factor 7 要求 "将请求人工审核建模为标准 tool call"。已有完整的 HumanReviewTool 内部实现 （暂停/通知/恢复），但未注册到 MCP 工具列表，LLM 无法像调用其他工具一样主动请求人工审核。

注：AIGC Committee (:3002) 为独立项目，将来配合 MBE 生态使用，不在此次集成范围内。

13.2 方案

13.3 Tool Call 格式

LLM 可输出如下 tool call，完全符合 Factor 7 原文建议：

{
  "tool": "request_human_review",
  "params": {
    "content": "建议您每天服用阿司匹林 100mg",
    "reason": "涉及药物剂量建议，需医学专业确认",
    "urgency": "high",
    "session_id": "sess-001",
    "device_id": "dev-001"
  }
}

13.4 完整调用链

LLM 判断需要人工审核
  → 输出 tool call: request_human_review
  → MCPServer._handle_tools_call() 分发
  → _call_request_human_review()
      → HumanReviewTool.request_review()
          → 写入 human_review_requests 表
          → CheckpointManager.pause_session()
          → Celery 通知审核员
      → 写入决策审计 (HumanReviewDecision)
      → 返回审核状态给 LLM

13.5 测试

TestFactor7HumanReviewToolCall — 9 个测试项：

• MISES_TOOLS 包含 request_human_review
• inputSchema 必填字段验证
• urgency/content_type 枚举验证
• description 覆盖关键场景
• 服务未启用时优雅降级
• 参数缺失时返回错误
• 成功请求审核返回审核信息
• 工具总数 = 8 验证

全部 73 个测试通过（57 E2E + 16 单元测试），无回归。

十四、Factor 11 缺口补全：出站 Webhook + 行为提醒 + Celery Beat 部署

14.1 问题

Factor 11 (Trigger from anywhere) 的核心触发源（MCP/REST/前端/设备/WebSocket）已大量实现， 但缺少三项：出站 Webhook 分发、行为跟踪定时提醒、Celery Beat 生产部署。

14.2 缺口 1：出站 Webhook 系统

架构：

LoopEventBus.publish(event)
  → WebhookDispatcher._on_event()      (内存订阅匹配)
  → celery.send_task("dispatch_webhook") (fire-and-forget)
  → dispatch_webhook()
      → HMAC-SHA256 签名 (X-MBE-Signature: sha256=...)
      → httpx.post(url, body, headers)
      → 成功 → 记录 / 失败 → 指数退避重试（最多 5 次）

支持的事件类型（12 种）：eval_triggered/completed/failed、hope_feedback/negative_threshold、 training_suggested、expert_quarantined/recovered、kb_updated、human_review_completed、 session_paused/resumed。

14.3 缺口 2：行为跟踪提醒

新文件: src/tasks/behavior_tasks.py

策略: Celery Beat 每天 09:00 扫描暂停超过 3 天的活跃 checkpoint，分级发送提醒：

提醒含个性化上下文（从 checkpoint 提取困扰/愿望/方案选择），通过 notification_tasks 发送。 去重机制：behavior_reminders_log 表确保每个 session + tier 仅发送一次。

14.4 缺口 3：Celery Beat 生产部署

文件: docker-compose.prod.yml

新增 3 个服务：

14.5 测试

TestFactor11WebhookDispatcher — 8 个测试 + TestFactor11BehaviorReminders — 7 个测试：

• WebhookSubscription 模型 + HMAC 签名计算
• 事件类型白名单 + 无效事件拒绝
• 订阅增删查 + 事件匹配分发
• payload 清理
• 提醒分级配置 + 天数→级别映射
• 上下文提示构建 + 阶段标签覆盖
• Celery Beat schedule 包含提醒任务

全部 88 个测试通过（72 E2E + 16 单元测试），无回归。

Phase 5：运维自动化（第 9-10 周）— 2026-02-12 实现

15.1 目标

基于 12-Factor 原则（特别是 Factor 9 错误压缩进上下文），建立完整的内部运维自动化体系：

15.2 Ops Agent（运维 Agent）

文件: src/deploy/ops_agent.py

核心能力：

• 命令白名单：按风险等级（SAFE/NORMAL/DANGEROUS）分类，BLOCKED 模式（rm -rf、sudo 等）直接拦截
• 审计链路：所有执行的命令写入 decision_audit 表
• 超时控制：asyncio.subprocess 超时保护
• 历史与统计：执行历史、成功率、平均耗时

15.3 CodeRunner（Factor 9 核心实现）

文件: src/deploy/code_runner.py

实现 "code → run → check → fix → retry" 闭环：

1. run_code() — 执行代码片段（Python/JS/TS/Shell）
2. run_test() — 运行 pytest，解析通过/失败/跳过
3. run_with_retry() — Factor 9 重试循环，错误提取 → fix_callback → 重试
4. _extract_error_summary() — 结构化错误摘要（Python traceback / JS Error / Shell 错误）

25 个单元测试全部通过。

15.4 BenchmarkRunner（Benchmark 自动化）

文件: src/benchmark/runner.py

自动化编排层，驱动 MBE-Bench v1.0 的 6 维度评估：

1. health_check() — 依赖检查（Suite/测试数据/报告目录/OpsAgent）
2. run_full() — 完整 pipeline（检查→执行→报告→基线对比→审计）
3. _compare_with_baseline() — 与历史基线对比（improved/regressed/stable）
4. _generate_human_summary() — 人类可读报告（Factor 9 应用）
5. API 端点：run / health / history / trend / baseline

21 个单元测试全部通过。

15.5 知识库维护自动化

文件: src/knowledge/kb_maintenance.py + src/api/admin/graph_api.py + src/tasks/knowledge_maintenance_tasks.py

补齐知识库维护 3 个缺口：

Celery Beat 任务：

15.6 应用市场发布 & 一键部署 & 收益集成

文件: src/api/admin/app_market_api.py + src/api/admin/deploy_api.py + src/api/admin/revenue_market_api.py

补齐开发者全流程 3 个核心缺口：

应用市场 API 端点：

部署目标：Docker 本地 / Cloudflare Pages / Vercel / 本地开发

收益分成比例：开发者 70% / 平台 30%

15.7 Desktop 新功能页面

新增文件：

• mbe-desktop/src/renderer/pages/NewProject/index.tsx
• mbe-desktop/src/renderer/pages/Benchmark/index.tsx
• mbe-desktop/src/renderer/pages/Deploy/index.tsx
• mbe-desktop/src/renderer/pages/Publish/index.tsx
• mbe-desktop/src/renderer/pages/Chat/index.tsx ← 新增
• mbe-desktop/src/renderer/pages/Revenue/index.tsx ← 新增
• mbe-desktop/src/renderer/components/Onboarding.tsx ← 新增
• mbe-desktop/src/renderer/services/appMarketService.ts
• mbe-desktop/src/renderer/services/deployService.ts
• mbe-desktop/src/renderer/services/benchmarkService.ts

15.7a 用户旅程优化（断裂连接修复）

本节补齐了用户全流程中的所有断裂连接，确保三条路径（A 发布专家、B 开发应用、C 终端使用）均形成闭环并可互通。

15.8 VS Code 扩展

目录: mbe-vscode-extension/

MBE 的 VS Code 集成扩展，支持在 IDE 内直接使用 MBE 能力：

配置项：mbe.serverUrl、mbe.apiKey、mbe.defaultExpert、mbe.language

15.9 测试

15.10 管理员全流程修复（Phase 8）

P0 — 安全鉴权

为 6 个无鉴权 API 模块统一补齐 require_admin_api / require_super_admin_api 依赖注入：

新增 common.py 辅助函数：

• require_admin_api — 返回 401 JSON（替代 302 重定向）
• require_super_admin_api — 数据库实时验证 super_admin 角色
• require_admin_or_operator_api — 含 operator 只读场景

P1 — Admin 管理页面

文件：src/api/admin/admin_pages.py（Jinja2 服务端渲染） 导航栏已更新：运营下拉菜单新增 4 个入口。

P2 — Desktop 角色区分

MBE 12-Factor Agents 实施计划 v3.0 | Phase 8 管理员全流程修复（鉴权 + Admin UI + 角色守卫） | 2026-02-12