1210 lines
45 KiB
Python
1210 lines
45 KiB
Python
"""Evidence-grounded, privacy-minimising resume generation pipeline."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from collections.abc import Iterable, Mapping, Sequence
|
|
from enum import StrEnum
|
|
from pathlib import Path
|
|
from typing import Any, Literal, TypeVar, get_args
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, ValidationError, model_validator
|
|
|
|
from .backend import LLMBackend
|
|
from .models import (
|
|
CandidateProfile,
|
|
ContentPlan,
|
|
EvidenceItem,
|
|
EvidenceMap,
|
|
EvidenceMatchType,
|
|
GenerationConfig,
|
|
JobAnalysis,
|
|
JobPosting,
|
|
QualityCategory,
|
|
QualityFinding,
|
|
QualityReport,
|
|
QualitySeverity,
|
|
ResumeDraft,
|
|
ResumeMode,
|
|
)
|
|
from .output_constraints import as_quality_findings, validate_output_constraints
|
|
from .quality import (
|
|
RUBRIC_WEIGHTS,
|
|
apply_deterministic_score_caps,
|
|
compute_coverage,
|
|
compute_evaluation_fingerprint,
|
|
compute_evaluation_policy_fingerprint,
|
|
compute_weighted_overall,
|
|
)
|
|
from .validators import (
|
|
contains_blocking_posting_field,
|
|
contains_public_blind_origin,
|
|
validate_resume_draft,
|
|
)
|
|
|
|
|
|
ModelT = TypeVar("ModelT", bound=BaseModel)
|
|
|
|
_STAGE_PROMPTS = {
|
|
"analyze-job": "analyze-job.md",
|
|
"map-evidence": "map-evidence.md",
|
|
"plan-content": "plan-content.md",
|
|
"draft-resume": "draft-resume.md",
|
|
"evaluate-resume": "evaluate-resume.md",
|
|
"repair-resume": "repair-resume.md",
|
|
}
|
|
_BASE_PROMPT = "base-system.md"
|
|
_MAX_PROMPT_BYTES = 256_000
|
|
_REDACTION = "[REDACTED]"
|
|
_PRIVATE_FACT_TOKEN_MIN_LENGTH = 8
|
|
_PUBLIC_BLIND_SCHOOL_PATTERNS = (
|
|
re.compile(
|
|
r"(?<![가-힣A-Za-z0-9])"
|
|
r"[가-힣A-Za-z0-9·]{2,}(?:대학교|고등학교|중학교)"
|
|
r"(?=에서|의|를|은|는|이|가|졸업|재학|수료|[\s,.)]|$)",
|
|
re.IGNORECASE,
|
|
),
|
|
re.compile(
|
|
r"(?<![가-힣A-Za-z0-9])"
|
|
r"(?!최대|상대|절대|확대|세대|일대|휴대|군대|근대|현대)"
|
|
r"[가-힣]{1,8}대"
|
|
r"(?=에서|의|를|은|는|이|가|출신|졸업|재학|수료|전공|[\s,.)]|$)"
|
|
),
|
|
re.compile(
|
|
r"(?<![A-Za-z0-9])(?:SNU|KAIST|POSTECH|UNIST|GIST|DGIST)"
|
|
r"(?=에서|의|를|은|는|이|가|출신|졸업|재학|수료|전공|[\s,.)]|$)",
|
|
re.IGNORECASE,
|
|
),
|
|
)
|
|
|
|
# These correspond to the four release-critical areas in docs/quality-rubric.md.
|
|
MAJOR_QUALITY_CATEGORIES = frozenset(
|
|
{
|
|
QualityCategory.EVIDENCE,
|
|
QualityCategory.JOB_ALIGNMENT,
|
|
QualityCategory.KOREAN_LANGUAGE,
|
|
QualityCategory.PRIVACY,
|
|
}
|
|
)
|
|
MINIMUM_OVERALL_SCORE = 90.0
|
|
MINIMUM_EVIDENCE_COVERAGE = 1.0
|
|
MINIMUM_REQUIREMENT_COVERAGE = 0.80
|
|
MINIMUM_MAJOR_CATEGORY_SCORE = 80.0
|
|
|
|
|
|
class PipelineError(RuntimeError):
|
|
"""Base error for prompt, backend, or cross-stage contract failures."""
|
|
|
|
|
|
class PromptLoadError(PipelineError):
|
|
"""A required, repository-owned prompt could not be loaded safely."""
|
|
|
|
|
|
class BackendCallError(PipelineError):
|
|
"""A backend call failed or returned data outside its output schema."""
|
|
|
|
|
|
class StageIntegrityError(PipelineError):
|
|
"""A valid stage model contains broken cross-stage references."""
|
|
|
|
|
|
class PipelineStatus(StrEnum):
|
|
PASSED = "passed"
|
|
NEEDS_USER_INPUT = "needs_user_input"
|
|
|
|
|
|
class PipelineResult(BaseModel):
|
|
"""Approved canonical draft, or a bounded request for missing evidence."""
|
|
|
|
model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
|
|
|
|
status: PipelineStatus
|
|
analysis: JobAnalysis
|
|
evidence_map: EvidenceMap | None = None
|
|
content_plan: ContentPlan | None = None
|
|
draft: ResumeDraft | None = None
|
|
quality_report: QualityReport | None = None
|
|
deterministic_findings: list[QualityFinding] = Field(default_factory=list)
|
|
repair_attempts: int = Field(ge=0, le=2)
|
|
gate_failures: list[str] = Field(default_factory=list)
|
|
questions: list[str] = Field(default_factory=list, max_length=3)
|
|
|
|
@model_validator(mode="after")
|
|
def validate_outcome(self) -> "PipelineResult":
|
|
if self.status is PipelineStatus.PASSED:
|
|
if any(
|
|
artifact is None
|
|
for artifact in (
|
|
self.evidence_map,
|
|
self.content_plan,
|
|
self.draft,
|
|
self.quality_report,
|
|
)
|
|
):
|
|
raise ValueError("a passed pipeline result requires every artifact")
|
|
if self.gate_failures:
|
|
raise ValueError("a passed pipeline result cannot have gate failures")
|
|
if self.questions:
|
|
raise ValueError("a passed pipeline result cannot ask follow-up questions")
|
|
elif not self.questions:
|
|
raise ValueError("needs_user_input must include at least one question")
|
|
return self
|
|
|
|
@property
|
|
def passed(self) -> bool:
|
|
return self.status is PipelineStatus.PASSED
|
|
|
|
|
|
class _PromptLoader:
|
|
"""Small loader intentionally independent from the parallel prompts module."""
|
|
|
|
def __init__(self, prompt_dir: Path) -> None:
|
|
self._root = prompt_dir.expanduser().resolve()
|
|
if not self._root.is_dir():
|
|
raise PromptLoadError(f"prompt directory does not exist: {self._root}")
|
|
|
|
def read(self, filename: str) -> str:
|
|
if Path(filename).name != filename:
|
|
raise PromptLoadError("prompt filename must not contain a path")
|
|
path = (self._root / filename).resolve()
|
|
try:
|
|
path.relative_to(self._root)
|
|
except ValueError as exc:
|
|
raise PromptLoadError("prompt path escapes prompt directory") from exc
|
|
try:
|
|
size = path.stat().st_size
|
|
if size > _MAX_PROMPT_BYTES:
|
|
raise PromptLoadError(f"prompt is unexpectedly large: {filename}")
|
|
content = path.read_text(encoding="utf-8").strip()
|
|
except (OSError, UnicodeError) as exc:
|
|
raise PromptLoadError(f"could not read required prompt: {filename}") from exc
|
|
if not content:
|
|
raise PromptLoadError(f"required prompt is empty: {filename}")
|
|
return content
|
|
|
|
|
|
class _PrivacyContext:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
exact_tokens: Iterable[str],
|
|
phone_numbers: Iterable[str],
|
|
) -> None:
|
|
# Longest first prevents a short token from partially masking a longer one.
|
|
self.exact_tokens = tuple(
|
|
sorted(
|
|
{token.strip() for token in exact_tokens if token and token.strip()},
|
|
key=len,
|
|
reverse=True,
|
|
)
|
|
)
|
|
self.exact_patterns = tuple(
|
|
_private_token_pattern(token) for token in self.exact_tokens
|
|
)
|
|
self.phone_patterns = tuple(
|
|
re.compile(r"(?<!\d)" + r"[\s().-]*".join(map(re.escape, digits)) + r"(?!\d)")
|
|
for value in phone_numbers
|
|
if len(digits := re.sub(r"\D", "", value)) >= 8
|
|
)
|
|
|
|
def redact(self, value: Any) -> Any:
|
|
if isinstance(value, str):
|
|
redacted = value
|
|
for pattern in self.exact_patterns:
|
|
redacted = pattern.sub(_REDACTION, redacted)
|
|
for pattern in self.phone_patterns:
|
|
redacted = pattern.sub(_REDACTION, redacted)
|
|
return redacted
|
|
if isinstance(value, Mapping):
|
|
return {str(key): self.redact(item) for key, item in value.items()}
|
|
if isinstance(value, tuple):
|
|
return [self.redact(item) for item in value]
|
|
if isinstance(value, list):
|
|
return [self.redact(item) for item in value]
|
|
if isinstance(value, set):
|
|
return [self.redact(item) for item in sorted(value, key=str)]
|
|
return value
|
|
|
|
|
|
def _private_token_pattern(token: str) -> re.Pattern[str]:
|
|
r"""Compile a PII token without missing Korean postpositions.
|
|
|
|
Python's ``\w`` includes Hangul, so a generic word boundary does not match
|
|
``김하늘은``. Multi-character Hangul identity tokens are therefore matched
|
|
as literal substrings. Single-character names retain boundaries to avoid
|
|
erasing common Korean syllables throughout an otherwise safe payload.
|
|
"""
|
|
|
|
if re.search(r"[가-힣]", token) and len(token) >= 2:
|
|
compact_token = re.sub(r"\s+", "", token)
|
|
flexible_token = r"\s*".join(
|
|
re.escape(character) for character in compact_token
|
|
)
|
|
return re.compile(
|
|
r"(?<![가-힣A-Za-z0-9])"
|
|
+ flexible_token
|
|
+ r"(?=(?:은|는|이|가|을|를|의|에게|께서|으로|입니다|이라고|"
|
|
r"[\s,.)]|$))",
|
|
flags=re.I,
|
|
)
|
|
prefix = r"(?<!\w)" if token[0].isalnum() else ""
|
|
suffix = r"(?!\w)" if token[-1].isalnum() else ""
|
|
return re.compile(prefix + re.escape(token) + suffix, flags=re.I)
|
|
|
|
|
|
def _model_payload(model: BaseModel) -> dict[str, Any]:
|
|
# ``exclude_computed_fields`` is newer than the project's Pydantic floor.
|
|
# Excluding declared computed names explicitly keeps compatibility with
|
|
# Pydantic 2.10 while avoiding read-only values in LLM payloads.
|
|
dumped = model.model_dump(mode="json")
|
|
return _strip_computed_fields(dumped, type(model))
|
|
|
|
|
|
def _nested_model_types(annotation: Any) -> tuple[type[BaseModel], ...]:
|
|
if isinstance(annotation, type) and issubclass(annotation, BaseModel):
|
|
return (annotation,)
|
|
discovered: list[type[BaseModel]] = []
|
|
for argument in get_args(annotation):
|
|
for model_type in _nested_model_types(argument):
|
|
if model_type not in discovered:
|
|
discovered.append(model_type)
|
|
return tuple(discovered)
|
|
|
|
|
|
def _strip_computed_fields(
|
|
value: Mapping[str, Any], model_type: type[BaseModel]
|
|
) -> dict[str, Any]:
|
|
"""Remove only declared read-only fields, preserving strict extra checks."""
|
|
|
|
cleaned = dict(value)
|
|
for field_name in model_type.model_computed_fields:
|
|
cleaned.pop(field_name, None)
|
|
for field_name, field in model_type.model_fields.items():
|
|
if field_name not in cleaned:
|
|
continue
|
|
nested_types = _nested_model_types(field.annotation)
|
|
if not nested_types:
|
|
continue
|
|
nested_value = cleaned[field_name]
|
|
if isinstance(nested_value, Mapping):
|
|
for nested_type in nested_types:
|
|
nested_value = _strip_computed_fields(nested_value, nested_type)
|
|
cleaned[field_name] = nested_value
|
|
elif isinstance(nested_value, (list, tuple)):
|
|
items: list[Any] = []
|
|
for item in nested_value:
|
|
if isinstance(item, Mapping):
|
|
for nested_type in nested_types:
|
|
item = _strip_computed_fields(item, nested_type)
|
|
items.append(item)
|
|
cleaned[field_name] = items
|
|
return cleaned
|
|
|
|
|
|
def _visible_facts(
|
|
profile: CandidateProfile,
|
|
config: GenerationConfig,
|
|
analysis: JobAnalysis | None = None,
|
|
) -> tuple[EvidenceItem, ...]:
|
|
"""Return only evidence that may cross the LLM boundary."""
|
|
|
|
# Consent controls whether a renderer may insert a sensitive value; it does
|
|
# not grant an external generation model access to that value. Keeping this
|
|
# boundary unconditional also prevents an employer-form toggle from
|
|
# silently broadening the model's data access.
|
|
visible: list[EvidenceItem] = []
|
|
for fact in profile.facts:
|
|
if fact.confidential or fact.sensitive_category is not None:
|
|
continue
|
|
if config.resume_mode is ResumeMode.PUBLIC_BLIND:
|
|
blind_text = " ".join(
|
|
[
|
|
fact.content,
|
|
*(str(key) for key in fact.metrics),
|
|
*(str(value) for value in fact.metrics.values()),
|
|
*fact.keywords,
|
|
]
|
|
)
|
|
if any(
|
|
pattern.search(blind_text)
|
|
for pattern in _PUBLIC_BLIND_SCHOOL_PATTERNS
|
|
) or contains_public_blind_origin(blind_text):
|
|
continue
|
|
fact_text = " ".join(
|
|
[
|
|
fact.content,
|
|
*(str(key) for key in fact.metrics),
|
|
*(str(value) for value in fact.metrics.values()),
|
|
*fact.keywords,
|
|
]
|
|
)
|
|
if (
|
|
analysis is not None
|
|
and contains_blocking_posting_field(fact_text, profile, analysis)
|
|
):
|
|
continue
|
|
visible.append(fact)
|
|
return tuple(visible)
|
|
|
|
|
|
def _privacy_context(
|
|
profile: CandidateProfile,
|
|
excluded_facts: Sequence[EvidenceItem],
|
|
) -> _PrivacyContext:
|
|
contact = profile.contact
|
|
identity_tokens = [
|
|
profile.name,
|
|
profile.name_en or "",
|
|
contact.email or "",
|
|
*contact.links,
|
|
]
|
|
# If an excluded fact is echoed by a malformed draft, it must still not be
|
|
# sent back to the judge or repairer. IDs are intentionally not tokens:
|
|
# reference integrity rejects them and common short IDs can collide with job
|
|
# text. The actual private fact values are redacted.
|
|
for fact in excluded_facts:
|
|
private_values = [fact.content, fact.source_reference or ""]
|
|
identity_tokens.extend(
|
|
value
|
|
for value in private_values
|
|
if len(value.strip()) >= _PRIVATE_FACT_TOKEN_MIN_LENGTH
|
|
)
|
|
return _PrivacyContext(
|
|
exact_tokens=identity_tokens,
|
|
phone_numbers=[contact.phone] if contact.phone else [],
|
|
)
|
|
|
|
|
|
def _candidate_facts_payload(facts: Sequence[EvidenceItem]) -> list[dict[str, Any]]:
|
|
"""Project only generation-relevant fields across the model boundary."""
|
|
|
|
allowed_fields = (
|
|
"evidence_id",
|
|
"category",
|
|
"content",
|
|
"date_range",
|
|
"verification_status",
|
|
"metrics",
|
|
"keywords",
|
|
)
|
|
payload: list[dict[str, Any]] = []
|
|
for fact in facts:
|
|
dumped = _model_payload(fact)
|
|
payload.append({field: dumped[field] for field in allowed_fields})
|
|
return payload
|
|
|
|
|
|
class ResumePipeline:
|
|
"""Orchestrate structured generation, validation, judging, and repair."""
|
|
|
|
def __init__(
|
|
self,
|
|
backend: LLMBackend,
|
|
*,
|
|
prompt_dir: str | Path | None = None,
|
|
max_repair_attempts: int = 2,
|
|
) -> None:
|
|
if not 0 <= max_repair_attempts <= 2:
|
|
raise ValueError("max_repair_attempts must be between 0 and 2")
|
|
self._backend = backend
|
|
# Keep runtime assets inside the import package so wheel installations do
|
|
# not depend on a repository-level directory that is absent after install.
|
|
default_prompt_dir = Path(__file__).resolve().with_name("prompt_templates")
|
|
self._prompts = _PromptLoader(
|
|
Path(prompt_dir) if prompt_dir is not None else default_prompt_dir
|
|
)
|
|
self._system_prompt = self._prompts.read(_BASE_PROMPT)
|
|
self._task_prompts = {
|
|
stage: self._prompts.read(filename)
|
|
for stage, filename in _STAGE_PROMPTS.items()
|
|
}
|
|
self._evaluation_policy_fingerprint = (
|
|
compute_evaluation_policy_fingerprint(
|
|
system_prompt=self._system_prompt,
|
|
evaluator_prompt=self._task_prompts["evaluate-resume"],
|
|
)
|
|
)
|
|
self._max_repair_attempts = max_repair_attempts
|
|
|
|
def run(
|
|
self,
|
|
profile: CandidateProfile | Mapping[str, Any],
|
|
posting: JobPosting | Mapping[str, Any],
|
|
config: GenerationConfig | Mapping[str, Any],
|
|
) -> PipelineResult:
|
|
"""Generate and gate one canonical resume draft."""
|
|
|
|
validated_profile = CandidateProfile.model_validate(profile)
|
|
validated_posting = JobPosting.model_validate(posting)
|
|
validated_config = GenerationConfig.model_validate(config)
|
|
if not validated_config.strict_evidence:
|
|
raise PipelineError(
|
|
"release pipeline requires strict_evidence=true; false is lint-only"
|
|
)
|
|
validated_config.assert_profile_compatible(validated_profile)
|
|
|
|
initial_visible_facts = _visible_facts(
|
|
validated_profile, validated_config
|
|
)
|
|
initial_visible_ids = {
|
|
fact.evidence_id for fact in initial_visible_facts
|
|
}
|
|
initial_excluded_facts = tuple(
|
|
fact
|
|
for fact in validated_profile.facts
|
|
if fact.evidence_id not in initial_visible_ids
|
|
)
|
|
privacy = _privacy_context(validated_profile, initial_excluded_facts)
|
|
|
|
analysis = self._invoke(
|
|
"analyze-job",
|
|
JobAnalysis,
|
|
{"job_posting": _model_payload(validated_posting)},
|
|
privacy,
|
|
)
|
|
self._integrity(
|
|
"analyze-job", lambda: analysis.assert_matches_posting(validated_posting)
|
|
)
|
|
|
|
# Posting-specific blind/redaction rules are known only after analysis.
|
|
# Recompute the allowlist before any candidate fact crosses the model
|
|
# boundary, then use the same rules again during final validation.
|
|
visible_facts = _visible_facts(
|
|
validated_profile, validated_config, analysis
|
|
)
|
|
visible_ids = {fact.evidence_id for fact in visible_facts}
|
|
excluded_facts = tuple(
|
|
fact
|
|
for fact in validated_profile.facts
|
|
if fact.evidence_id not in visible_ids
|
|
)
|
|
privacy = _privacy_context(validated_profile, excluded_facts)
|
|
facts_payload = _candidate_facts_payload(visible_facts)
|
|
|
|
if not visible_facts:
|
|
return PipelineResult(
|
|
status=PipelineStatus.NEEDS_USER_INPUT,
|
|
analysis=analysis,
|
|
repair_attempts=0,
|
|
gate_failures=["no_generation_safe_evidence"],
|
|
questions=[
|
|
"지원 직무와 관련된 경력·프로젝트·교육 사실을 최소 1개 제공해 주세요."
|
|
],
|
|
)
|
|
|
|
evidence_map = self._invoke(
|
|
"map-evidence",
|
|
EvidenceMap,
|
|
{
|
|
"job_analysis": _model_payload(analysis),
|
|
"candidate_facts": facts_payload,
|
|
},
|
|
privacy,
|
|
)
|
|
self._integrity(
|
|
"map-evidence",
|
|
lambda: evidence_map.assert_referential_integrity(
|
|
validated_profile, analysis
|
|
),
|
|
)
|
|
self._assert_visible_references(
|
|
"map-evidence",
|
|
(
|
|
evidence_id
|
|
for match in evidence_map.matches
|
|
for evidence_id in match.evidence_ids
|
|
),
|
|
visible_ids,
|
|
)
|
|
|
|
if all(
|
|
match.match_type is EvidenceMatchType.GAP
|
|
for match in evidence_map.matches
|
|
):
|
|
required_by_id = {
|
|
requirement.requirement_id: requirement
|
|
for requirement in analysis.requirements
|
|
}
|
|
questions = [
|
|
(
|
|
f"{required_by_id[match.requirement_id].text!r}을(를) 입증할 "
|
|
"구체적인 경험·기간·역할·결과가 있나요?"
|
|
)
|
|
for match in evidence_map.matches
|
|
if match.requirement_id in required_by_id
|
|
][:3]
|
|
return PipelineResult(
|
|
status=PipelineStatus.NEEDS_USER_INPUT,
|
|
analysis=analysis,
|
|
evidence_map=evidence_map,
|
|
repair_attempts=0,
|
|
gate_failures=["all_requirements_gap"],
|
|
questions=questions
|
|
or ["공고 요건과 연결할 수 있는 검증 가능한 경험을 제공해 주세요."],
|
|
)
|
|
|
|
content_plan = self._invoke(
|
|
"plan-content",
|
|
ContentPlan,
|
|
{
|
|
"candidate_id": validated_profile.candidate_id,
|
|
"job_analysis": _model_payload(analysis),
|
|
"evidence_map": _model_payload(evidence_map),
|
|
"generation_config": _model_payload(validated_config),
|
|
},
|
|
privacy,
|
|
)
|
|
self._integrity(
|
|
"plan-content",
|
|
lambda: content_plan.assert_referential_integrity(
|
|
validated_profile, analysis
|
|
),
|
|
)
|
|
self._integrity(
|
|
"plan-content",
|
|
lambda: content_plan.assert_matches_evidence_map(evidence_map),
|
|
)
|
|
if content_plan.mode is not validated_config.resume_mode:
|
|
raise StageIntegrityError(
|
|
"plan-content: content plan mode does not match generation config"
|
|
)
|
|
self._assert_visible_references(
|
|
"plan-content",
|
|
(
|
|
evidence_id
|
|
for section in content_plan.sections
|
|
for evidence_id in section.evidence_ids
|
|
),
|
|
visible_ids,
|
|
)
|
|
|
|
draft = self._invoke(
|
|
"draft-resume",
|
|
ResumeDraft,
|
|
{
|
|
"candidate_id": validated_profile.candidate_id,
|
|
"candidate_facts": facts_payload,
|
|
"job_analysis": _model_payload(analysis),
|
|
"content_plan": _model_payload(content_plan),
|
|
"generation_config": _model_payload(validated_config),
|
|
},
|
|
privacy,
|
|
)
|
|
self._assert_draft_integrity(
|
|
draft,
|
|
profile=validated_profile,
|
|
analysis=analysis,
|
|
evidence_map=evidence_map,
|
|
plan=content_plan,
|
|
config=validated_config,
|
|
visible_ids=visible_ids,
|
|
stage="draft-resume",
|
|
)
|
|
repair_evidence_ceiling = {
|
|
evidence_id
|
|
for section in draft.sections
|
|
for claim in section.claims
|
|
for evidence_id in claim.evidence_ids
|
|
}
|
|
|
|
repairs = 0
|
|
while True:
|
|
deterministic_findings = self._deterministic_findings(
|
|
validated_profile,
|
|
draft,
|
|
validated_config,
|
|
analysis,
|
|
visible_ids,
|
|
)
|
|
report = self._invoke(
|
|
"evaluate-resume",
|
|
QualityReport,
|
|
{
|
|
"candidate_facts": facts_payload,
|
|
"job_analysis": _model_payload(analysis),
|
|
"resume_draft": _model_payload(draft),
|
|
"deterministic_findings": [
|
|
_model_payload(finding)
|
|
for finding in deterministic_findings
|
|
],
|
|
"quality_rubric": self._quality_rubric(validated_config),
|
|
},
|
|
privacy,
|
|
)
|
|
# The backend judges content; the trusted harness binds the report
|
|
# to the exact canonical draft and computes objective coverage after
|
|
# structured validation. Neither value is trusted to the judge.
|
|
coverage = compute_coverage(
|
|
draft, analysis, evidence_map, validated_profile
|
|
)
|
|
trusted_category_scores = apply_deterministic_score_caps(
|
|
report.category_scores, deterministic_findings
|
|
)
|
|
try:
|
|
weighted_overall = compute_weighted_overall(trusted_category_scores)
|
|
except ValueError as exc:
|
|
raise StageIntegrityError(
|
|
"evaluate-resume: quality report omitted weighted rubric categories"
|
|
) from exc
|
|
evaluation_fingerprint = compute_evaluation_fingerprint(
|
|
validated_profile,
|
|
draft,
|
|
validated_posting,
|
|
analysis,
|
|
evidence_map,
|
|
content_plan,
|
|
validated_config,
|
|
policy_fingerprint=self._evaluation_policy_fingerprint,
|
|
)
|
|
report = report.model_copy(
|
|
update={
|
|
"category_scores": trusted_category_scores,
|
|
"draft_fingerprint": draft.fingerprint(),
|
|
"evaluation_fingerprint": evaluation_fingerprint,
|
|
"overall_score": weighted_overall,
|
|
"evidence_coverage": coverage.evidence,
|
|
"requirement_coverage": coverage.requirements,
|
|
"minimum_score": max(
|
|
MINIMUM_OVERALL_SCORE,
|
|
validated_config.minimum_quality_score,
|
|
),
|
|
"minimum_evidence_coverage": max(
|
|
MINIMUM_EVIDENCE_COVERAGE,
|
|
validated_config.minimum_evidence_coverage,
|
|
),
|
|
"minimum_requirement_coverage": max(
|
|
MINIMUM_REQUIREMENT_COVERAGE,
|
|
validated_config.minimum_requirement_coverage,
|
|
),
|
|
}
|
|
)
|
|
self._assert_report_integrity(
|
|
report,
|
|
draft,
|
|
visible_ids,
|
|
evaluation_fingerprint=evaluation_fingerprint,
|
|
stage="evaluate-resume",
|
|
)
|
|
gate_failures = self._gate_failures(
|
|
deterministic_findings, report, validated_config
|
|
)
|
|
if not gate_failures:
|
|
return PipelineResult(
|
|
status=PipelineStatus.PASSED,
|
|
analysis=analysis,
|
|
evidence_map=evidence_map,
|
|
content_plan=content_plan,
|
|
draft=draft,
|
|
quality_report=report,
|
|
deterministic_findings=deterministic_findings,
|
|
repair_attempts=repairs,
|
|
)
|
|
|
|
repair_findings = self._claim_repair_findings(
|
|
deterministic_findings, report.findings, draft
|
|
)
|
|
if repairs >= self._max_repair_attempts or not repair_findings:
|
|
return PipelineResult(
|
|
status=PipelineStatus.NEEDS_USER_INPUT,
|
|
analysis=analysis,
|
|
evidence_map=evidence_map,
|
|
content_plan=content_plan,
|
|
draft=draft,
|
|
quality_report=report,
|
|
deterministic_findings=deterministic_findings,
|
|
repair_attempts=repairs,
|
|
gate_failures=gate_failures,
|
|
questions=self._questions(
|
|
repair_findings,
|
|
gate_failures,
|
|
report,
|
|
),
|
|
)
|
|
|
|
repaired = self._invoke(
|
|
"repair-resume",
|
|
ResumeDraft,
|
|
{
|
|
"candidate_facts": facts_payload,
|
|
"resume_draft": _model_payload(draft),
|
|
"approved_findings": [
|
|
_model_payload(finding) for finding in repair_findings
|
|
],
|
|
"generation_config": _model_payload(validated_config),
|
|
},
|
|
privacy,
|
|
)
|
|
targeted_claim_ids = {
|
|
finding.claim_id
|
|
for finding in repair_findings
|
|
if finding.claim_id is not None
|
|
}
|
|
self._assert_draft_integrity(
|
|
repaired,
|
|
profile=validated_profile,
|
|
analysis=analysis,
|
|
evidence_map=evidence_map,
|
|
plan=content_plan,
|
|
config=validated_config,
|
|
visible_ids=visible_ids,
|
|
stage="repair-resume",
|
|
)
|
|
self._assert_repair_scope(
|
|
previous=draft,
|
|
repaired=repaired,
|
|
targeted_claim_ids=targeted_claim_ids,
|
|
evidence_ceiling=repair_evidence_ceiling,
|
|
)
|
|
draft = repaired
|
|
repairs += 1
|
|
|
|
def generate(
|
|
self,
|
|
profile: CandidateProfile | Mapping[str, Any],
|
|
posting: JobPosting | Mapping[str, Any],
|
|
config: GenerationConfig | Mapping[str, Any],
|
|
) -> PipelineResult:
|
|
"""Compatibility-friendly synonym for :meth:`run`."""
|
|
|
|
return self.run(profile, posting, config)
|
|
|
|
def _invoke(
|
|
self,
|
|
stage: str,
|
|
output_model: type[ModelT],
|
|
user_payload: Mapping[str, Any],
|
|
privacy: _PrivacyContext,
|
|
) -> ModelT:
|
|
safe_payload = privacy.redact(user_payload)
|
|
try:
|
|
raw = self._backend.complete_json(
|
|
stage=stage,
|
|
system_prompt=self._system_prompt,
|
|
task_prompt=self._task_prompts[stage],
|
|
user_payload=safe_payload,
|
|
output_model=output_model,
|
|
)
|
|
except Exception as exc:
|
|
raise BackendCallError(f"{stage}: backend call failed") from exc
|
|
|
|
if isinstance(raw, BaseModel):
|
|
candidate: Any = {
|
|
field_name: getattr(raw, field_name)
|
|
for field_name in type(raw).model_fields
|
|
}
|
|
elif isinstance(raw, Mapping):
|
|
candidate = _strip_computed_fields(raw, output_model)
|
|
else:
|
|
raise BackendCallError(
|
|
f"{stage}: backend returned neither a model nor a mapping"
|
|
)
|
|
try:
|
|
return output_model.model_validate(candidate)
|
|
except (ValidationError, TypeError, ValueError) as exc:
|
|
# Do not interpolate raw output: it may contain candidate PII.
|
|
raise BackendCallError(
|
|
f"{stage}: backend output failed {output_model.__name__} validation"
|
|
) from exc
|
|
|
|
@staticmethod
|
|
def _integrity(stage: str, check: Any) -> None:
|
|
try:
|
|
check()
|
|
except (ValidationError, TypeError, ValueError) as exc:
|
|
raise StageIntegrityError(f"{stage}: {exc}") from exc
|
|
|
|
@staticmethod
|
|
def _assert_visible_references(
|
|
stage: str,
|
|
references: Iterable[str],
|
|
visible_ids: set[str],
|
|
) -> None:
|
|
hidden = sorted(set(references) - visible_ids)
|
|
if hidden:
|
|
raise StageIntegrityError(
|
|
f"{stage}: references evidence withheld by privacy policy: {hidden}"
|
|
)
|
|
|
|
def _assert_draft_integrity(
|
|
self,
|
|
draft: ResumeDraft,
|
|
*,
|
|
profile: CandidateProfile,
|
|
analysis: JobAnalysis,
|
|
evidence_map: EvidenceMap,
|
|
plan: ContentPlan,
|
|
config: GenerationConfig,
|
|
visible_ids: set[str],
|
|
stage: str,
|
|
) -> None:
|
|
self._integrity(
|
|
stage, lambda: draft.assert_referential_integrity(profile, analysis)
|
|
)
|
|
self._integrity(
|
|
stage, lambda: draft.assert_matches_evidence_map(evidence_map)
|
|
)
|
|
if draft.mode is not config.resume_mode:
|
|
raise StageIntegrityError(
|
|
f"{stage}: draft mode does not match generation config"
|
|
)
|
|
if draft.mode is not plan.mode:
|
|
raise StageIntegrityError(f"{stage}: draft mode does not match content plan")
|
|
|
|
plan_by_id = {section.section_id: section for section in plan.sections}
|
|
draft_by_id = {section.section_id: section for section in draft.sections}
|
|
if set(plan_by_id) != set(draft_by_id):
|
|
raise StageIntegrityError(
|
|
f"{stage}: draft sections do not match planned sections"
|
|
)
|
|
all_references: list[str] = []
|
|
for section_id, section in draft_by_id.items():
|
|
planned = plan_by_id[section_id]
|
|
if section.section_type is not planned.section_type:
|
|
raise StageIntegrityError(
|
|
f"{stage}: section {section_id!r} changed planned type"
|
|
)
|
|
if section.order != planned.order:
|
|
raise StageIntegrityError(
|
|
f"{stage}: section {section_id!r} changed planned order"
|
|
)
|
|
if len(section.claims) > planned.bullet_budget:
|
|
raise StageIntegrityError(
|
|
f"{stage}: section {section_id!r} exceeds bullet budget"
|
|
)
|
|
allowed_evidence = set(planned.evidence_ids)
|
|
allowed_requirements = set(planned.requirement_ids)
|
|
for claim in section.claims:
|
|
all_references.extend(claim.evidence_ids)
|
|
if not set(claim.evidence_ids) <= allowed_evidence:
|
|
raise StageIntegrityError(
|
|
f"{stage}: claim {claim.claim_id!r} uses unplanned evidence"
|
|
)
|
|
if not set(claim.requirement_ids) <= allowed_requirements:
|
|
raise StageIntegrityError(
|
|
f"{stage}: claim {claim.claim_id!r} uses unplanned requirements"
|
|
)
|
|
self._assert_visible_references(stage, all_references, visible_ids)
|
|
|
|
def _deterministic_findings(
|
|
self,
|
|
profile: CandidateProfile,
|
|
draft: ResumeDraft,
|
|
config: GenerationConfig,
|
|
analysis: JobAnalysis,
|
|
visible_ids: set[str],
|
|
) -> list[QualityFinding]:
|
|
try:
|
|
raw_findings = validate_resume_draft(
|
|
profile, draft, config, analysis=analysis
|
|
)
|
|
findings = [QualityFinding.model_validate(item) for item in raw_findings]
|
|
findings.extend(
|
|
as_quality_findings(
|
|
validate_output_constraints(
|
|
draft,
|
|
config,
|
|
analysis=analysis,
|
|
)
|
|
)
|
|
)
|
|
except (ValidationError, TypeError, ValueError) as exc:
|
|
raise StageIntegrityError(
|
|
"deterministic-validate: validator returned invalid findings"
|
|
) from exc
|
|
self._assert_findings_integrity(
|
|
findings, draft, visible_ids, stage="deterministic-validate"
|
|
)
|
|
return findings
|
|
|
|
def _assert_report_integrity(
|
|
self,
|
|
report: QualityReport,
|
|
draft: ResumeDraft,
|
|
visible_ids: set[str],
|
|
*,
|
|
evaluation_fingerprint: str,
|
|
stage: str,
|
|
) -> None:
|
|
if report.draft_id != draft.draft_id:
|
|
raise StageIntegrityError(
|
|
f"{stage}: quality report references a different draft"
|
|
)
|
|
if report.draft_fingerprint != draft.fingerprint():
|
|
raise StageIntegrityError(
|
|
f"{stage}: quality report fingerprint does not match draft content"
|
|
)
|
|
if report.evaluation_fingerprint != evaluation_fingerprint:
|
|
raise StageIntegrityError(
|
|
f"{stage}: quality report fingerprint does not match evaluation context"
|
|
)
|
|
self._assert_findings_integrity(report.findings, draft, visible_ids, stage)
|
|
|
|
@staticmethod
|
|
def _assert_findings_integrity(
|
|
findings: Sequence[QualityFinding],
|
|
draft: ResumeDraft,
|
|
visible_ids: set[str],
|
|
stage: str,
|
|
) -> None:
|
|
known_claims = {
|
|
claim.claim_id for section in draft.sections for claim in section.claims
|
|
}
|
|
errors: list[str] = []
|
|
for finding in findings:
|
|
if finding.claim_id is None and finding.location is None:
|
|
errors.append(
|
|
f"finding {finding.finding_id!r} has no claim_id or location"
|
|
)
|
|
if finding.claim_id is not None and finding.claim_id not in known_claims:
|
|
errors.append(
|
|
f"finding {finding.finding_id!r} references unknown claim"
|
|
)
|
|
hidden = sorted(set(finding.evidence_ids) - visible_ids)
|
|
if hidden:
|
|
errors.append(
|
|
f"finding {finding.finding_id!r} references unknown evidence {hidden}"
|
|
)
|
|
if errors:
|
|
raise StageIntegrityError(f"{stage}: {'; '.join(errors)}")
|
|
|
|
@staticmethod
|
|
def _quality_rubric(config: GenerationConfig) -> dict[str, Any]:
|
|
return {
|
|
"dimension_weights": {
|
|
category.value: weight
|
|
for category, weight in RUBRIC_WEIGHTS.items()
|
|
},
|
|
"minimum_overall_score": max(
|
|
MINIMUM_OVERALL_SCORE, config.minimum_quality_score
|
|
),
|
|
"minimum_evidence_coverage": max(
|
|
MINIMUM_EVIDENCE_COVERAGE, config.minimum_evidence_coverage
|
|
),
|
|
"minimum_requirement_coverage": max(
|
|
MINIMUM_REQUIREMENT_COVERAGE,
|
|
config.minimum_requirement_coverage,
|
|
),
|
|
"minimum_major_category_score": MINIMUM_MAJOR_CATEGORY_SCORE,
|
|
"major_categories": sorted(
|
|
category.value for category in MAJOR_QUALITY_CATEGORIES
|
|
),
|
|
}
|
|
|
|
@staticmethod
|
|
def _gate_failures(
|
|
deterministic: Sequence[QualityFinding],
|
|
report: QualityReport,
|
|
config: GenerationConfig,
|
|
) -> list[str]:
|
|
failures: list[str] = []
|
|
blocking_rules = sorted(
|
|
{finding.code for finding in deterministic if finding.blocking}
|
|
)
|
|
if blocking_rules:
|
|
failures.append(
|
|
"deterministic_blocking:" + ",".join(blocking_rules)
|
|
)
|
|
judge_blocking = sorted(
|
|
{finding.code for finding in report.findings if finding.blocking}
|
|
)
|
|
if judge_blocking:
|
|
failures.append("judge_blocking:" + ",".join(judge_blocking))
|
|
|
|
required_score = max(MINIMUM_OVERALL_SCORE, config.minimum_quality_score)
|
|
if report.overall_score < required_score:
|
|
failures.append(
|
|
f"overall_score:{report.overall_score:g}<{required_score:g}"
|
|
)
|
|
required_evidence = max(
|
|
MINIMUM_EVIDENCE_COVERAGE, config.minimum_evidence_coverage
|
|
)
|
|
if report.evidence_coverage < required_evidence:
|
|
failures.append(
|
|
"evidence_coverage:"
|
|
f"{report.evidence_coverage:g}<{required_evidence:g}"
|
|
)
|
|
required_requirements = max(
|
|
MINIMUM_REQUIREMENT_COVERAGE,
|
|
config.minimum_requirement_coverage,
|
|
)
|
|
if report.requirement_coverage < required_requirements:
|
|
failures.append(
|
|
"requirement_coverage:"
|
|
f"{report.requirement_coverage:g}<{required_requirements:g}"
|
|
)
|
|
for category in sorted(MAJOR_QUALITY_CATEGORIES, key=lambda item: item.value):
|
|
score = report.category_scores.get(category)
|
|
if score is None:
|
|
failures.append(f"major_category_missing:{category.value}")
|
|
elif score < MINIMUM_MAJOR_CATEGORY_SCORE:
|
|
failures.append(
|
|
f"major_category:{category.value}:"
|
|
f"{score:g}<{MINIMUM_MAJOR_CATEGORY_SCORE:g}"
|
|
)
|
|
return failures
|
|
|
|
@staticmethod
|
|
def _claim_repair_findings(
|
|
deterministic: Sequence[QualityFinding],
|
|
judged: Sequence[QualityFinding],
|
|
draft: ResumeDraft,
|
|
) -> list[QualityFinding]:
|
|
known_claims = {
|
|
claim.claim_id for section in draft.sections for claim in section.claims
|
|
}
|
|
selected: list[QualityFinding] = []
|
|
seen: set[tuple[str, str | None, str]] = set()
|
|
for finding in [*deterministic, *judged]:
|
|
if (
|
|
finding.claim_id not in known_claims
|
|
or finding.severity is QualitySeverity.INFO
|
|
):
|
|
continue
|
|
fingerprint = (finding.code, finding.claim_id, finding.message)
|
|
if fingerprint in seen:
|
|
continue
|
|
seen.add(fingerprint)
|
|
selected.append(finding)
|
|
return selected
|
|
|
|
@staticmethod
|
|
def _assert_repair_scope(
|
|
*,
|
|
previous: ResumeDraft,
|
|
repaired: ResumeDraft,
|
|
targeted_claim_ids: set[str],
|
|
evidence_ceiling: set[str],
|
|
) -> None:
|
|
if (
|
|
previous.candidate_id != repaired.candidate_id
|
|
or previous.posting_id != repaired.posting_id
|
|
or previous.mode is not repaired.mode
|
|
or previous.title != repaired.title
|
|
):
|
|
raise StageIntegrityError(
|
|
"repair-resume: claim repair changed draft-level identity or metadata"
|
|
)
|
|
previous_sections = {section.section_id: section for section in previous.sections}
|
|
repaired_sections = {section.section_id: section for section in repaired.sections}
|
|
if set(previous_sections) != set(repaired_sections):
|
|
raise StageIntegrityError("repair-resume: claim repair changed section set")
|
|
|
|
for section_id, old_section in previous_sections.items():
|
|
new_section = repaired_sections[section_id]
|
|
if (
|
|
old_section.section_type is not new_section.section_type
|
|
or old_section.heading != new_section.heading
|
|
or old_section.order != new_section.order
|
|
):
|
|
raise StageIntegrityError(
|
|
"repair-resume: claim repair changed section metadata"
|
|
)
|
|
old_claims = {claim.claim_id: claim for claim in old_section.claims}
|
|
new_claims = {claim.claim_id: claim for claim in new_section.claims}
|
|
if set(old_claims) != set(new_claims):
|
|
raise StageIntegrityError(
|
|
"repair-resume: claim repair changed claim set"
|
|
)
|
|
for claim_id, old_claim in old_claims.items():
|
|
new_claim = new_claims[claim_id]
|
|
if old_claim.order != new_claim.order:
|
|
raise StageIntegrityError(
|
|
"repair-resume: claim repair changed claim order"
|
|
)
|
|
if claim_id not in targeted_claim_ids and old_claim != new_claim:
|
|
raise StageIntegrityError(
|
|
"repair-resume: non-targeted claim was modified"
|
|
)
|
|
if not set(new_claim.evidence_ids) <= evidence_ceiling:
|
|
raise StageIntegrityError(
|
|
"repair-resume: repair introduced evidence outside the "
|
|
"original draft"
|
|
)
|
|
|
|
@staticmethod
|
|
def _questions(
|
|
repair_findings: Sequence[QualityFinding],
|
|
gate_failures: Sequence[str],
|
|
report: QualityReport,
|
|
) -> list[str]:
|
|
questions: list[str] = []
|
|
seen_claims: set[str] = set()
|
|
for finding in repair_findings:
|
|
claim_id = finding.claim_id
|
|
if claim_id is None or claim_id in seen_claims:
|
|
continue
|
|
seen_claims.add(claim_id)
|
|
questions.append(
|
|
f"{claim_id} 주장을 보완할 수 있는 검증 가능한 사실이나 "
|
|
"정확한 수치를 제공해 주시겠습니까?"
|
|
)
|
|
if len(questions) == 3:
|
|
return questions
|
|
|
|
if report.evidence_coverage < MINIMUM_EVIDENCE_COVERAGE:
|
|
questions.append(
|
|
"근거가 연결되지 않은 주장을 뒷받침하거나 삭제할 수 있도록 "
|
|
"추가 사실을 제공해 주시겠습니까?"
|
|
)
|
|
if len(questions) < 3 and any(
|
|
failure.startswith("requirement_coverage:") for failure in gate_failures
|
|
):
|
|
questions.append(
|
|
"아직 다루지 못한 직무 요건과 관련된 실제 경험이나 산출물이 "
|
|
"있다면 제공해 주시겠습니까?"
|
|
)
|
|
if len(questions) < 3:
|
|
deficient_categories = [
|
|
category.value
|
|
for category in sorted(
|
|
MAJOR_QUALITY_CATEGORIES, key=lambda item: item.value
|
|
)
|
|
if report.category_scores.get(category, -1)
|
|
< MINIMUM_MAJOR_CATEGORY_SCORE
|
|
]
|
|
if deficient_categories:
|
|
questions.append(
|
|
"주요 품질 영역("
|
|
+ ", ".join(deficient_categories)
|
|
+ ")을 보완할 추가 근거를 제공해 주시겠습니까?"
|
|
)
|
|
if not questions:
|
|
questions.append(
|
|
"품질 기준을 충족하려면 어떤 주장을 유지해야 하는지와 이를 "
|
|
"뒷받침할 추가 근거를 확인해 주시겠습니까?"
|
|
)
|
|
# Stable de-duplication and a hard API bound.
|
|
return list(dict.fromkeys(questions))[:3]
|
|
|
|
|
|
def run_pipeline(
|
|
backend: LLMBackend,
|
|
profile: CandidateProfile | Mapping[str, Any],
|
|
posting: JobPosting | Mapping[str, Any],
|
|
config: GenerationConfig | Mapping[str, Any],
|
|
*,
|
|
prompt_dir: str | Path | None = None,
|
|
max_repair_attempts: Literal[0, 1, 2] = 2,
|
|
) -> PipelineResult:
|
|
"""One-call convenience wrapper around :class:`ResumePipeline`."""
|
|
|
|
return ResumePipeline(
|
|
backend,
|
|
prompt_dir=prompt_dir,
|
|
max_repair_attempts=max_repair_attempts,
|
|
).run(profile, posting, config)
|
|
|
|
|
|
__all__ = [
|
|
"BackendCallError",
|
|
"LLMBackend",
|
|
"MAJOR_QUALITY_CATEGORIES",
|
|
"PipelineError",
|
|
"PipelineResult",
|
|
"PipelineStatus",
|
|
"PromptLoadError",
|
|
"ResumePipeline",
|
|
"StageIntegrityError",
|
|
"run_pipeline",
|
|
]
|