ZNJJ-api-server/src/agent/collect.py

"""采集节点 LLM 接入（作业 4）。

职责边界：
- LLM 一次调用同时产出 ``<state>XXXX</state>正文``——
  不得为复判状态再调用第二次（计划作业 4-3）；
- 代码校验前缀合法性、阶段枚举与白名单守卫，
  非法提议不写状态（受控修复重试：前缀/非法提议最多各修复一次）；
- 字段候选经 fields.validate_candidate 校验后进入
  accident_fields / confirm_status / turn_patch（三态语义）；
- 澄清/无回复计数是代码侧簿记，只作为 LLM 判断的输入，
  转不转人工由模型按提示词决定（代码不代替模型决定）。

失败语义（计划作业 4-10）：前缀缺失/非法且修复失败时，抛出
``CollectReplyError``——不提交业务状态、不泄漏原始模型内容；
调用方（API 层）将其转换为可识别错误响应。
"""

from collections.abc import Callable
import json
import re

from .fields import (
    FieldCandidate,
    FieldStatus,
    validate_candidate,
)
from .llm import LLMService
from .prompts import COLLECTION, render_prompt
from .state import AgentState
from .transitions import COLLECT_STAGE, is_legal_transition

STATE_PREFIX_RE = re.compile(r"^\s*<state>\s*(\d{4})\s*</state>")
_STATE_TAG_RE = re.compile(r"</?state>")

# 采集阶段澄清阈值（来源：prompts/collection.md）
COLLECT_FAILURE_LIMIT = 5

# 受控修复指令（计划作业 4-10）。追加在系统提示词之后，
# 明确要求输出格式，供模型自我修正。
_REPAIR_SUFFIX = (
    "\n\n【输出修复】你上一次的输出不合规：必须且只能以"
    " <state>四位状态码</state> 开头（如 <state>1002</state>），"
    "正文部分不得再出现 <state> 或 </state> 标签。"
    "请重新输出一条完全合规的回复。"
)


class CollectReplyError(Exception):
    """采集回复不可用：前缀缺失/非法且修复失败。

    不携带原始模型内容，避免向上层泄漏。
    """


def parse_state_reply(text: str) -> tuple[str, str]:
    """解析 ``<state>XXXX</state>`` 前缀，返回 (stage_code, 正文)。

    严格模式：无有效前缀、或正文再出现 state 标签（泄漏），
    都视为不合规并抛 ValueError，交由 run_collect 决定是否修复。
    """
    match = STATE_PREFIX_RE.match(text)
    if not match:
        raise ValueError("回复缺少有效的 <state> 前缀")
    body = text[match.end():].strip()
    if _STATE_TAG_RE.search(body):
        raise ValueError("正文中再次出现 state 标签")
    return match.group(1), body


def _try_parse(text: str) -> tuple[str, str] | None:
    """宽松入口：不合规返回 None 而不是抛错（供修复流程判断）。"""
    try:
        return parse_state_reply(text)
    except ValueError:
        return None


async def validated_reply(
    llm: LLMService,
    system_prompt: str,
    user_message: str,
    current: str,
    *,
    validator: Callable[[str, str], bool] = is_legal_transition,
    history: list[dict[str, str]] | None = None,
) -> tuple[str, str]:
    """模型同时产出下一阶段和正文；格式与转换共用一次修复预算。

    只校验模型提议，不按输入长度或关键字替模型决定。非法输出不写回
    图状态；修复仍失败则抛出不含模型原文的错误。
    """
    for attempt in range(2):
        prompt = system_prompt
        if attempt:
            prompt += (
                _REPAIR_SUFFIX
                + f"\n当前阶段为 {current}。请重新检查该阶段允许的转换"
                "和当前 clientMode，不得跳步或切换到另一模式的出口。"
            )
        raw = await llm.complete(prompt, user_message, history=history)
        parsed = _try_parse(raw)
        if parsed is not None and validator(current, parsed[0]):
            return raw, parsed[0]
    raise CollectReplyError("模型回复格式或阶段转换不合规（一次修复后仍失败）")


# 工作流 chatNode 的 history 配置为 50；保留独立消息及其角色。
_HISTORY_MESSAGES = 50


def conversation_history(state: AgentState) -> list[dict[str, str]]:
    """历史不拼入 user 消息，不剥除历史 assistant 的状态前缀。"""
    return [dict(message) for message in (state.get("dialogue") or [])[-_HISTORY_MESSAGES:]]


def context_prompt(state: AgentState) -> str:
    """运行时上下文随 system 提示提供；业务判断仍由模型负责。"""
    return "\n\n【当前会话上下文】\n" + json.dumps({
        "stage_code": state.get("stage_code") or COLLECT_STAGE,
        "client_mode": state.get("client_mode") or "direct",
        "person_index": state.get("person_index") or 1,
        "accident_fields": state.get("accident_fields") or {},
        "person_fields": state.get("person_fields") or {},
        "no_reply_count_before_this_turn": state.get("no_reply_count") or 0,
    }, ensure_ascii=False)


def _build_user_message(state: AgentState) -> str:
    """对齐工作流 Text Editor#2：只传用户本轮原始文本。"""
    return state.get("input_text", "")



def _extract_candidates(state: AgentState, body: str) -> list[FieldCandidate]:
    """从模型正文中提取字段候选（计划作业 4-4）。

    参考版返回空列表：字段提取依赖模型的结构化输出，留给真实
    LLM 接入后实现（同一回复内 JSON 解析优先；确需独立提取调用时，
    记录原因/调用次数/延迟）。validate_candidate 的字段级校验
    已由单元测试覆盖，此处先接通数据流。
    """
    return []


def _apply_candidates(
    state: AgentState, candidates: list[FieldCandidate]
) -> tuple[dict, dict, dict]:
    """把校验通过的候选写入正式字段。

    返回 (accident_fields, confirm_status, turn_patch)。
    只有 CONFIRMED 的候选才写入正式字段并进入本轮差量；
    CANDIDATE/UNKNOWN 只更新 confirm_status，供后续核实节点消费
    （计划作业 4-5 的三态语义）。
    """
    accident_fields = dict(state.get("accident_fields") or {})
    confirm_status = dict(state.get("confirm_status") or {})
    patch: dict = {}
    for cand in candidates:
        validated = validate_candidate(cand)
        confirm_status[validated.key] = validated.status
        if validated.status is FieldStatus.CONFIRMED:
            accident_fields[validated.key] = validated.value
            patch[validated.key] = validated.value
    return accident_fields, confirm_status, patch


def _bookkeeping(state: AgentState) -> tuple[int, int]:
    """代码侧上下文簿记：无回复计数 + 澄清计数。

    返回 (no_reply_count, clarify_count)。注意这只是给 LLM 的
    判断输入，是否转人工由模型按提示词决定。

    澄清失败次数由模型结合语义与历史判断；这里透传已有值，不把
    每次普通输入都当作一次失败。
    """
    event = state.get("client_event") or ""
    no_reply_count = state.get("no_reply_count") or 0
    if event == "no_reply":
        no_reply_count += 1
    elif (state.get("input_text") or "").strip():
        no_reply_count = 0

    return no_reply_count, state.get("clarify_count") or 0


async def run_collect(state: AgentState, llm: LLMService) -> dict:
    """采集节点核心：一轮对话的 LLM 接入（参考实现）。

    流程：一次模型调用 → 前缀校验（不合规修复一次）→
    白名单守卫（非法提议修复一次）→ 字段候选落库 → 返回状态增量。
    任何一步修复后仍不合规都抛 CollectReplyError：不提交业务状态、
    错误信息不携带原始模型内容。
    """
    system_prompt = render_prompt(COLLECTION, state) + context_prompt(state)
    user_message = _build_user_message(state)
    current = state.get("stage_code") or COLLECT_STAGE
    raw, stage = await validated_reply(
        llm, system_prompt, user_message, current, history=conversation_history(state)
    )


    # 4) 字段候选提取与三态落库（提取本身见 _extract_candidates）
    accident_fields, confirm_status, patch = _apply_candidates(
        state, _extract_candidates(state, raw)
    )
    no_reply_count, clarify_count = _bookkeeping(state)

    # 5) 状态增量：response_text 保留完整前缀文本，API 层负责解析；
    #    dialogue 追加本轮消息，供下一轮历史注入（计划作业 4-5）
    dialogue = list(state.get("dialogue") or [])
    dialogue.append({"role": "user", "content": state.get("input_text", "")})
    dialogue.append({"role": "assistant", "content": raw})
    return {
        "turn_count": (state.get("turn_count") or 0) + 1,
        "no_reply_count": no_reply_count,
        "clarify_count": clarify_count,
        "response_text": raw,
        "stage_code": stage,
        "raw_status_code": stage,
        "accident_fields": accident_fields,
        "confirm_status": confirm_status,
        "turn_patch": patch,
        "dialogue": dialogue,
    }