Create-knowledge-find-decode/decode_content/gates.py

"""Creation gates used before and during deep decode."""
from __future__ import annotations

import json
from typing import Any, Callable

from core.llm import chat_json as default_chat_json
from core.prompts import load_prompt
from core.text_limits import DIRECTIVE_CONTEXT_MAX_CHARS, clip_text
from decode_content.models import GateResult


GATE_ADMIT = load_prompt("gate_admit")
GATE_REFUTE = load_prompt("gate_refute")
GATE_TIEBREAK = load_prompt("gate_tiebreak")
GATE_HOW_ADMIT = load_prompt("gate_how_admit")
GATE_HOW_REFUTE = load_prompt("gate_how_refute")
GATE_HOW_TIEBREAK = load_prompt("gate_how_tiebreak")
GATE_WHY_REFUTE = load_prompt("gate_why_refute")

ChatJsonFn = Callable[..., dict[str, Any]]


def vote_bool(
    system: str,
    user: str,
    key: str,
    *,
    on_fail: bool,
    chat_json_fn: ChatJsonFn = default_chat_json,
    timeout: int = 90,
) -> bool:
    try:
        return bool(chat_json_fn(system, user, timeout=timeout).get(key))
    except Exception:
        return on_fail


def creation_gate(read_text: str, *, chat_json_fn: ChatJsonFn = default_chat_json) -> GateResult:
    """Decide whether the read content is reusable creation knowledge."""

    v_admit = vote_bool(GATE_ADMIT, read_text, "in_scope", on_fail=True, chat_json_fn=chat_json_fn)
    v_refute = not vote_bool(GATE_REFUTE, read_text, "out_of_scope", on_fail=False, chat_json_fn=chat_json_fn)
    if v_admit == v_refute:
        return GateResult(
            passed=v_admit,
            reason=f"admit={v_admit}/refute={v_refute} 一致",
            details={"admit": v_admit, "refute": v_refute},
        )
    v_tie = vote_bool(GATE_TIEBREAK, read_text, "in_scope", on_fail=False, chat_json_fn=chat_json_fn)
    return GateResult(
        passed=v_tie,
        reason=f"admit={v_admit}/refute={v_refute} 分歧→裁决={v_tie}",
        details={"admit": v_admit, "refute": v_refute, "tiebreak": v_tie},
    )


def chain_signals(steps: list[dict[str, Any]]) -> list[str]:
    sigs: list[str] = []
    outs = [(s.get("output") or "") for s in steps]
    independent_count = 0
    for idx, step in enumerate(steps):
        if idx == 0:
            continue
        step_input = step.get("input") or ""
        if not (("←" in step_input) or any(o and o[:4] in step_input for o in outs[:idx])):
            independent_count += 1
    if independent_count >= 2:
        sigs.append(f"{independent_count} 个后步的 input 未指向前步产出（各自起头）")
    for i in range(len(outs)):
        for j in range(i + 1, len(outs)):
            left, right = outs[i], outs[j]
            if left and right and (left in right or right in left):
                sigs.append(f"步骤{i + 1}与{j + 1}产出近义（{left} / {right}）")
                break
    return sigs


def how_gate(knowledge: dict[str, Any], *, chat_json_fn: ChatJsonFn = default_chat_json) -> GateResult:
    payload = json.dumps(
        {
            "purpose": knowledge.get("purpose"),
            "steps": [
                {
                    "input": step.get("input"),
                    "方法": clip_text(step.get("directive") or "", DIRECTIVE_CONTEXT_MAX_CHARS),
                    "产出": step.get("output"),
                }
                for step in knowledge.get("steps", [])
            ],
            "代码信号": chain_signals(knowledge.get("steps", [])),
        },
        ensure_ascii=False,
    )
    v_admit = vote_bool(GATE_HOW_ADMIT, payload, "is_real_how", on_fail=True, chat_json_fn=chat_json_fn)
    v_refute = not vote_bool(GATE_HOW_REFUTE, payload, "is_fake", on_fail=False, chat_json_fn=chat_json_fn)
    if v_admit == v_refute:
        return GateResult(
            passed=v_admit,
            reason=f"admit={v_admit}/refute={v_refute} 一致",
            details={"admit": v_admit, "refute": v_refute},
        )
    v_tie = vote_bool(GATE_HOW_TIEBREAK, payload, "is_real_how", on_fail=False, chat_json_fn=chat_json_fn)
    return GateResult(
        passed=v_tie,
        reason=f"admit={v_admit}/refute={v_refute} 分歧→裁决={v_tie}",
        details={"admit": v_admit, "refute": v_refute, "tiebreak": v_tie},
    )


def why_refute_gate(knowledge: dict[str, Any], *, chat_json_fn: ChatJsonFn = default_chat_json) -> GateResult:
    payload = json.dumps({"阐述": knowledge.get("阐述")}, ensure_ascii=False)
    try:
        result = chat_json_fn(GATE_WHY_REFUTE, payload, timeout=90)
    except Exception:
        return GateResult(passed=True, reason="API错误→保留", details={"not_why": False})
    not_why = bool(result.get("not_why"))
    verdict = str(result.get("verdict") or "")
    reason = str(result.get("reason") or "")
    return GateResult(
        passed=not not_why,
        reason=reason,
        details={"not_why": not_why, "verdict": verdict},
    )