generated from kgod/ai-review-template
426 lines
16 KiB
Python
426 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from app.claim_validator import validate_proposal
|
|
from app.experience_optimizer import (
|
|
FallbackExperienceOptimizer,
|
|
OpenAIExperienceOptimizer,
|
|
RuleStructuredExperienceOptimizer,
|
|
_OPTIMIZATION_REPAIR_PROMPT,
|
|
normalize_fact_ledger,
|
|
required_material_fact_ids,
|
|
)
|
|
|
|
|
|
class FakeCompletion:
|
|
def __init__(self, output: dict[str, Any]) -> None:
|
|
self.output = output
|
|
self.calls: list[dict[str, Any]] = []
|
|
self.prompts: list[str] = []
|
|
|
|
def complete(self, *, schema, schema_name, system_prompt, payload):
|
|
self.calls.append({"schema_name": schema_name, "payload": payload})
|
|
self.prompts.append(system_prompt)
|
|
return schema.model_validate(self.output)
|
|
|
|
|
|
class SequentialFakeCompletion:
|
|
def __init__(self, outputs: list[dict[str, Any]]) -> None:
|
|
self.outputs = outputs
|
|
self.calls: list[dict[str, Any]] = []
|
|
self.prompts: list[str] = []
|
|
|
|
def complete(self, *, schema, schema_name, system_prompt, payload):
|
|
self.calls.append({"schema_name": schema_name, "payload": payload})
|
|
self.prompts.append(system_prompt)
|
|
output_index = min(len(self.calls) - 1, len(self.outputs) - 1)
|
|
return schema.model_validate(self.outputs[output_index])
|
|
|
|
|
|
class FakeRetriever:
|
|
def retrieve(self, **kwargs):
|
|
return [
|
|
{
|
|
"id": "rag_1",
|
|
"title": "Backend reference",
|
|
"content": "A reference example reports an 80% throughput gain.",
|
|
"original": "Built a service.",
|
|
"optimized": "Improved throughput by 80%.",
|
|
"points": "Action and result",
|
|
}
|
|
]
|
|
|
|
|
|
class FakeEmbedder:
|
|
pass
|
|
|
|
|
|
def fact_ledger() -> list[dict[str, str]]:
|
|
return [
|
|
{
|
|
"id": "fact_1",
|
|
"source": "user_form",
|
|
"field": "description",
|
|
"text": "Built the backend for a course submission system.",
|
|
},
|
|
{
|
|
"id": "fact_2",
|
|
"source": "user_answer",
|
|
"field": "answer",
|
|
"text": "It supported 20 classmates submitting assignments.",
|
|
},
|
|
{
|
|
"id": "fact_3",
|
|
"source": "user_answer",
|
|
"field": "answer",
|
|
"text": "Used FastAPI and PostgreSQL.",
|
|
},
|
|
]
|
|
|
|
|
|
def valid_output() -> dict[str, Any]:
|
|
text = (
|
|
"Used FastAPI and PostgreSQL to build the course submission backend, "
|
|
"supporting 20 classmates submitting assignments."
|
|
)
|
|
return {
|
|
"optimized_description": text,
|
|
"bullets": [text],
|
|
"star": {
|
|
"situation": "Course submission scenario",
|
|
"task": "Backend development",
|
|
"action": "Implemented the API with FastAPI and PostgreSQL",
|
|
"result": "Supported 20 classmates",
|
|
},
|
|
"changes": ["Reorganized the action and result"],
|
|
"missing_facts": [],
|
|
"claims": [
|
|
{
|
|
"text": text,
|
|
"evidence_ids": ["fact_1", "fact_2", "fact_3"],
|
|
"claim_type": "action",
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def test_openai_optimizer_keeps_rag_as_style_reference_only() -> None:
|
|
completion = FakeCompletion(valid_output())
|
|
optimizer = OpenAIExperienceOptimizer(completion, FakeRetriever(), FakeEmbedder())
|
|
|
|
proposal = optimizer.optimize(
|
|
{"description": "Built the backend for a course submission system."},
|
|
context={"target_position": "Backend Engineer", "entry_type": "project_experience"},
|
|
facts=fact_ledger(),
|
|
)
|
|
|
|
payload = completion.calls[0]["payload"]
|
|
assert payload["user_fact_ledger"] == fact_ledger()
|
|
assert payload["style_references"][0]["id"] == "rag_1"
|
|
assert "80%" in str(payload["style_references"])
|
|
assert all("80%" not in fact["text"] for fact in payload["user_fact_ledger"])
|
|
assert "用户提供的经历" in completion.prompts[0]
|
|
assert proposal["source"] == "ai_expanded"
|
|
|
|
|
|
def test_claim_validator_filters_rag_claim_without_rejecting_grounded_text() -> None:
|
|
proposal = valid_output()
|
|
proposal["claims"].append(
|
|
{
|
|
"text": "Improved throughput by 80%.",
|
|
"evidence_ids": ["rag_1"],
|
|
"claim_type": "result",
|
|
}
|
|
)
|
|
|
|
validated = validate_proposal(proposal, fact_ledger())
|
|
|
|
assert validated["optimized_description"] == valid_output()["optimized_description"]
|
|
assert len(validated["claims"]) == 1
|
|
assert "unsupported_evidence_reference" in validated["validation_warnings"]
|
|
|
|
|
|
def test_unconfirmed_metric_is_retained_as_model_written_resume_prose() -> None:
|
|
proposal = valid_output()
|
|
proposal["optimized_description"] = (
|
|
"Used FastAPI and PostgreSQL to build the course submission backend. "
|
|
"Improved submission efficiency by 80%."
|
|
)
|
|
proposal["bullets"] = [proposal["optimized_description"]]
|
|
|
|
validated = validate_proposal(proposal, fact_ledger())
|
|
|
|
assert "80%" in validated["optimized_description"]
|
|
assert not validated["unconfirmed_suggestions"]
|
|
|
|
|
|
def test_counted_object_expansion_is_retained_not_a_hard_failure() -> None:
|
|
proposal = valid_output()
|
|
proposal["optimized_description"] = "Delivered 20 features for the course platform."
|
|
proposal["bullets"] = [proposal["optimized_description"]]
|
|
|
|
validated = validate_proposal(proposal, fact_ledger())
|
|
|
|
assert validated["optimized_description"] == "Delivered 20 features for the course platform."
|
|
assert not validated["unconfirmed_suggestions"]
|
|
|
|
|
|
def test_new_technical_term_is_retained_for_controlled_role_expansion() -> None:
|
|
proposal = valid_output()
|
|
proposal["optimized_description"] = "Built the backend with FastAPI, PostgreSQL, and Redis."
|
|
proposal["bullets"] = [proposal["optimized_description"]]
|
|
|
|
validated = validate_proposal(proposal, fact_ledger())
|
|
|
|
assert "Redis" in validated["optimized_description"]
|
|
|
|
|
|
def test_omitted_material_fact_triggers_single_repair() -> None:
|
|
initial = valid_output()
|
|
initial["optimized_description"] = "Built the course submission backend with FastAPI and PostgreSQL."
|
|
initial["bullets"] = [initial["optimized_description"]]
|
|
completion = SequentialFakeCompletion([initial, valid_output()])
|
|
optimizer = OpenAIExperienceOptimizer(completion, FakeRetriever(), FakeEmbedder())
|
|
|
|
proposal = optimizer.optimize(
|
|
{"description": "Built the backend for a course submission system."},
|
|
context={"entry_type": "project_experience"},
|
|
facts=fact_ledger(),
|
|
)
|
|
|
|
assert len(completion.calls) == 2
|
|
assert completion.calls[1]["schema_name"] == "experience_optimization_repair"
|
|
assert "20 classmates" in proposal["optimized_description"]
|
|
assert proposal["omitted_fact_ids"] == []
|
|
|
|
|
|
def test_omitted_material_fact_after_repair_is_flagged_and_relaxed() -> None:
|
|
incomplete = valid_output()
|
|
incomplete["optimized_description"] = "Built the course submission backend with FastAPI and PostgreSQL."
|
|
incomplete["bullets"] = [incomplete["optimized_description"]]
|
|
completion = SequentialFakeCompletion([incomplete, incomplete])
|
|
optimizer = OpenAIExperienceOptimizer(completion, FakeRetriever(), FakeEmbedder())
|
|
|
|
proposal = optimizer.optimize(
|
|
{"description": "Built the backend for a course submission system."},
|
|
context={"entry_type": "project_experience"},
|
|
facts=fact_ledger(),
|
|
)
|
|
|
|
assert len(completion.calls) == 2
|
|
assert proposal["omitted_fact_ids"] == ["fact_2"]
|
|
assert "material_fact_omitted_after_repair" in proposal.get("validation_warnings", [])
|
|
|
|
|
|
def test_imported_long_description_triggers_repair_when_candidate_is_compressed() -> None:
|
|
imported_facts = [
|
|
{
|
|
"id": "imported_description",
|
|
"source": "user_form",
|
|
"field": "description",
|
|
"text": (
|
|
"搭建全栈求职平台，支持 WebSocket 流式预览；使用 sentence-transformers "
|
|
"与 pgvector 实现岗位语义检索；通过 ASGI 部署和 asyncpg 缓解并发瓶颈；"
|
|
"交付 58 个 API 接口，使用 Docker Compose 编排 7 个服务，并支持 PDF/DOCX 导出。"
|
|
),
|
|
}
|
|
]
|
|
compressed = valid_output()
|
|
compressed["optimized_description"] = "构建全栈求职平台，集成简历生成、JD 分析和模拟面试功能。"
|
|
compressed["bullets"] = [compressed["optimized_description"]]
|
|
completion = SequentialFakeCompletion([compressed, compressed])
|
|
optimizer = OpenAIExperienceOptimizer(completion, FakeRetriever(), FakeEmbedder())
|
|
|
|
proposal = optimizer.optimize(
|
|
{"description": imported_facts[0]["text"]},
|
|
context={"entry_type": "project_experience"},
|
|
facts=imported_facts,
|
|
)
|
|
|
|
assert len(completion.calls) == 2
|
|
assert "material_fact_omitted_after_repair" in proposal.get("validation_warnings", [])
|
|
assert proposal["omitted_fact_ids"] == [
|
|
"imported_description_part_1",
|
|
"imported_description_part_2",
|
|
"imported_description_part_3",
|
|
"imported_description_part_4",
|
|
]
|
|
|
|
def test_optimizer_passes_completed_deep_interview_context_to_model() -> None:
|
|
completion = FakeCompletion(valid_output())
|
|
optimizer = OpenAIExperienceOptimizer(completion, FakeRetriever(), FakeEmbedder())
|
|
history = [{"question_id": "q_1", "dimension": "personal_contribution", "answer": "Implemented API endpoints.", "status": "answered"}]
|
|
|
|
optimizer.optimize(
|
|
{"description": "Built the backend for a course submission system."},
|
|
context={
|
|
"entry_type": "project_experience",
|
|
"optimization_mode": "deep",
|
|
"interview_completion": {"is_sufficient": True, "blocking_gaps": []},
|
|
"completed_dimensions": ["personal_contribution"],
|
|
"question_history": history,
|
|
},
|
|
facts=fact_ledger(),
|
|
)
|
|
|
|
payload = completion.calls[0]["payload"]
|
|
assert payload["deep_interview"]["completion"]["is_sufficient"] is True
|
|
assert payload["deep_interview"]["completed_dimensions"] == ["personal_contribution"]
|
|
assert payload["deep_interview"]["question_history"] == history
|
|
|
|
|
|
def test_optimizer_keeps_model_prose_when_it_expands_beyond_literal_evidence() -> None:
|
|
output = valid_output()
|
|
output["optimized_description"] = (
|
|
"Built the backend for a course submission system. "
|
|
"It supported 20 classmates submitting assignments. "
|
|
"Used FastAPI and PostgreSQL. "
|
|
"Migrated 20 services to Redis and improved throughput by 80%."
|
|
)
|
|
output["bullets"] = [output["optimized_description"]]
|
|
completion = FakeCompletion(output)
|
|
optimizer = OpenAIExperienceOptimizer(completion, FakeRetriever(), FakeEmbedder())
|
|
|
|
proposal = optimizer.optimize(
|
|
{"description": "Built the backend for a course submission system."},
|
|
context={"entry_type": "project_experience"},
|
|
facts=fact_ledger(),
|
|
)
|
|
|
|
assert "Migrated 20 services" in proposal["optimized_description"]
|
|
assert not proposal["unconfirmed_suggestions"]
|
|
assert len(completion.calls) == 1
|
|
|
|
|
|
def test_optimizer_succeeds_when_retriever_has_no_documents() -> None:
|
|
class EmptyRetriever:
|
|
def retrieve(self, **kwargs):
|
|
return []
|
|
|
|
completion = FakeCompletion(valid_output())
|
|
optimizer = OpenAIExperienceOptimizer(completion, EmptyRetriever(), FakeEmbedder())
|
|
|
|
proposal = optimizer.optimize(
|
|
{"description": "Built the backend for a course submission system."},
|
|
context={"entry_type": "project_experience"},
|
|
facts=fact_ledger(),
|
|
)
|
|
|
|
assert proposal["optimized_description"]
|
|
assert completion.calls[0]["payload"]["style_references"] == []
|
|
|
|
|
|
def test_optimizer_succeeds_when_retriever_is_unavailable() -> None:
|
|
class BrokenRetriever:
|
|
def retrieve(self, **kwargs):
|
|
raise RuntimeError("vector store unavailable")
|
|
|
|
completion = FakeCompletion(valid_output())
|
|
optimizer = OpenAIExperienceOptimizer(completion, BrokenRetriever(), FakeEmbedder())
|
|
|
|
proposal = optimizer.optimize(
|
|
{"description": "Built the backend for a course submission system."},
|
|
context={"entry_type": "project_experience"},
|
|
facts=fact_ledger(),
|
|
)
|
|
|
|
assert proposal["optimized_description"]
|
|
assert completion.calls[0]["payload"]["style_references"] == []
|
|
|
|
|
|
def test_rule_optimizer_reports_insufficient_facts_without_creating_content() -> None:
|
|
proposal = RuleStructuredExperienceOptimizer().optimize({}, context={}, facts=[])
|
|
|
|
assert proposal["optimized_description"] == ""
|
|
assert proposal["fallback_reason"] == "insufficient_user_facts"
|
|
|
|
|
|
def test_fallback_optimizer_marks_rule_source_when_model_fails() -> None:
|
|
class BrokenOptimizer:
|
|
def optimize(self, entry, *, context, facts):
|
|
raise RuntimeError("model unavailable")
|
|
|
|
optimizer = FallbackExperienceOptimizer(
|
|
BrokenOptimizer(), RuleStructuredExperienceOptimizer()
|
|
)
|
|
proposal = optimizer.optimize(
|
|
{"description": "Built an API."}, context={}, facts=fact_ledger()[:1]
|
|
)
|
|
|
|
assert proposal["source"] == "rule_structured"
|
|
assert proposal["optimized_description"]
|
|
|
|
def test_claim_validator_decodes_literal_unicode_escapes_in_suggestions() -> None:
|
|
proposal = valid_output()
|
|
proposal["unconfirmed_suggestions"] = [r"\u8FD8\u53EF\u8865\u5145\u7ED3\u679C"]
|
|
|
|
validated = validate_proposal(proposal, fact_ledger())
|
|
|
|
assert validated["unconfirmed_suggestions"] == ["还可补充结果"]
|
|
|
|
def test_normalize_fact_ledger_splits_multiline_description_into_part_facts() -> None:
|
|
facts = [
|
|
{
|
|
"id": "fact_1",
|
|
"source": "user_form",
|
|
"field": "description",
|
|
"text": (
|
|
"全栈 AI 求职助手平台，包含 5 大功能模块：\n"
|
|
"1. AI 对话式简历生成助手\n"
|
|
"2. 简历导入 (PDF/DOCX 智能解析)\n"
|
|
"技术栈: 前端 Next.js 14.2 + React 18.3"
|
|
),
|
|
}
|
|
]
|
|
ledger = normalize_fact_ledger(facts)
|
|
|
|
parts = [fact for fact in ledger if fact["field"] == "description_part"]
|
|
assert [part["id"] for part in parts] == [
|
|
"fact_1_part_1",
|
|
"fact_1_part_2",
|
|
"fact_1_part_3",
|
|
"fact_1_part_4",
|
|
]
|
|
assert parts[1]["text"] == "AI 对话式简历生成助手"
|
|
assert parts[3]["text"] == "前端 Next.js 14.2 + React 18.3"
|
|
assert any(fact["field"] == "description" for fact in ledger)
|
|
|
|
|
|
def test_normalize_fact_ledger_keeps_single_sentence_description_unsplit() -> None:
|
|
facts = [
|
|
{
|
|
"id": "fact_1",
|
|
"source": "user_form",
|
|
"field": "description",
|
|
"text": "Built the backend for a course submission system.",
|
|
}
|
|
]
|
|
ledger = normalize_fact_ledger(facts)
|
|
|
|
assert [fact["id"] for fact in ledger] == ["fact_1"]
|
|
|
|
|
|
def test_required_fact_ids_prefer_description_parts_over_parent() -> None:
|
|
facts = [
|
|
{
|
|
"id": "fact_1",
|
|
"source": "user_form",
|
|
"field": "description",
|
|
"text": "全栈 AI 求职助手平台，包含 5 大功能模块：\n1. AI 对话式简历生成助手\n2. 简历导入智能解析",
|
|
},
|
|
{"id": "fact_2", "source": "user_answer", "field": "answer", "text": "服务 300 名学生。"},
|
|
]
|
|
ledger = normalize_fact_ledger(facts)
|
|
|
|
required = required_material_fact_ids(ledger)
|
|
|
|
assert "fact_1" not in required
|
|
assert {"fact_1_part_1", "fact_1_part_2", "fact_1_part_3", "fact_2"} <= set(required)
|
|
|
|
|
|
|
|
def test_optimization_repair_prompt_keeps_star_structure() -> None:
|
|
"""修复稿与首发同构：STAR 结构要求不得在修复阶段丢失。"""
|
|
assert "STAR" in _OPTIMIZATION_REPAIR_PROMPT
|