从"硬编答案"到"我不知道"：Agent 决策信心机制

在 LangGraph 状态里增加 confidence + uncertain_reason + human_decision 三个字段。

from typing import TypedDict, Literal, Optional
from langgraph.graph import StateGraph, END

# ===== v3 Loop 状态：加 3 个新字段 =====
class LoopState(TypedDict):
    # ===== v1-v2 原有字段 =====
    run_id: str
    feedback: dict
    candidates: list[dict]
    selected: list[dict]
    owned_tasks: list[dict]
    cut_outputs: list[dict]
    publish_results: list[dict]
    errors: list[dict]

    # ===== v3 新增：决策信心相关 =====
    last_decision: Optional[dict]            # 最近一次决策结果
    confidence: float                        # 0-1，决策信心
    uncertain_reason: Optional[str]          "我不知道"的原因
    decision_history: list[dict]             # 所有决策历史（含信心）
    human_decision: Optional[dict]           # Frank 拍板结果
    human_decision_required: bool            # 是否需要 Frank 拍板
    daily_interrupt_count: int               # 当日已触发拍板次数（上限 10）

关键设计：

• confidence < 0.6 → human_decision_required = True → 触发 interrupt

• human_decision 记录 Frank 拍板内容（"选 A" / "选 B" / "跳过" / "重新选"），供后续 run 借鉴

• daily_interrupt_count 防止 Frank 拍板疲劳（D2 决策点：日上限 10 次）

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.prebuilt import ToolNode
import random

# ===== 决策节点：每次决策都带 confidence =====
async def decision_node(state: LoopState) -> dict:
    """
    调用 Qwen-Max 做选剧/剪辑/发布决策。
    返回结果必须包含 confidence 字段。
    """
    user_input = state.get("last_user_input", "")
    feedback = state."feedback"

    # Qwen-Max 返回 JSON：{action, params, confidence, reason}
    response = await qwen_max.decide(
        prompt=build_decision_prompt(user_input, feedback),
        tools=state."available_tools",  # 来自 M1 工具箱
        schema={
            "action": "string",
            "params": "object",
            "confidence": "number",  # 0-1
            "reason": "string",
        }
    )

    confidence = response["confidence"]
    daily_count = state.get("daily_interrupt_count", 0)

    # ===== 论文 3 核心：信心分层处理 =====
    if confidence >= 0.8:
        # 高信心：直接执行
        return {
            "last_decision": response,
            "confidence": confidence,
            "human_decision_required": False,
            "decision_history": state.get("decision_history", []) + [{
                "decision": response, "layer": "auto_high", "ts": now()
            }],
        }

    elif confidence >= 0.6:
        # 中信心：执行 + 飞书告警
        await feishu.send_alert(
            title="⚠️ AI Loop 中信心决策",
            body=f"决策: {response['action']} 信心: {confidence} 原因: {response['reason']}",
            level="warn",
        )
        return {
            "last_decision": response,
            "confidence": confidence,
            "human_decision_required": False,
            "decision_history": state.get("decision_history", []) + [{
                "decision": response, "layer": "auto_mid_alert", "ts": now()
            }],
        }

    # 低信心：检查日拍板上限
    if daily_count >= 10:
        # 超过上限：降级到中信心执行
        await feishu.send_alert(
            title="🚨 AI Loop 拍板超限，降级自动执行",
            body=f"日拍板已 {daily_count} 次，降级自动。决策: {response['action']}",
            level="critical",
        )
        return {
            "last_decision": response,
            "confidence": confidence,
            "human_decision_required": False,
            "decision_history": state.get("decision_history", []) + [{
                "decision": response, "layer": "auto_degraded", "ts": now()
            }],
        }

    # 正常低信心：触发拍板
    await feishu.send_decision_card(
        title="🤔 AI Loop 需要你拍板",
        decision=response,
        run_id=state["run_id"],
    )
    return {
        "last_decision": response,
        "confidence": confidence,
        "uncertain_reason": response["reason"],
        "human_decision_required": True,
        "daily_interrupt_count": daily_count + 1,
    }


# ===== 路由：检查是否需要等 Frank 拍板 =====
def route_after_decision(state: LoopState) -> Literal["human_review", "execute"]:
    if state.get("human_decision_required", False):
        return "human_review"
    return "execute"


# ===== Human Review 节点：等 Frank 拍板 =====
def human_review_node(state: LoopState) -> dict:
    """
    LangGraph interrupt 模式：暂停，等 Frank 在飞书卡片上点按钮。
    """
    # interrupt() 会暂停整个 graph，等外部输入
    human_input = interrupt({
        "question": state["uncertain_reason"],
        "ai_suggestion": state["last_decision"],
        "options": ["采用 AI 建议", "选 A 方案", "选 B 方案", "跳过", "重新选"],
    })

    # Frank 拍板后继续
    return {
        "human_decision": human_input,
        "human_decision_required": False,
        "decision_history": state.get("decision_history", []) + [{
            "decision": human_input, "layer": "human_review", "ts": now()
        }],
    }

Frank 在飞书群里直接看到卡片，点按钮拍板，Agent 自动继续。

点击"采用 AI 建议"或选 ①/②/③ 触发 LangGraph interrupt 恢复； 点击"跳过"则该 run 放弃，记录到 history。