144 lines
6.0 KiB
Python
144 lines
6.0 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)
|
|
|
|
|
|
class ScoreResult(BaseModel):
|
|
criterion_id: str
|
|
score: float = Field(ge=0)
|
|
max_score: float = Field(gt=0)
|
|
reason: str = Field(min_length=1)
|
|
evidence: list[EvidenceResult]
|
|
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 > self.max_score or (self.score > 0 and not self.evidence):
|
|
raise ValueError("score exceeds maximum or lacks evidence")
|
|
return self
|
|
|
|
|
|
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
|
|
|
|
|
|
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()}})
|
|
body = {"model": self.model, "response_format": {"type": "json_object"},
|
|
"messages": [{"role": "system", "content": prompt + " 严格输出一个JSON对象。"}, {"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 = type(exc).__name__
|
|
if attempt < self.retries:
|
|
time.sleep(min(2 ** attempt, 5))
|
|
raise ModelError(f"模型请求或JSON校验失败（{last_error}），已重试{self.retries}次")
|
|
|
|
|
|
def validate_score(result: ScoreResult, criterion: dict, chunks: list[dict]):
|
|
if result.criterion_id != criterion["criterion_id"] or result.max_score != criterion["max_score"]:
|
|
raise ModelError("模型评分项或满分与规则版本不一致")
|
|
by_id = {chunk["chunk_id"]: chunk for chunk in chunks}
|
|
|
|
def locate_quote(text: str, quote: str):
|
|
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"))
|
|
|
|
if quote in text:
|
|
return quote
|
|
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]
|
|
|
|
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("模型证据未能匹配检索原文")
|
|
evidence.chunk_id, evidence.page, evidence.quote = chunk["chunk_id"], chunk["page"], quote
|
|
return result
|