aerospaceboy
ebe33392ca
Preserve document revisions and evidence, harden indexing and export behavior, add explicit material selection and direct-folder import, validate model JSON schemas with safe error messages, and retain student selection while loading scoring data.
261 lines
13 KiB
Python
261 lines
13 KiB
Python
import base64
|
||
import json
|
||
import time
|
||
import unicodedata
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import httpx
|
||
from pydantic import BaseModel, Field, ValidationError, model_validator
|
||
|
||
from .config import get_settings
|
||
|
||
|
||
class EvidenceResult(BaseModel):
|
||
chunk_id: int
|
||
page: int = Field(ge=1)
|
||
quote: str = Field(min_length=1)
|
||
#: Which required-evidence target this quotation is offered to support.
|
||
target_id: str = ""
|
||
|
||
|
||
class ScoreResult(BaseModel):
|
||
criterion_id: str
|
||
#: ``None`` means the model declined to score because admissible evidence is absent.
|
||
score: float | None = Field(default=None, ge=0)
|
||
max_score: float = Field(gt=0)
|
||
reason: str = Field(min_length=1)
|
||
evidence: list[EvidenceResult] = Field(default_factory=list)
|
||
missing_evidence: list[str] = Field(default_factory=list)
|
||
suggestion: str = ""
|
||
confidence: float = Field(ge=0, le=1)
|
||
|
||
@model_validator(mode="after")
|
||
def valid_range(self):
|
||
if self.score is not None and self.score > self.max_score:
|
||
raise ValueError("score exceeds maximum")
|
||
if self.score and not self.evidence:
|
||
raise ValueError("a positive score requires evidence")
|
||
return self
|
||
|
||
|
||
class EvidenceGap(BaseModel):
|
||
"""A content target that still has no body evidence after targeted re-checking."""
|
||
|
||
target_id: str = ""
|
||
target_label: str = ""
|
||
searched_page_types: list[str] = Field(default_factory=list)
|
||
reason: str = ""
|
||
|
||
|
||
class FactResult(BaseModel):
|
||
student_no: str = ""
|
||
name: str = ""
|
||
topic: str = ""
|
||
direction: str = ""
|
||
goals: list[str] = Field(default_factory=list)
|
||
metrics: list[str] = Field(default_factory=list)
|
||
hardware: list[str] = Field(default_factory=list)
|
||
software: list[str] = Field(default_factory=list)
|
||
experiments: list[str] = Field(default_factory=list)
|
||
data: list[str] = Field(default_factory=list)
|
||
schedule: list[str] = Field(default_factory=list)
|
||
cost: list[str] = Field(default_factory=list)
|
||
conclusions: list[str] = Field(default_factory=list)
|
||
|
||
|
||
class ConsistencyIssue(BaseModel):
|
||
description: str
|
||
chunk_ids: list[int] = Field(min_length=2)
|
||
|
||
|
||
class ConsistencyResult(BaseModel):
|
||
issues: list[ConsistencyIssue] = Field(default_factory=list)
|
||
|
||
|
||
class ModelError(RuntimeError):
|
||
pass
|
||
|
||
|
||
def _safe_error_detail(exc):
|
||
"""Describe failures without provider bodies, headers, URLs or exception text."""
|
||
if isinstance(exc, httpx.HTTPStatusError):
|
||
code = exc.response.status_code
|
||
hint = {401: "认证失败，请检查 API Key", 403: "访问被拒绝，请检查模型权限",
|
||
402: "额度不足，请检查账户余额", 429: "请求限流，请稍后重试或降低并发",
|
||
400: "请求参数不被模型支持", 404: "端点或模型不存在"}.get(
|
||
code, "模型服务异常，请稍后重试" if code >= 500 else "请求被拒绝，请检查模型设置")
|
||
return f"HTTP {code}：{hint}"
|
||
if isinstance(exc, httpx.ConnectError):
|
||
return "ConnectError：无法连接模型端点，请检查 Docker 网络、DNS 或代理"
|
||
if isinstance(exc, httpx.TimeoutException):
|
||
return "模型请求超时，请稍后重试或调整超时设置"
|
||
if isinstance(exc, ValidationError):
|
||
fields = sorted({str(error["loc"][0]) for error in exc.errors(include_input=False)
|
||
if error["loc"] and str(error["loc"][0]).isidentifier()})
|
||
return "ValidationError：模型返回字段不符合格式" + (f"（{', '.join(fields)}）" if fields else "")
|
||
return type(exc).__name__
|
||
|
||
|
||
class DeepSeekClient:
|
||
def __init__(self, *, base_url=None, model=None, api_key=None, timeout=None, retries=None, transport=None):
|
||
settings = get_settings()
|
||
self.base_url = (base_url or settings.deepseek_base_url).rstrip("/")
|
||
self.model = model or settings.deepseek_model
|
||
self.api_key = api_key or settings.deepseek_api_key
|
||
self.timeout = timeout or settings.model_timeout_seconds
|
||
self.retries = settings.model_retries if retries is None else retries
|
||
self.transport = transport
|
||
if not self.api_key:
|
||
raise ModelError("未配置DeepSeek API Key")
|
||
|
||
def json_call(self, prompt: str, payload: dict[str, Any], schema: type[BaseModel], *, images: list[str] | None = None):
|
||
content: list[dict] = [{"type": "text", "text": json.dumps(payload, ensure_ascii=False)}]
|
||
for image in images or []:
|
||
path = Path(image)
|
||
if not path.is_file():
|
||
raise ModelError(f"页面图像不存在: {path.name}")
|
||
content.append({"type": "image_url", "image_url": {"url": "data:image/png;base64," + base64.b64encode(path.read_bytes()).decode()}})
|
||
output_schema = json.dumps(schema.model_json_schema(), ensure_ascii=False)
|
||
body = {"model": self.model, "response_format": {"type": "json_object"},
|
||
"messages": [{"role": "system", "content": prompt + " 严格输出一个JSON对象，字段与类型遵循以下JSON Schema：" + output_schema}, {"role": "user", "content": content}],
|
||
"temperature": 0}
|
||
last_error = None
|
||
for attempt in range(self.retries + 1):
|
||
try:
|
||
with httpx.Client(timeout=self.timeout, transport=self.transport) as client:
|
||
response = client.post(self.base_url + "/chat/completions", json=body,
|
||
headers={"Authorization": f"Bearer {self.api_key}"})
|
||
response.raise_for_status()
|
||
raw = response.json()["choices"][0]["message"]["content"]
|
||
return schema.model_validate(json.loads(raw))
|
||
except (httpx.HTTPError, ValueError, KeyError, IndexError, ValidationError) as exc:
|
||
# Never include response bodies or headers: providers sometimes echo credentials.
|
||
last_error = _safe_error_detail(exc)
|
||
if attempt < self.retries:
|
||
time.sleep(min(2 ** attempt, 5))
|
||
raise ModelError(f"模型请求或JSON校验失败（{last_error}），已重试{self.retries}次")
|
||
|
||
def tool_call(self, messages: list[dict], tools: list[dict], *, tool_choice="auto"):
|
||
"""OpenAI-compatible native tool calls. Credentials and provider bodies never enter logs."""
|
||
body = {"model": self.model, "messages": messages, "tools": tools,
|
||
"tool_choice": tool_choice, "temperature": 0}
|
||
try:
|
||
with httpx.Client(timeout=min(self.timeout, 180), transport=self.transport) as client:
|
||
response = client.post(self.base_url + "/chat/completions", json=body,
|
||
headers={"Authorization": f"Bearer {self.api_key}"})
|
||
if response.status_code in (400, 404, 422):
|
||
raise ModelError("当前端点或模型不支持原生 tool_calls；请在模型设置中配置支持工具调用的 OpenAI 兼容端点和模型")
|
||
response.raise_for_status()
|
||
message = response.json()["choices"][0]["message"]
|
||
if not isinstance(message, dict) or not isinstance(message.get("tool_calls", []), list):
|
||
raise ModelError("模型返回格式不包含有效的 tool_calls 消息")
|
||
return {"role": "assistant", "content": message.get("content"), "tool_calls": message.get("tool_calls", [])}
|
||
except ModelError:
|
||
raise
|
||
except (httpx.HTTPError, KeyError, IndexError, ValueError, TypeError) as exc:
|
||
raise ModelError(f"对话模型请求失败（{type(exc).__name__}）；请检查端点、模型和网络") from exc
|
||
|
||
def verify_tool_support(self):
|
||
"""Small capability probe without student or course content."""
|
||
probe = [{"type": "function", "function": {"name": "wisegrade_ping",
|
||
"description": "Return a ping for compatibility verification.",
|
||
"parameters": {"type": "object", "properties": {}, "additionalProperties": False}}}]
|
||
# Some compatible endpoints implement auto tool selection but reject "required".
|
||
response = self.tool_call([{"role": "user", "content": "For compatibility verification, call wisegrade_ping now and do not answer in text."}], probe)
|
||
if not any(call.get("function", {}).get("name") == "wisegrade_ping" for call in response["tool_calls"]):
|
||
raise ModelError("当前端点或模型未返回原生 tool_calls；请配置支持工具调用的端点和模型")
|
||
|
||
|
||
def locate_quote(text: str, quote: str):
|
||
"""Locate a model-written quotation inside the retrieved text.
|
||
|
||
PDF line breaks and punctuation differ from what the model reproduces, so matching
|
||
is done on NFKC-normalized text with whitespace and punctuation removed, while the
|
||
returned string is still the exact source substring.
|
||
"""
|
||
if not text or not quote:
|
||
return None
|
||
if quote in text:
|
||
return quote
|
||
|
||
def comparable(value: str):
|
||
return "".join(char for original in value for char in unicodedata.normalize("NFKC", original)
|
||
if not char.isspace() and not unicodedata.category(char).startswith("P"))
|
||
|
||
needle = comparable(quote)
|
||
if len(needle) < 4:
|
||
return None
|
||
positions = []
|
||
normalized = []
|
||
for index, original in enumerate(text):
|
||
for char in unicodedata.normalize("NFKC", original):
|
||
if not char.isspace() and not unicodedata.category(char).startswith("P"):
|
||
normalized.append(char)
|
||
positions.append(index)
|
||
start = "".join(normalized).find(needle)
|
||
if start < 0:
|
||
return None
|
||
return text[positions[start]:positions[start + len(needle) - 1] + 1]
|
||
|
||
|
||
def validate_score(result: ScoreResult, criterion: dict, chunks: list[dict], *, targets=None,
|
||
require_content_evidence: bool | None = None):
|
||
"""Check identity, quotation fidelity and the evidence-source policy.
|
||
|
||
The model may understand evidence and propose re-checks; the page permissions are
|
||
enforced here, in code, and cannot be argued away by the model's own reasoning.
|
||
"""
|
||
from .evidence import (USE_CONTENT, content_evidence_allowed, heading_only_text,
|
||
normalize_targets, page_type_allowed)
|
||
|
||
if result.criterion_id != criterion["criterion_id"] or result.max_score != criterion["max_score"]:
|
||
raise ModelError("模型评分项或满分与规则版本不一致")
|
||
if require_content_evidence is None:
|
||
require_content_evidence = any(target["use"] == USE_CONTENT
|
||
for target in normalize_targets(criterion.get("required_evidence")))
|
||
targets = targets if targets is not None else normalize_targets(criterion.get("required_evidence"))
|
||
by_target = {target["target_id"]: target for target in targets}
|
||
default_content = next((target for target in targets if target["use"] == USE_CONTENT), None)
|
||
by_id = {chunk["chunk_id"]: chunk for chunk in chunks}
|
||
has_content_evidence = False
|
||
|
||
for evidence in result.evidence:
|
||
chunk = by_id.get(evidence.chunk_id)
|
||
quote = locate_quote(chunk["text"], evidence.quote) if chunk else None
|
||
if quote is None and len(evidence.quote) >= 8:
|
||
matches = [(candidate, located) for candidate in chunks
|
||
if (located := locate_quote(candidate["text"], evidence.quote)) is not None]
|
||
if len(matches) == 1:
|
||
chunk, quote = matches[0]
|
||
if quote is None:
|
||
raise ModelError("模型证据未能匹配检索原文")
|
||
page_type = chunk.get("page_type") or "unknown"
|
||
if evidence.target_id and evidence.target_id not in by_target:
|
||
raise ModelError("模型证据包含未知 target_id")
|
||
target = by_target.get(evidence.target_id)
|
||
if target is None and not evidence.target_id:
|
||
# Empty identifiers are accepted only for pre-object legacy rubrics.
|
||
if any(isinstance(value, dict) for value in criterion.get("required_evidence", [])):
|
||
raise ModelError("模型证据缺少 target_id")
|
||
target = default_content or (targets[0] if targets else None)
|
||
if target is None:
|
||
raise ModelError("模型证据未关联任何必备证据目标")
|
||
if chunk.get("target_ids") and target["target_id"] not in chunk["target_ids"]:
|
||
raise ModelError("模型证据未关联该片段的检索目标")
|
||
if not page_type_allowed(page_type, target["allowed_page_types"]):
|
||
raise ModelError(f"模型证据来自不允许的页面类型（{page_type}），不能支持「{target['label']}」")
|
||
if target["use"] == USE_CONTENT and not content_evidence_allowed(page_type):
|
||
raise ModelError(f"目录、摘要或参考文献页不能作为「{target['label']}」的内容证据")
|
||
if target["use"] == USE_CONTENT and heading_only_text(quote):
|
||
raise ModelError(f"「{target['label']}」的证据只有章节标题，没有实质内容")
|
||
if any(char in quote for char in ("\ufffd", "\ufffe", "\uffff")):
|
||
raise ModelError("模型证据含未解析字符，请核对原页")
|
||
has_content_evidence |= target["use"] == USE_CONTENT
|
||
evidence.chunk_id, evidence.page, evidence.quote = chunk["chunk_id"], chunk["page"], quote
|
||
evidence.target_id = target["target_id"]
|
||
if result.score is not None and result.score > 0 and require_content_evidence and not has_content_evidence:
|
||
raise ModelError("正分内容判断必须包含合法内容证据，不能仅引用结构信息")
|
||
return result
|
||
|