generated from kgod/ai-review-template
439 lines
15 KiB
Python
439 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import logging
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
from app.llm_services import (
|
|
AnchorExtractionOutput,
|
|
LLMServiceError,
|
|
OpenAICompatibleStructuredClient,
|
|
OpenAIExperienceExtractor,
|
|
OpenAIResumeRewriter,
|
|
_diagnostic_logger,
|
|
log_ai_event,
|
|
)
|
|
from app.experience_optimizer import OpenAIExperienceOptimizer
|
|
from app.main import create_app
|
|
from app.settings import Settings, load_settings
|
|
from app.skill_suggester import OpenAISkillSuggester, RuleBasedSkillSuggester
|
|
|
|
|
|
class FakeCompletions:
|
|
def __init__(self, responses: list[str | Exception]) -> None:
|
|
self.responses = list(responses)
|
|
self.calls: list[dict[str, Any]] = []
|
|
|
|
def create(self, **kwargs: Any) -> Any:
|
|
self.calls.append(kwargs)
|
|
response = self.responses.pop(0)
|
|
if isinstance(response, Exception):
|
|
raise response
|
|
message = SimpleNamespace(content=response, parsed=None, refusal=None)
|
|
return SimpleNamespace(choices=[SimpleNamespace(message=message)])
|
|
|
|
|
|
class FakeOpenAI:
|
|
def __init__(self, responses: list[str | Exception]) -> None:
|
|
self.completions = FakeCompletions(responses)
|
|
self.chat = SimpleNamespace(completions=self.completions)
|
|
|
|
|
|
def llm_settings(**overrides: Any) -> Settings:
|
|
values: dict[str, Any] = {
|
|
"llm_provider": "openai",
|
|
"openai_api_key": "test-key-not-a-secret",
|
|
"openai_base_url": "https://example.test/v1",
|
|
"openai_model": "test-model",
|
|
"openai_timeout_seconds": 12.0,
|
|
"openai_max_retries": 2,
|
|
"structured_output_retries": 1,
|
|
"structured_output_mode": "json_schema",
|
|
"fallback_to_rules": False,
|
|
}
|
|
values.update(overrides)
|
|
return Settings(**values)
|
|
|
|
|
|
def anchor_response() -> str:
|
|
return json.dumps(
|
|
{
|
|
"record_type": "work_experience",
|
|
"field_updates": {
|
|
"school": None,
|
|
"major": None,
|
|
"degree": None,
|
|
"company": "星河科技有限公司",
|
|
"position": "产品经理",
|
|
"project_name": None,
|
|
"project_role": None,
|
|
"start_date": "2022-03",
|
|
"end_date_or_present": "present",
|
|
},
|
|
"evidence_spans": [
|
|
{"field": "company", "quote": "星河科技有限公司"},
|
|
{"field": "position", "quote": "产品经理"},
|
|
{"field": "start_date", "quote": "2022年3月"},
|
|
{"field": "end_date_or_present", "quote": "至今"},
|
|
],
|
|
"ambiguities": [],
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
def test_openai_skill_suggester_uses_target_and_record_facts() -> None:
|
|
fake = FakeOpenAI([json.dumps({"skills": ["FastAPI", "Redis", "OpenAPI"]})])
|
|
suggester = OpenAISkillSuggester(
|
|
OpenAICompatibleStructuredClient(llm_settings(), fake),
|
|
RuleBasedSkillSuggester(),
|
|
)
|
|
profile = {
|
|
"target_position": "后端工程师",
|
|
"records": {
|
|
"project_experience": [
|
|
{
|
|
"record_type": "project_experience",
|
|
"description": "使用 Python 和 FastAPI 开发订单接口，接入 Redis 缓存。",
|
|
}
|
|
]
|
|
},
|
|
"tags": {"skills": ["Python"]},
|
|
}
|
|
|
|
suggestions = suggester.suggest(profile)
|
|
|
|
assert suggestions[:3] == ["FastAPI", "Redis", "OpenAPI"]
|
|
assert "Python" not in suggestions
|
|
request = fake.completions.calls[0]
|
|
assert request["model"] == "test-model"
|
|
request_text = str(request["messages"])
|
|
assert "后端工程师" in request_text
|
|
assert "订单接口" in request_text
|
|
|
|
|
|
def test_openai_skill_suggester_falls_back_to_rules() -> None:
|
|
fake = FakeOpenAI([RuntimeError("offline")])
|
|
profile = {"target_position": "后端工程师", "tags": {"skills": []}}
|
|
suggester = OpenAISkillSuggester(
|
|
OpenAICompatibleStructuredClient(llm_settings(), fake),
|
|
RuleBasedSkillSuggester(),
|
|
)
|
|
|
|
assert "MySQL" in suggester.suggest(profile)
|
|
|
|
|
|
def test_anchor_extraction_retries_validates_and_redacts_phone() -> None:
|
|
fake = FakeOpenAI(["not-json", anchor_response()])
|
|
completion = OpenAICompatibleStructuredClient(llm_settings(), fake)
|
|
extractor = OpenAIExperienceExtractor(completion)
|
|
|
|
patch = extractor.extract_anchor(
|
|
"我从2022年3月至今在星河科技有限公司担任产品经理，电话13800138000",
|
|
"work_experience",
|
|
["company", "position", "start_date", "end_date_or_present"],
|
|
)
|
|
|
|
assert patch == {
|
|
"company": "星河科技有限公司",
|
|
"position": "产品经理",
|
|
"start_date": "2022-03",
|
|
"end_date_or_present": "present",
|
|
}
|
|
assert len(fake.completions.calls) == 2
|
|
call = fake.completions.calls[-1]
|
|
assert call["model"] == "test-model"
|
|
assert call["timeout"] == 12.0
|
|
assert call["response_format"]["type"] == "json_schema"
|
|
serialized_messages = json.dumps(call["messages"], ensure_ascii=False)
|
|
assert "13800138000" not in serialized_messages
|
|
assert "[手机号已脱敏]" in serialized_messages
|
|
|
|
|
|
def test_experience_extraction_uses_pydantic_and_exact_evidence() -> None:
|
|
response = json.dumps(
|
|
{
|
|
"title": "后端工程师",
|
|
"organization": "星河科技",
|
|
"role": "后端工程师",
|
|
"highlights": ["优化接口耗时，降低30%"],
|
|
"metrics": ["30%", "99%"],
|
|
"confidence": 0.93,
|
|
"evidence_spans": [
|
|
{"field": "organization", "quote": "星河科技"},
|
|
{"field": "role", "quote": "后端工程师"},
|
|
{"field": "highlights", "quote": "优化接口耗时，降低30%"},
|
|
],
|
|
"ambiguities": [],
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
extractor = OpenAIExperienceExtractor(
|
|
OpenAICompatibleStructuredClient(llm_settings(), FakeOpenAI([response]))
|
|
)
|
|
|
|
result = extractor.extract("在星河科技担任后端工程师，优化接口耗时，降低30%")
|
|
|
|
assert result.organization == "星河科技"
|
|
assert result.highlights == ["优化接口耗时，降低30%"]
|
|
assert result.metrics == ["30%"]
|
|
assert result.confidence == 0.95
|
|
|
|
|
|
def test_sdk_boundary_redacts_email_wechat_and_split_phone() -> None:
|
|
fake = FakeOpenAI([anchor_response()])
|
|
completion = OpenAICompatibleStructuredClient(llm_settings(), fake)
|
|
completion.complete(
|
|
schema=AnchorExtractionOutput,
|
|
schema_name="resume_anchor_extraction",
|
|
system_prompt="extract",
|
|
payload={
|
|
"user_text": (
|
|
"手机 138-0013-8000，邮箱 user@example.com，微信号: resume_helper"
|
|
)
|
|
},
|
|
)
|
|
|
|
request_text = json.dumps(fake.completions.calls[0]["messages"], ensure_ascii=False)
|
|
assert "138-0013-8000" not in request_text
|
|
assert "user@example.com" not in request_text
|
|
assert "resume_helper" not in request_text
|
|
assert "[手机号已脱敏]" in request_text
|
|
assert "[邮箱已脱敏]" in request_text
|
|
assert "[微信号已脱敏]" in request_text
|
|
|
|
|
|
def test_json_object_mode_includes_the_pydantic_schema() -> None:
|
|
fake = FakeOpenAI([anchor_response()])
|
|
settings = llm_settings(structured_output_mode="json_object")
|
|
completion = OpenAICompatibleStructuredClient(settings, fake)
|
|
|
|
completion.complete(
|
|
schema=AnchorExtractionOutput,
|
|
schema_name="resume_anchor_extraction",
|
|
system_prompt="提取事实。",
|
|
payload={"user_text": "在星河科技担任产品经理"},
|
|
)
|
|
|
|
call = fake.completions.calls[0]
|
|
assert call["response_format"] == {"type": "json_object"}
|
|
assert "output_json_schema" in call["messages"][1]["content"]
|
|
assert "只返回" in call["messages"][0]["content"]
|
|
|
|
|
|
def test_rewriter_preserves_confirmed_descriptions_without_second_model_call() -> None:
|
|
fake = FakeOpenAI([])
|
|
rewriter = OpenAIResumeRewriter(
|
|
OpenAICompatibleStructuredClient(llm_settings(), fake)
|
|
)
|
|
profile = {
|
|
"name": "张三",
|
|
"phone": "13800138000",
|
|
"phone_source": "manual",
|
|
"job_type": "social",
|
|
"anchor_type": "work_experience",
|
|
"anchor": {
|
|
"company": "星河科技",
|
|
"position": "后端工程师",
|
|
"start_date": "2022-01",
|
|
"end_date_or_present": "present",
|
|
},
|
|
"experiences": [
|
|
{
|
|
"title": "后端工程师",
|
|
"organization": "星河科技",
|
|
"role": "后端工程师",
|
|
"description": "优化接口耗时，降低30%。",
|
|
}
|
|
],
|
|
}
|
|
|
|
resume = rewriter.rewrite(profile)
|
|
|
|
assert resume["basics"]["masked_phone"] == "138****8000"
|
|
item = resume["sections"][1]["items"][0]
|
|
assert item["description"] == "优化接口耗时，降低30%。"
|
|
assert "resume_bullets" not in item
|
|
assert fake.completions.calls == []
|
|
|
|
def test_settings_load_dotenv_and_create_app_wires_openai_defaults(
|
|
tmp_path, monkeypatch
|
|
) -> None:
|
|
env_file = tmp_path / ".env"
|
|
env_file.write_text(
|
|
"\n".join(
|
|
[
|
|
"RESUME_AGENT_LLM_PROVIDER=openai",
|
|
"OPENAI_API_KEY=dummy-key",
|
|
"OPENAI_BASE_URL=https://gateway.test",
|
|
"OPENAI_MODEL=test-model",
|
|
"RESUME_AGENT_LLM_FALLBACK_TO_RULES=false",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
for name in (
|
|
"RESUME_AGENT_LLM_PROVIDER",
|
|
"OPENAI_API_KEY",
|
|
"OPENAI_BASE_URL",
|
|
"OPENAI_MODEL",
|
|
"RESUME_AGENT_LLM_FALLBACK_TO_RULES",
|
|
):
|
|
monkeypatch.delenv(name, raising=False)
|
|
settings = load_settings(env_file)
|
|
fake = FakeOpenAI([anchor_response()])
|
|
|
|
application = create_app(
|
|
database_path=tmp_path / "llm.db",
|
|
settings=settings,
|
|
openai_client=fake,
|
|
)
|
|
|
|
assert isinstance(application.state.resume_agent.extractor, OpenAIExperienceExtractor)
|
|
assert isinstance(
|
|
application.state.resume_agent.experience_optimizer,
|
|
OpenAIExperienceOptimizer,
|
|
)
|
|
assert settings.openai_base_url == "https://gateway.test"
|
|
assert "dummy-key" not in repr(settings)
|
|
|
|
def test_settings_loads_volcengine_ark_configuration(tmp_path, monkeypatch) -> None:
|
|
env_file = tmp_path / ".env"
|
|
env_file.write_text(
|
|
"\n".join(
|
|
[
|
|
"RESUME_AGENT_LLM_PROVIDER=volcengine",
|
|
"VOLCENGINE_API_KEY=test-volcengine-key",
|
|
"VOLCENGINE_BASE_URL=https://ark.example.test/api/v3",
|
|
"VOLCENGINE_MODEL=ep-test-endpoint",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
for name in (
|
|
"RESUME_AGENT_LLM_PROVIDER",
|
|
"OPENAI_API_KEY",
|
|
"OPENAI_BASE_URL",
|
|
"OPENAI_MODEL",
|
|
"VOLCENGINE_API_KEY",
|
|
"VOLCENGINE_BASE_URL",
|
|
"VOLCENGINE_MODEL",
|
|
):
|
|
monkeypatch.delenv(name, raising=False)
|
|
|
|
settings = load_settings(env_file)
|
|
|
|
assert settings.llm_provider == "volcengine"
|
|
assert settings.use_openai is True
|
|
assert settings.openai_base_url == "https://ark.example.test/api/v3"
|
|
assert settings.openai_model == "ep-test-endpoint"
|
|
assert "test-volcengine-key" not in repr(settings)
|
|
|
|
|
|
def test_volcengine_requires_an_inference_endpoint_id(tmp_path, monkeypatch) -> None:
|
|
env_file = tmp_path / ".env"
|
|
env_file.write_text(
|
|
"\n".join(
|
|
[
|
|
"RESUME_AGENT_LLM_PROVIDER=volcengine",
|
|
"VOLCENGINE_API_KEY=test-volcengine-key",
|
|
"VOLCENGINE_BASE_URL=https://ark.example.test/api/v3",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
for name in (
|
|
"RESUME_AGENT_LLM_PROVIDER",
|
|
"VOLCENGINE_API_KEY",
|
|
"VOLCENGINE_BASE_URL",
|
|
"VOLCENGINE_MODEL",
|
|
):
|
|
monkeypatch.delenv(name, raising=False)
|
|
|
|
try:
|
|
load_settings(env_file)
|
|
except ValueError as exc:
|
|
assert str(exc) == "VOLCENGINE_MODEL cannot be blank"
|
|
else:
|
|
raise AssertionError("Expected a missing Volcengine model configuration error")
|
|
|
|
def test_invalid_structured_output_reports_precise_reason() -> None:
|
|
client = OpenAICompatibleStructuredClient(
|
|
llm_settings(structured_output_retries=0), FakeOpenAI(["not-json"])
|
|
)
|
|
|
|
try:
|
|
client.complete(
|
|
schema=AnchorExtractionOutput,
|
|
schema_name="invalid_output_test",
|
|
system_prompt="Return structured data.",
|
|
payload={"value": "safe"},
|
|
)
|
|
except LLMServiceError as exc:
|
|
assert exc.reason_code == "structured_output_invalid"
|
|
assert exc.trace_id and exc.trace_id.startswith("ai_")
|
|
else:
|
|
raise AssertionError("LLMServiceError was not raised")
|
|
|
|
|
|
def test_timeout_reports_gateway_timeout() -> None:
|
|
client = OpenAICompatibleStructuredClient(
|
|
llm_settings(structured_output_retries=0), FakeOpenAI([TimeoutError()])
|
|
)
|
|
|
|
try:
|
|
client.complete(
|
|
schema=AnchorExtractionOutput,
|
|
schema_name="timeout_test",
|
|
system_prompt="Return structured data.",
|
|
payload={"value": "safe"},
|
|
)
|
|
except LLMServiceError as exc:
|
|
assert exc.reason_code == "gateway_timeout"
|
|
else:
|
|
raise AssertionError("LLMServiceError was not raised")
|
|
|
|
|
|
def test_empty_model_content_reports_empty_result() -> None:
|
|
client = OpenAICompatibleStructuredClient(
|
|
llm_settings(structured_output_retries=0), FakeOpenAI([""])
|
|
)
|
|
|
|
try:
|
|
client.complete(
|
|
schema=AnchorExtractionOutput,
|
|
schema_name="empty_result_test",
|
|
system_prompt="Return structured data.",
|
|
payload={"value": "safe"},
|
|
)
|
|
except LLMServiceError as exc:
|
|
assert exc.reason_code == "empty_result"
|
|
else:
|
|
raise AssertionError("LLMServiceError was not raised")
|
|
|
|
|
|
def test_diagnostic_log_excludes_sensitive_message_fields() -> None:
|
|
stream = io.StringIO()
|
|
handler = logging.StreamHandler(stream)
|
|
logger = _diagnostic_logger()
|
|
logger.addHandler(handler)
|
|
try:
|
|
log_ai_event(
|
|
"redaction_test",
|
|
trace_id="trace-safe",
|
|
prompt="PROMPT_SECRET",
|
|
payload="PAYLOAD_SECRET",
|
|
response="RESPONSE_SECRET",
|
|
content="CONTENT_SECRET",
|
|
)
|
|
finally:
|
|
logger.removeHandler(handler)
|
|
|
|
output = stream.getvalue()
|
|
assert "redaction_test" in output
|
|
assert "trace-safe" in output
|
|
assert "PROMPT_SECRET" not in output
|
|
assert "PAYLOAD_SECRET" not in output
|
|
assert "RESPONSE_SECRET" not in output
|
|
assert "CONTENT_SECRET" not in output
|