alert-contract-expiring-hr/scripts/service/llm_decision.py

"""AI 续聘建议统一入口：优先 LLM，本地降级规则版。

宿主环境（jiangchang_skill_core 可用）：
    调用 jiangchang_skill_core.llm_manager.call_llm_json，
    把材料文本摘要 + 业绩数据 + 教师信息组成 prompt。

本地（jiangchang_skill_core 不可用）：
    降级到 task_generator 规则版（基于分数阈值）。
    reason 文案加上「已结合材料包（含 Y 个文件，X 字内容）」前缀。

返回值：
    (suggestion, reason, decision_source)
    decision_source ∈ {"llm", "rules", "rules_no_materials"}
"""

from __future__ import annotations

import sqlite3
from typing import Tuple

from service import task_generator
from util.constants import DEFAULT_MATERIALS, SUGGESTIONS
from util.logging_setup import get_skill_logger

logger = get_skill_logger()

SYSTEM_PROMPT = """你是高校人事处续聘评审助手。请根据教师信息和已提交的续聘材料包内容，判断该教师是否建议续聘。

要求：
1. 严格基于材料文本内容判断；缺少关键材料的请在 reason 里指出
2. suggestion 必须从这三项中选一：
   - "renew"：建议续聘
   - "observe"：常规续聘评审（建议复核）
   - "non_renew"：不建议续聘
3. reason 用 1 句话，不超过 30 字
4. 输出严格 JSON：{"suggestion": "...", "reason": "..."}
"""


def _build_user_prompt(
    teacher_info: dict,
    perf_records: list[sqlite3.Row],
    materials_summary: str,
    materials_files: list[str],
) -> str:
    perf_lines = []
    for p in perf_records:
        perf_lines.append(
            f"- {p['year']} 年度：综合 {p['performance_score']:.0f} "
            f"（教学 {p['teaching_score']:.0f} / 科研 {p['research_score']:.0f}）"
            f"，师德 {p['moral_teaching_score']:.0f}"
        )
    perf_block = "\n".join(perf_lines) if perf_lines else "（无业绩数据）"

    checklist_bullets = "\n".join(f"- {m}" for m in materials_files) if materials_files else "（无材料文件）"

    return f"""## 教师信息
- 姓名：{teacher_info.get('teacher_name', '?')}
- 单位：{teacher_info.get('unit_name', '?')}
- 岗位：{teacher_info.get('position', '?')}
- 合同起止：{teacher_info.get('contract_start', '?')} ~ {teacher_info.get('contract_end', '?')}
- 距到期：{teacher_info.get('days_to_expire', '?')} 天
- 预警等级：{teacher_info.get('alert_level', '?')}

## 近 3 年业绩
{perf_block}

## 已提交的续聘材料包（已 OCR/解析）

### 文件清单
{checklist_bullets}

### 材料文本摘要
{materials_summary if materials_summary else '（材料文本尚未解析，将基于规则判断）'}

请基于以上真实材料内容给出续聘建议。"""


def _call_llm(
    teacher_info: dict,
    perf_records: list[sqlite3.Row],
    materials_summary: str,
    materials_files: list[str],
) -> dict | None:
    """调宿主 jiangchang_skill_core.llm_manager.call_llm_json。
    返回 dict（含 suggestion / reason）或 None（不可用 / 失败）。
    """
    try:
        from jiangchang_skill_core.llm_manager import call_llm_json  # type: ignore
    except ImportError:
        return None

    user = _build_user_prompt(teacher_info, perf_records, materials_summary, materials_files)
    try:
        result = call_llm_json(system=SYSTEM_PROMPT, user=user)
    except Exception as e:
        logger.warning("llm_decision: LLM 调用失败降级: %s", e)
        return None

    if not isinstance(result, dict):
        return None
    sug = result.get("suggestion")
    reason = result.get("reason", "")
    if sug not in SUGGESTIONS:
        logger.warning("llm_decision: LLM 返回非法 suggestion=%r", sug)
        return None
    return {"suggestion": sug, "reason": reason}


def decide_suggestion(
    perf_records: list[sqlite3.Row],
    materials_summary: str,
    materials_files: list[str],
    teacher_info: dict,
) -> Tuple[str, str, str]:
    """返回 (suggestion, reason, decision_source)。

    decision_source:
        "llm" - 来自宿主 LLM
        "rules" - 规则版（已结合材料摘要）
        "rules_no_materials" - 规则版（无材料数据）
    """
    files = materials_files or list(DEFAULT_MATERIALS)

    # 1) 优先 LLM
    llm = _call_llm(teacher_info, perf_records, materials_summary, files)
    if llm is not None:
        logger.info(
            "llm_decision: LLM 决策 %s (%s) materials=%d 字",
            llm["suggestion"], llm["reason"], len(materials_summary),
        )
        return (llm["suggestion"], llm["reason"], "llm")

    # 2) 降级规则版
    sug, base_reason = task_generator.suggest(perf_records)

    if materials_summary:
        if files:
            prefix = (
                f"已结合材料包（{len(files)} 份文件，"
                f"共 {len(materials_summary)} 字内容）综合判断："
            )
        else:
            prefix = f"已结合材料包（{len(materials_summary)} 字）综合判断："
        source = "rules"
    elif files:
        prefix = f"已结合材料包（{len(files)} 份文件，待解析）综合判断："
        source = "rules"
    else:
        prefix = "（无材料数据，基于业绩规则判断）"
        source = "rules_no_materials"

    return (sug, prefix + base_reason, source)