generated from kgod/ai-review-template
372 lines
17 KiB
Python
372 lines
17 KiB
Python
"""Persistent light/deep optimization operations for ResumeAgent."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from typing import Any
|
|
|
|
from pydantic import ValidationError
|
|
from uuid import uuid4
|
|
|
|
from .claim_validator import validate_proposal
|
|
from .optimization_tiers import tier_config_for_session
|
|
from .fsm import FSMError
|
|
from .llm_services import LLMServiceError, log_ai_event
|
|
from .optimization_models import OptimizationRunView, OptimizationStartRequest
|
|
from .resume_document import (
|
|
DocumentError,
|
|
confirm_proposal,
|
|
entry_fingerprint,
|
|
find_entry,
|
|
reject_proposal,
|
|
set_pending_proposal,
|
|
)
|
|
from .resume_editing import _to_fsm
|
|
|
|
_OPTIMIZATION_EXCEPTIONS = (LLMServiceError, ValidationError, KeyError, TypeError, ValueError)
|
|
|
|
|
|
class OptimizationFlowMixin:
|
|
database: Any
|
|
experience_optimizer: Any
|
|
|
|
def set_target_position(self, session_id: str, target_position: str) -> dict[str, Any]:
|
|
"""Persist a user-confirmed target position from any recommendation source."""
|
|
normalized = target_position.strip()
|
|
if not normalized or len(normalized) > 32:
|
|
raise FSMError(
|
|
"invalid_target_position",
|
|
"Target position must be 1-32 characters",
|
|
status_code=422,
|
|
)
|
|
with self.database.transaction(immediate=True) as connection:
|
|
session = self._session_or_404(connection, session_id)
|
|
profile = dict(session["profile"])
|
|
profile["target_position"] = normalized
|
|
profile["target_position_confirmed"] = True
|
|
updated = self.database.update_session(
|
|
connection,
|
|
session_id,
|
|
stage=session["stage"],
|
|
profile=profile,
|
|
)
|
|
return {
|
|
"target_position": updated["profile"]["target_position"],
|
|
"target_position_confirmed": True,
|
|
}
|
|
|
|
def optimize_light(self, session_id: str, request: OptimizationStartRequest) -> OptimizationRunView:
|
|
with self.database.transaction(immediate=True) as connection:
|
|
session, resume, section, entry = self._entry(connection, session_id, request.entry_id)
|
|
context = self._context(session, section, request.instruction)
|
|
context["optimization_mode"] = "light"
|
|
facts = self._facts(entry)
|
|
try:
|
|
proposal = validate_proposal(
|
|
self.experience_optimizer.optimize(deepcopy(entry), context=context, facts=facts), facts
|
|
)
|
|
except _OPTIMIZATION_EXCEPTIONS as exc:
|
|
self._raise_optimization_ai_failed(exc, session_id, request.entry_id)
|
|
tier = tier_config_for_session(session)
|
|
gap_report: list[dict[str, Any]] | None = None
|
|
content = self._set_proposal(resume["content"], request.entry_id, proposal)
|
|
self.database.update_resume(connection, session_id, content)
|
|
run = self.database.create_optimization_run(
|
|
connection, run_id=f"opt_{uuid4().hex}", session_id=session_id,
|
|
entry_id=request.entry_id, mode="light", status="proposal_pending",
|
|
source_revision=resume["revision"],
|
|
state={
|
|
"facts": facts,
|
|
"star": proposal.get("star") or {},
|
|
"gap_report": gap_report,
|
|
"tier": tier.tier,
|
|
},
|
|
proposal=proposal,
|
|
)
|
|
return self._view(run, self._action_response(session, None))
|
|
|
|
def confirm_deep_optimization(self, session_id: str, run_id: str) -> OptimizationRunView:
|
|
with self.database.transaction(immediate=True) as connection:
|
|
session, resume, run = self._run(connection, session_id, run_id, "proposal_pending")
|
|
proposal = run.get("proposal") or {}
|
|
content = self._materialize_run_proposal(
|
|
resume["content"], run["entry_id"], proposal, run["source_revision"], resume["revision"]
|
|
)
|
|
try:
|
|
content = confirm_proposal(content, run["entry_id"])
|
|
except DocumentError as exc:
|
|
raise _to_fsm(exc) from exc
|
|
self.database.update_resume(connection, session_id, content)
|
|
run = self.database.update_optimization_run(
|
|
connection, session_id=session_id, run_id=run_id, status="confirmed",
|
|
state=run["state"], proposal=proposal,
|
|
)
|
|
return self._view(run, self._action_response(session, None))
|
|
|
|
def reject_optimization(self, session_id: str, run_id: str) -> OptimizationRunView:
|
|
with self.database.transaction(immediate=True) as connection:
|
|
session, resume, run = self._run(connection, session_id, run_id, "proposal_pending")
|
|
found = find_entry(resume["content"], run["entry_id"])
|
|
content = resume["content"]
|
|
if found is not None and "pending_proposal" in found[1]:
|
|
try:
|
|
content = reject_proposal(content, run["entry_id"])
|
|
except DocumentError as exc:
|
|
raise _to_fsm(exc) from exc
|
|
self.database.update_resume(connection, session_id, content)
|
|
run = self.database.update_optimization_run(
|
|
connection, session_id=session_id, run_id=run_id, status="rejected",
|
|
state=run["state"], proposal=run.get("proposal"),
|
|
)
|
|
return self._view(run, self._action_response(session, None))
|
|
|
|
def _materialize_run_proposal(
|
|
self,
|
|
content: dict[str, Any],
|
|
entry_id: str,
|
|
proposal: dict[str, Any],
|
|
source_revision: int,
|
|
current_revision: int,
|
|
) -> dict[str, Any]:
|
|
found = find_entry(content, entry_id)
|
|
if found is None:
|
|
raise FSMError("entry_not_found", "Entry not found in resume", status_code=404)
|
|
entry = found[1]
|
|
pending = entry.get("pending_proposal")
|
|
if isinstance(pending, dict):
|
|
if self._proposal_matches_run(pending, proposal, entry):
|
|
return content
|
|
raise FSMError(
|
|
"optimization_stale",
|
|
"Another proposal is pending for this entry; restart optimization",
|
|
status_code=409,
|
|
)
|
|
source_fingerprint = str(proposal.get("based_on") or "")
|
|
if source_fingerprint:
|
|
stale = source_fingerprint != entry_fingerprint(entry)
|
|
else:
|
|
stale = current_revision != source_revision
|
|
if stale:
|
|
raise FSMError(
|
|
"optimization_stale",
|
|
"Resume changed; restart optimization",
|
|
status_code=409,
|
|
)
|
|
materialized = self._set_proposal(content, entry_id, proposal)
|
|
refreshed = find_entry(materialized, entry_id)
|
|
if refreshed is None or refreshed[1]["pending_proposal"].get("based_on") != entry_fingerprint(entry):
|
|
raise FSMError("optimization_stale", "Resume changed; restart optimization", status_code=409)
|
|
return materialized
|
|
|
|
@staticmethod
|
|
def _proposal_value(proposal: dict[str, Any], key: str) -> Any:
|
|
"""Normalize optional proposal fields before checking an existing pending proposal."""
|
|
value = proposal.get(key)
|
|
if key in {"changes", "missing_facts", "unconfirmed_suggestions", "optional_enhancements", "validation_warnings", "omitted_fact_ids"}:
|
|
return list(value or [])
|
|
if key == "star":
|
|
return value or {}
|
|
return value
|
|
|
|
@classmethod
|
|
def _proposal_matches_run(
|
|
cls, pending: dict[str, Any], proposal: dict[str, Any], entry: dict[str, Any]
|
|
) -> bool:
|
|
source_fingerprint = str(proposal.get("based_on") or "")
|
|
if source_fingerprint and source_fingerprint != entry_fingerprint(entry):
|
|
return False
|
|
if pending.get("based_on") != (source_fingerprint or entry_fingerprint(entry)):
|
|
return False
|
|
return all(
|
|
cls._proposal_value(pending, key) == cls._proposal_value(proposal, key)
|
|
for key in (
|
|
"optimized_description",
|
|
"source",
|
|
"changes",
|
|
"missing_facts",
|
|
"unconfirmed_suggestions",
|
|
"optional_enhancements",
|
|
"validation_warnings",
|
|
"star",
|
|
"omitted_fact_ids",
|
|
)
|
|
)
|
|
|
|
def _optimization_failure_message(exc: Exception) -> str:
|
|
if not isinstance(exc, LLMServiceError):
|
|
reason_code = (
|
|
"structured_output_invalid"
|
|
if isinstance(exc, ValidationError)
|
|
else "optimization_state_invalid"
|
|
)
|
|
return f"AI \u4f18\u5316\u672a\u80fd\u751f\u6210\u53ef\u7528\u7ed3\u679c\uff08{reason_code}\uff09\uff0c\u8bf7\u7a0d\u540e\u91cd\u8bd5\u3002"
|
|
detail = str(exc.safe_summary or exc.reason_code)
|
|
messages = {
|
|
"equivalent_result": "\u6a21\u578b\u8fd4\u56de\u7684\u6539\u5199\u4e0e\u539f\u63cf\u8ff0\u53d8\u5316\u8fc7\u5c0f\uff0c\u8bf7\u8865\u5145\u5177\u4f53\u884c\u52a8\u6216\u7ed3\u679c\u540e\u91cd\u8bd5\u3002",
|
|
"structured_fields_only": "AI \u53ea\u8fd4\u56de\u4e86\u8868\u5355\u5b57\u6bb5\uff0c\u6ca1\u6709\u5f62\u6210\u7b80\u5386\u5316\u7684\u7ecf\u5386\u53d9\u8ff0\u3002\u8bf7\u8865\u5145\u7ecf\u5386\u63cf\u8ff0\u540e\u91cd\u8bd5\u3002",
|
|
"grounding_rejected": "AI \u4f18\u5316\u7a3f\u5305\u542b\u65e0\u6cd5\u7531\u5df2\u586b\u5199\u4fe1\u606f\u9a8c\u8bc1\u7684\u5185\u5bb9\uff0c\u5df2\u88ab\u62e6\u622a\u3002\u8bf7\u8865\u5145\u53ef\u786e\u8ba4\u7684\u4e8b\u5b9e\u540e\u91cd\u8bd5\u3002",
|
|
"insufficient_facts": "\u5f53\u524d\u53ef\u786e\u8ba4\u4fe1\u606f\u4e0d\u8db3\uff0c\u65e0\u6cd5\u751f\u6210\u53ef\u9a8c\u8bc1\u7684\u4f18\u5316\u7a3f\u3002\u8bf7\u8865\u5145\u4f60\u505a\u4e86\u4ec0\u4e48\u3001\u5982\u4f55\u5b8c\u6210\u6216\u6709\u4ec0\u4e48\u7ed3\u679c\u3002",
|
|
"empty_result": "AI \u670d\u52a1\u672a\u8fd4\u56de\u53ef\u7528\u7684\u4f18\u5316\u7a3f\uff0c\u8bf7\u7a0d\u540e\u91cd\u8bd5\u3002",
|
|
}
|
|
return messages.get(
|
|
detail,
|
|
f"AI \u4f18\u5316\u672a\u80fd\u751f\u6210\u53ef\u7528\u7ed3\u679c\uff08{exc.reason_code}\uff09\uff0c\u8bf7\u7a0d\u540e\u91cd\u8bd5\u3002",
|
|
)
|
|
|
|
@staticmethod
|
|
def _raise_optimization_ai_failed(
|
|
exc: Exception,
|
|
session_id: str,
|
|
entry_id: str,
|
|
run_id: str | None = None,
|
|
) -> None:
|
|
if isinstance(exc, LLMServiceError):
|
|
reason_code = exc.reason_code
|
|
trace_id = exc.trace_id
|
|
stage = exc.stage
|
|
else:
|
|
reason_code = (
|
|
"structured_output_invalid"
|
|
if isinstance(exc, ValidationError)
|
|
else "optimization_state_invalid"
|
|
)
|
|
trace_id = None
|
|
stage = "optimization_flow"
|
|
log_ai_event(
|
|
"deep_optimization_request_failed",
|
|
session_id=session_id,
|
|
entry_id=entry_id,
|
|
run_id=run_id,
|
|
reason_code=reason_code,
|
|
trace_id=trace_id,
|
|
stage=stage,
|
|
exception=type(exc).__name__,
|
|
)
|
|
raise FSMError(
|
|
"optimization_ai_failed",
|
|
OptimizationFlowMixin._optimization_failure_message(exc),
|
|
status_code=502,
|
|
) from exc
|
|
|
|
def _entry(self, connection: Any, session_id: str, entry_id: str) -> tuple[Any, Any, Any, Any]:
|
|
session, resume = self._session_or_404(connection, session_id), self._resume_or_409(connection, session_id)
|
|
found = find_entry(resume["content"], entry_id)
|
|
if found is None:
|
|
raise FSMError("entry_not_found", "Entry not found in resume", status_code=404)
|
|
return session, resume, found[0], found[1]
|
|
|
|
def list_active_optimization_runs(self, session_id: str) -> list[OptimizationRunView]:
|
|
with self.database.transaction() as connection:
|
|
self._session_or_404(connection, session_id)
|
|
runs = self.database.list_active_optimization_runs(connection, session_id)
|
|
return [self._view(run, None) for run in runs]
|
|
|
|
def _run(self, connection: Any, session_id: str, run_id: str, expected_status: str) -> tuple[Any, Any, Any]:
|
|
session, resume = self._session_or_404(connection, session_id), self._resume_or_409(connection, session_id)
|
|
run = self.database.fetch_optimization_run(connection, session_id, run_id)
|
|
if run is None:
|
|
raise FSMError("optimization_not_found", "Optimization run not found", status_code=404)
|
|
if run["status"] != expected_status:
|
|
raise FSMError("optimization_not_ready", "Optimization run is not ready for this action", status_code=409)
|
|
if expected_status == "question_pending" and run["source_revision"] != resume["revision"]:
|
|
raise FSMError("optimization_stale", "Resume changed; restart optimization", status_code=409)
|
|
return session, resume, run
|
|
|
|
@staticmethod
|
|
def _facts(entry: dict[str, Any]) -> list[dict[str, str]]:
|
|
keys = (
|
|
"title", "organization", "role", "company", "position", "project_name",
|
|
"project_role", "school", "major", "degree", "start_date",
|
|
"end_date_or_present", "name", "award", "date", "description",
|
|
)
|
|
facts: list[dict[str, str]] = []
|
|
for key in keys:
|
|
text = str(entry.get(key) or "").strip()
|
|
if text:
|
|
facts.append({
|
|
"id": f"fact_{len(facts) + 1}",
|
|
"source": "user_form",
|
|
"field": key,
|
|
"text": text,
|
|
})
|
|
return facts
|
|
|
|
@staticmethod
|
|
def _context(session: dict[str, Any], section: dict[str, Any], instruction: str | None) -> dict[str, Any]:
|
|
profile = session["profile"]
|
|
return {"job_type": profile.get("job_type"), "target_position": profile.get("target_position"), "major": (profile.get("anchor") or {}).get("major"), "entry_type": section.get("kind"), "instruction": instruction}
|
|
|
|
@staticmethod
|
|
def _set_proposal(content: dict[str, Any], entry_id: str, proposal: dict[str, Any]) -> dict[str, Any]:
|
|
optimized = str(proposal.get("optimized_description") or "").strip()
|
|
if not optimized:
|
|
raise FSMError(
|
|
"optimization_not_enough_facts",
|
|
"Please add an experience description or at least one usable experience field before optimizing.",
|
|
status_code=422,
|
|
)
|
|
try:
|
|
return set_pending_proposal(
|
|
content,
|
|
entry_id,
|
|
optimized,
|
|
source=proposal.get("source", "rule_structured"),
|
|
changes=proposal.get("changes") or [],
|
|
generation_source=proposal.get("generation_source"),
|
|
fallback_reason=proposal.get("fallback_reason"),
|
|
missing_facts=proposal.get("missing_facts"),
|
|
unconfirmed_suggestions=proposal.get("unconfirmed_suggestions"),
|
|
optional_enhancements=proposal.get("optional_enhancements"),
|
|
validation_warnings=proposal.get("validation_warnings"),
|
|
star=proposal.get("star"),
|
|
omitted_fact_ids=proposal.get("omitted_fact_ids"),
|
|
)
|
|
except DocumentError as exc:
|
|
raise _to_fsm(exc) from exc
|
|
def _view(self, run: dict[str, Any], action: Any) -> OptimizationRunView:
|
|
state = run["state"]
|
|
gaps = state.get("missing_dimensions") or []
|
|
remaining_high_priority_gaps = [
|
|
str(item.get("dimension") or "").strip()
|
|
for item in gaps
|
|
if isinstance(item, dict)
|
|
and item.get("priority") == "high"
|
|
and str(item.get("dimension") or "").strip()
|
|
]
|
|
gap_analysis = state.get("gap_analysis") or []
|
|
if gap_analysis:
|
|
remaining_high_priority_gaps = [
|
|
str(item.get("dimension") or "").strip()
|
|
for item in gap_analysis
|
|
if isinstance(item, dict)
|
|
and int(item.get("severity") or 0) >= 4
|
|
and str(item.get("dimension") or "").strip()
|
|
]
|
|
question = state.get("question") if isinstance(state.get("question"), dict) else None
|
|
completion = state.get("completion_decision") or {}
|
|
decision_source = (
|
|
state.get("decision_source")
|
|
or (question or {}).get("decision_source")
|
|
or completion.get("decision_source")
|
|
)
|
|
return OptimizationRunView(
|
|
id=run["id"],
|
|
mode=run["mode"],
|
|
status=run["status"],
|
|
entry_id=run["entry_id"],
|
|
question_count=int(state.get("question_count") or 0),
|
|
question=question,
|
|
proposal=run.get("proposal"),
|
|
covered_dimensions=[
|
|
str(item) for item in state.get("covered_dimensions") or [] if str(item).strip()
|
|
],
|
|
remaining_high_priority_gaps=list(dict.fromkeys(remaining_high_priority_gaps)),
|
|
decision_source=str(decision_source) if decision_source else None,
|
|
error_code=str(state.get("error_code")) if state.get("error_code") else None,
|
|
action=action,
|
|
gap_report=[dict(item) for item in state.get("gap_report") or []] or None,
|
|
tier=str(state.get("tier")) if state.get("tier") else None,
|
|
)
|
|
|
|
|