"""The model's contribution to a mover review: which categories deserve a look, and why. A department move on a senior engineer produces a packet with 238 items across ten categories. Nobody reads 238 rows. Handing that to a new manager as a flat list is the review-storm failure mode — it teaches people to close reviews unread, which is worse than not asking. So the model works at **category** granularity, not per item. Twenty-odd categories is a size a person can hold in their head; 238 rows is not. Per-item calls would also be 238 model calls per move, which is a cost nobody would sign off and a latency nobody would wait for. Three properties make the output safe to show a manager: **It cannot invent a number.** The fact package is built from the packet (counts per category, the before/after org path, the move kind), and every count the model may mention is in it. ``_validate`` rejects any integer in the narrative that is not one of the published facts, the same discipline ``app/intelligence/contracts.py`` enforces — because a recommendation that quietly overstates how much is at stake is a recommendation that gets acted on. **It cannot invent a category.** Recommendations are matched against the categories actually present in the packet; anything else is dropped. **It decides nothing.** The recommendation lands in ``ai_recommendation`` next to the human's ``decision``, never in it. Issue #623's policy is unchanged: a person decides, and nothing is executed from here at all. The packet is fully usable with none of this. When the model is absent or fails, ``ai_failure_code`` records why and the surfaces say so — rather than rendering a blank space that reads like "nothing to worry about". """ from __future__ import annotations import logging import re from collections import Counter from dataclasses import dataclass, field from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.hr.models import ( ITEM_DECISION_KEEP, ITEM_DECISION_REVOKE, MOVE_DEPARTMENT, HrRoleChangeReview, HrRoleChangeReviewItem, ) from app.hr_lifecycle.mover_holdings import REVIEW from app.integrations.models import PlatformSetting logger = logging.getLogger(__name__) # Same connector the rest of the platform's AI reads — one place to # configure a key (see ``app/domains/support/copilot.py``). _KEY_API = "ai_spending.llm.api_key" _KEY_MODEL = "ai_spending.llm.model" _KEY_REGION = "ai_spending.llm.region" #: The model may suggest only these. ``transfer`` is deliberately absent #: from the model's vocabulary: choosing WHO something moves to is a #: naming decision about a colleague, and a suggestion box is the wrong #: place for it. _ALLOWED_RECOMMENDATIONS = (ITEM_DECISION_KEEP, ITEM_DECISION_REVOKE) FAILURE_NO_KEY = "AI_NO_KEY" FAILURE_CALL_FAILED = "AI_CALL_FAILED" FAILURE_UNGROUNDED = "AI_UNGROUNDED" FAILURE_EMPTY = "AI_EMPTY" _MAX_SUMMARY_CHARS = 900 _MAX_REASON_CHARS = 200 @dataclass(slots=True) class CategoryFact: resource_type: str total: int undecided: int examples: list[str] = field(default_factory=list) @dataclass(slots=True) class ReviewFacts: move_kind: str before_dept: str | None after_dept: str | None before_title: str | None after_title: str | None reviewable_items: int categories: list[CategoryFact] def citable_numbers(self) -> set[int]: """Every integer the narrative is allowed to contain. Published as a set rather than checked by regex against the prompt, because the failure this guards against is the model being *plausible* — "回收其中 40 项" reads fine and is false, and a manager acting on a review has no way to tell. """ numbers = {self.reviewable_items, len(self.categories)} for cat in self.categories: numbers.add(cat.total) numbers.add(cat.undecided) return numbers async def build_facts( db: AsyncSession, review: HrRoleChangeReview, ) -> ReviewFacts: """Deterministic facts for one packet. No model involved.""" rows = ( await db.execute( select( HrRoleChangeReviewItem.resource_type, HrRoleChangeReviewItem.label, HrRoleChangeReviewItem.decision, ).where( HrRoleChangeReviewItem.review_id == review.id, HrRoleChangeReviewItem.disposition == REVIEW, ) ) ).all() totals: Counter[str] = Counter() undecided: Counter[str] = Counter() examples: dict[str, list[str]] = {} for resource_type, label, decision in rows: totals[resource_type] += 1 if decision is None: undecided[resource_type] += 1 # Three examples per category. Enough for a human to recognise # what the category means in this person's case; few enough that # the prompt does not become the packet. bucket = examples.setdefault(resource_type, []) if len(bucket) < 3 and label: bucket.append(label[:80]) categories = [ CategoryFact( resource_type=rt, total=totals[rt], undecided=undecided[rt], examples=examples.get(rt, []), ) for rt in sorted(totals, key=lambda k: (-totals[k], k)) ] return ReviewFacts( move_kind=review.move_kind, before_dept=review.before_dept_path or review.before_dept_code, after_dept=review.after_dept_path or review.after_dept_code, before_title=review.before_job_title, after_title=review.after_job_title, reviewable_items=sum(totals.values()), categories=categories, ) _SYSTEM = ( "你在协助一位刚接手某位同事的**新经理**做转岗权限复核。" "Helix 已经把这位同事名下持有的东西按类别统计好了 —— 你拿到的就是全部依据。\n" "你的任务只有两件:\n" "① 用 3-5 句中文说明这次转岗**值得优先看哪几类**、为什么;\n" "② 对每个类别给一个初步建议(keep 或 revoke)和一句不超过 40 字的理由。\n\n" "硬性约束(违反即整份作废):\n" "- **只能使用给定事实里出现过的数字**。不要相加、不要估算、不要说「大约」。" "需要提到数量时,原样使用给定的那个数。\n" "- **只能对给定的类别作出建议**,不要发明类别。\n" "- 你给的是**建议**,不是决定。不要写「已回收」「将删除」这类完成态。\n" "- 不要提到任何具体的人名。\n" "- 拿不准的类别就建议 keep 并说明需要人工确认 —— 对一次转岗来说," "误留的代价是一次复核,误撤的代价是同事第二天上不了班。\n" "严格输出单个 JSON 对象。" ) def _user_prompt(facts: ReviewFacts) -> str: lines = [ "## 这次移动", f"- 类型: {facts.move_kind}", f"- 原部门: {facts.before_dept or '(未知)'}", f"- 新部门: {facts.after_dept or '(未知)'}", f"- 原职级: {facts.before_title or '(未知)'}", f"- 新职级: {facts.after_title or '(未知)'}", "", "## 待复核总量", f"- 需要决定的条目总数: {facts.reviewable_items}", f"- 涉及类别数: {len(facts.categories)}", "", "## 各类别(这是你能引用的全部数字)", ] for cat in facts.categories: sample = "、".join(cat.examples) if cat.examples else "(无示例)" lines.append( f"- {cat.resource_type}: 共 {cat.total} 项," f"其中 {cat.undecided} 项尚未决定。示例: {sample}" ) lines += [ "", "## 输出格式", '{"summary_zh": "...", "summary_en": "...", "categories": ' '[{"resource_type": "...", "recommendation": "keep|revoke", ' '"reason": "..."}]}', ] return "\n".join(lines) _INT_RE = re.compile(r"\d+") def _validate( payload: dict[str, Any], facts: ReviewFacts, ) -> tuple[str, str, dict[str, tuple[str, str]]]: """Return ``(summary_zh, summary_en, {resource_type: (rec, reason)})``. Raises ``ValueError`` when the reply is ungrounded. Rejecting the whole narrative rather than trimming it is deliberate: a summary with one invented number in it is not a summary with a small error, it is a summary nobody can trust the rest of. """ summary_zh = str(payload.get("summary_zh") or "").strip() summary_en = str(payload.get("summary_en") or "").strip() if not summary_zh: raise ValueError("empty summary") allowed = facts.citable_numbers() for text in (summary_zh, summary_en): for token in _INT_RE.findall(text): # Years and other 4-digit tokens are not counts; the prompt # forbids dates, and treating "2026" as a claim about the # packet would reject correct narratives. value = int(token) if len(token) >= 4: continue if value not in allowed: raise ValueError( f"ungrounded number {value} (published: {sorted(allowed)})", ) known = {c.resource_type for c in facts.categories} recommendations: dict[str, tuple[str, str]] = {} for entry in payload.get("categories") or []: if not isinstance(entry, dict): continue rtype = str(entry.get("resource_type") or "").strip() rec = str(entry.get("recommendation") or "").strip().lower() if rtype not in known or rec not in _ALLOWED_RECOMMENDATIONS: continue reason = str(entry.get("reason") or "").strip()[:_MAX_REASON_CHARS] recommendations[rtype] = (rec, reason) return summary_zh[:_MAX_SUMMARY_CHARS], summary_en[:_MAX_SUMMARY_CHARS], recommendations async def _client(db: AsyncSession): from app.domains.bailian.llm_client import BailianLLMClient async def _setting(key: str) -> str | None: row = await db.get(PlatformSetting, key) value = row.value if row else None return value if isinstance(value, str) else None api_key = (await _setting(_KEY_API) or "").strip() if not api_key: return None return BailianLLMClient( api_key=api_key, model=await _setting(_KEY_MODEL), region=await _setting(_KEY_REGION), timeout=45.0, purpose="role_change.review") async def annotate_review( db: AsyncSession, review: HrRoleChangeReview, ) -> HrRoleChangeReview: """Attach the model's narrative and per-category suggestions. Never raises. Every failure path records ``ai_failure_code`` and leaves the packet exactly as usable as it was — the deterministic per-category grouping is what a reviewer acts on, and the narrative is a convenience on top of it. """ review.ai_failure_code = None facts = await build_facts(db, review) if not facts.categories: review.ai_failure_code = FAILURE_EMPTY return review client = await _client(db) if client is None: review.ai_failure_code = FAILURE_NO_KEY return review try: payload = await client.chat_json( system=_SYSTEM, user=_user_prompt(facts), max_tokens=2000, temperature=0.2, ) except Exception as exc: # noqa: BLE001 — a packet without a narrative is fine logger.warning( "mover_review_ai: call failed for review=%s: %s", review.id, exc, ) review.ai_failure_code = FAILURE_CALL_FAILED return review try: summary_zh, summary_en, recommendations = _validate(payload, facts) except ValueError as exc: logger.warning( "mover_review_ai: ungrounded reply for review=%s: %s", review.id, exc, ) review.ai_failure_code = FAILURE_UNGROUNDED return review review.ai_summary_zh = summary_zh review.ai_summary_en = summary_en or None review.ai_model = getattr(client, "model", None) from datetime import UTC, datetime review.ai_generated_at = datetime.now(UTC) if recommendations: # Written per category onto the items that have no human answer # yet. An item somebody has already decided is never annotated: # a suggestion appearing next to a settled decision invites a # second look at a question that is closed. for rtype, (rec, reason) in recommendations.items(): items = ( await db.execute( select(HrRoleChangeReviewItem).where( HrRoleChangeReviewItem.review_id == review.id, HrRoleChangeReviewItem.resource_type == rtype, HrRoleChangeReviewItem.disposition == REVIEW, HrRoleChangeReviewItem.decision.is_(None), ) ) ).scalars().all() for item in items: item.ai_recommendation = rec item.ai_reason = reason or None return review def deterministic_priority(facts: ReviewFacts) -> list[str]: """Category order to show when there is no model narrative. Not a fallback that pretends to be the model — it is the ordering the surfaces use anyway. Volume first, because a category with 111 rows in it is where the reviewer's attention has to start, and a department move puts every category in play where a manager change mostly does not. """ if facts.move_kind == MOVE_DEPARTMENT: return [c.resource_type for c in facts.categories] # A manager change inside one department leaves ownership where it # is far more often than it moves it; access is the part that # genuinely re-opens. access_first = sorted( facts.categories, key=lambda c: (0 if "owner" not in c.resource_type else 1, -c.total), ) return [c.resource_type for c in access_first] __all__ = [ "FAILURE_CALL_FAILED", "FAILURE_EMPTY", "FAILURE_NO_KEY", "FAILURE_UNGROUNDED", "CategoryFact", "ReviewFacts", "annotate_review", "build_facts", "deterministic_priority", ]