From 3df898ca6953dcf1676b163b0daace8fe7a3346d Mon Sep 17 00:00:00 2001 From: SXP-Simon Date: Thu, 26 Mar 2026 22:58:04 +0800 Subject: [PATCH] =?UTF-8?q?feat(response=5Fvalidation):=20=E6=96=B0?= =?UTF-8?q?=E5=A2=9E=20Pydantic=20=E6=A0=A1=E9=AA=8C=E5=B1=82=EF=BC=8C?= =?UTF-8?q?=E5=81=A5=E5=A3=AE=E5=8C=96=E5=A4=84=E7=90=86=20LLM=20=E7=94=9F?= =?UTF-8?q?=E6=88=90=E5=86=85=E5=AE=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../analysis/analyzers/base_analyzer.py | 27 +++- .../analyzers/chat_quality_analyzer.py | 53 ++++++-- .../analyzers/golden_quote_analyzer.py | 6 + .../analysis/analyzers/topic_analyzer.py | 6 + .../analysis/analyzers/user_title_analyzer.py | 6 + .../analysis/utils/response_validation.py | 127 ++++++++++++++++++ 6 files changed, 207 insertions(+), 18 deletions(-) create mode 100644 src/infrastructure/analysis/utils/response_validation.py diff --git a/src/infrastructure/analysis/analyzers/base_analyzer.py b/src/infrastructure/analysis/analyzers/base_analyzer.py index 25ad855..388e460 100644 --- a/src/infrastructure/analysis/analyzers/base_analyzer.py +++ b/src/infrastructure/analysis/analyzers/base_analyzer.py @@ -240,17 +240,36 @@ class BaseAnalyzer(ABC, Generic[TDataObject]): """ success, parsed_data, error_msg = self.parse_structured_response(result_text) if success and parsed_data: - return True, parsed_data, None + validated_success, validated_data, validated_error = ( + self.validate_parsed_data(parsed_data) + ) + if validated_success and validated_data: + return True, validated_data, None + error_msg = validated_error or error_msg regex_data = self.extract_with_regex(result_text, self.get_max_count()) if regex_data: - logger.info( - f"{self.get_data_type()}结构化解析失败后，正则降级提取成功，获得 {len(regex_data)} 条数据" + validated_success, validated_data, validated_error = ( + self.validate_parsed_data(regex_data) ) - return True, regex_data, None + if validated_success and validated_data: + logger.info( + f"{self.get_data_type()}结构化解析失败后，正则降级提取成功，获得 {len(validated_data)} 条数据" + ) + return True, validated_data, None + error_msg = validated_error or error_msg return False, None, error_msg + def validate_parsed_data( + self, data_list: list[dict] + ) -> tuple[bool, list[dict] | None, str | None]: + """ + 解析结果的本地二次校验（默认直接通过）。 + 子类可重写为 Pydantic 校验。 + """ + return True, data_list, None + def _save_debug_data(self, prompt: str, session_id: str): """ 保存调试数据到文件 diff --git a/src/infrastructure/analysis/analyzers/chat_quality_analyzer.py b/src/infrastructure/analysis/analyzers/chat_quality_analyzer.py index 522aa96..be788dc 100644 --- a/src/infrastructure/analysis/analyzers/chat_quality_analyzer.py +++ b/src/infrastructure/analysis/analyzers/chat_quality_analyzer.py @@ -14,6 +14,7 @@ from ..utils.llm_utils import ( extract_response_text, extract_token_usage, ) +from ..utils.response_validation import validate_quality_review_item from ..utils.structured_output_schema import JSONObject, build_chat_quality_schema from .base_analyzer import BaseAnalyzer @@ -196,6 +197,11 @@ class ChatQualityAnalyzer(BaseAnalyzer[QualityReview]): summary=data.get("summary", "今天也是充满活力的一天。"), ) + def _validate_review_payload( + self, data: dict + ) -> tuple[bool, dict | None, str | None]: + return validate_quality_review_item(data) + async def _retry_parse_quality_object( self, *, @@ -244,11 +250,15 @@ class ChatQualityAnalyzer(BaseAnalyzer[QualityReview]): retry_text, self.get_data_type() ) if retry_success and retry_parsed_data: - return retry_parsed_data + valid, normalized, _ = self._validate_review_payload(retry_parsed_data) + if valid and normalized: + return normalized retry_regex_data = extract_quality_with_regex(retry_text) if retry_regex_data: - return retry_regex_data + valid, normalized, _ = self._validate_review_payload(retry_regex_data) + if valid and normalized: + return normalized return None @@ -354,11 +364,16 @@ class ChatQualityAnalyzer(BaseAnalyzer[QualityReview]): ) if success and parsed_data: - review = self._build_review_from_dict(parsed_data) - logger.info( - f"聊天质量汇总分析成功，解析到 {len(review.dimensions)} 个汇总维度" + valid, normalized, validation_error = self._validate_review_payload( + parsed_data ) - return review, usage + if valid and normalized: + review = self._build_review_from_dict(normalized) + logger.info( + f"聊天质量汇总分析成功，解析到 {len(review.dimensions)} 个汇总维度" + ) + return review, usage + error_msg = validation_error or error_msg repaired_data = await self._retry_parse_quality_object( original_prompt=prompt, @@ -444,19 +459,29 @@ class ChatQualityAnalyzer(BaseAnalyzer[QualityReview]): ) if success and parsed_data: - review = self._build_review_from_dict(parsed_data) - logger.debug( - f"聊天质量分析成功，解析到 {len(review.dimensions)} 个维度" + valid, normalized, validation_error = self._validate_review_payload( + parsed_data ) - return review, usage + if valid and normalized: + review = self._build_review_from_dict(normalized) + logger.debug( + f"聊天质量分析成功，解析到 {len(review.dimensions)} 个维度" + ) + return review, usage + error_msg = validation_error or error_msg regex_data = extract_quality_with_regex(result_text) if regex_data: - review = self._build_review_from_dict(regex_data) - logger.debug( - f"聊天质量首轮结构化失败后，正则提取成功，获得 {len(review.dimensions)} 个维度" + valid, normalized, validation_error = self._validate_review_payload( + regex_data ) - return review, usage + if valid and normalized: + review = self._build_review_from_dict(normalized) + logger.debug( + f"聊天质量首轮结构化失败后，正则提取成功，获得 {len(review.dimensions)} 个维度" + ) + return review, usage + error_msg = validation_error or error_msg repaired_data = await self._retry_parse_quality_object( original_prompt=prompt, diff --git a/src/infrastructure/analysis/analyzers/golden_quote_analyzer.py b/src/infrastructure/analysis/analyzers/golden_quote_analyzer.py index 21acb93..236c19d 100644 --- a/src/infrastructure/analysis/analyzers/golden_quote_analyzer.py +++ b/src/infrastructure/analysis/analyzers/golden_quote_analyzer.py @@ -9,6 +9,7 @@ from ....domain.models.data_models import GoldenQuote, TokenUsage from ....utils.logger import logger from ..utils import InfoUtils from ..utils.json_utils import extract_golden_quotes_with_regex +from ..utils.response_validation import validate_golden_quote_items from ..utils.structured_output_schema import JSONObject, build_golden_quotes_schema from .base_analyzer import BaseAnalyzer @@ -126,6 +127,11 @@ class GoldenQuoteAnalyzer(BaseAnalyzer[GoldenQuote]): logger.error(f"创建金句对象失败: {e}") return [] + def validate_parsed_data( + self, data_list: list[dict] + ) -> tuple[bool, list[dict] | None, str | None]: + return validate_golden_quote_items(data_list) + async def analyze_golden_quotes( self, messages: list[dict], diff --git a/src/infrastructure/analysis/analyzers/topic_analyzer.py b/src/infrastructure/analysis/analyzers/topic_analyzer.py index bbc38c3..8e272b0 100644 --- a/src/infrastructure/analysis/analyzers/topic_analyzer.py +++ b/src/infrastructure/analysis/analyzers/topic_analyzer.py @@ -10,6 +10,7 @@ from ....domain.models.data_models import SummaryTopic, TokenUsage from ....utils.logger import logger from ..utils import InfoUtils from ..utils.json_utils import extract_topics_with_regex +from ..utils.response_validation import validate_topic_items from ..utils.structured_output_schema import JSONObject, build_topics_schema from .base_analyzer import BaseAnalyzer @@ -262,6 +263,11 @@ class TopicAnalyzer(BaseAnalyzer[SummaryTopic]): logger.error(f"创建话题对象失败: {e}", exc_info=True) return [] + def validate_parsed_data( + self, data_list: list[dict] + ) -> tuple[bool, list[dict] | None, str | None]: + return validate_topic_items(data_list) + def extract_text_messages(self, messages: list[dict]) -> list[dict]: """ 从已清理的消息中提取文本消息用于话题分析。 diff --git a/src/infrastructure/analysis/analyzers/user_title_analyzer.py b/src/infrastructure/analysis/analyzers/user_title_analyzer.py index d10dbce..7c2b84d 100644 --- a/src/infrastructure/analysis/analyzers/user_title_analyzer.py +++ b/src/infrastructure/analysis/analyzers/user_title_analyzer.py @@ -6,6 +6,7 @@ from ....domain.models.data_models import TokenUsage, UserTitle from ....utils.logger import logger from ..utils.json_utils import extract_user_titles_with_regex +from ..utils.response_validation import validate_user_title_items from ..utils.structured_output_schema import JSONObject, build_user_titles_schema from .base_analyzer import BaseAnalyzer @@ -140,6 +141,11 @@ class UserTitleAnalyzer(BaseAnalyzer[UserTitle]): logger.error(f"创建用户称号对象失败: {e}") return [] + def validate_parsed_data( + self, data_list: list[dict] + ) -> tuple[bool, list[dict] | None, str | None]: + return validate_user_title_items(data_list) + def prepare_user_data( self, messages: list[dict], diff --git a/src/infrastructure/analysis/utils/response_validation.py b/src/infrastructure/analysis/utils/response_validation.py new file mode 100644 index 0000000..b5ebbbe --- /dev/null +++ b/src/infrastructure/analysis/utils/response_validation.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict, ValidationError, field_validator + + +class TopicItemModel(BaseModel): + model_config = ConfigDict(extra="forbid") + + topic: str + contributors: list[str] + detail: str + + @field_validator("topic", "detail", mode="before") + @classmethod + def _normalize_text(cls, value: object) -> str: + return str(value).strip() + + @field_validator("contributors", mode="before") + @classmethod + def _normalize_contributors(cls, value: object) -> list[str]: + if not isinstance(value, list): + return [] + contributors: list[str] = [] + for item in value: + text = str(item).strip() + if text: + contributors.append(text) + return contributors + + +class UserTitleItemModel(BaseModel): + model_config = ConfigDict(extra="forbid") + + name: str + user_id: str + title: str + mbti: str + reason: str + + @field_validator("name", "user_id", "title", "mbti", "reason", mode="before") + @classmethod + def _normalize_text(cls, value: object) -> str: + return str(value).strip() + + +class GoldenQuoteItemModel(BaseModel): + model_config = ConfigDict(extra="forbid") + + content: str + sender: str + reason: str + + @field_validator("content", "sender", "reason", mode="before") + @classmethod + def _normalize_text(cls, value: object) -> str: + return str(value).strip() + + +class QualityDimensionModel(BaseModel): + model_config = ConfigDict(extra="forbid") + + name: str + percentage: float + comment: str + + @field_validator("name", "comment", mode="before") + @classmethod + def _normalize_text(cls, value: object) -> str: + return str(value).strip() + + +class QualityReviewModel(BaseModel): + model_config = ConfigDict(extra="forbid") + + title: str + subtitle: str + dimensions: list[QualityDimensionModel] + summary: str + + @field_validator("title", "subtitle", "summary", mode="before") + @classmethod + def _normalize_text(cls, value: object) -> str: + return str(value).strip() + + +def validate_topic_items( + data_list: list[dict], +) -> tuple[bool, list[dict] | None, str | None]: + try: + normalized = [ + TopicItemModel.model_validate(item).model_dump() for item in data_list + ] + return True, normalized, None + except ValidationError as e: + return False, None, str(e) + + +def validate_user_title_items( + data_list: list[dict], +) -> tuple[bool, list[dict] | None, str | None]: + try: + normalized = [ + UserTitleItemModel.model_validate(item).model_dump() for item in data_list + ] + return True, normalized, None + except ValidationError as e: + return False, None, str(e) + + +def validate_golden_quote_items( + data_list: list[dict], +) -> tuple[bool, list[dict] | None, str | None]: + try: + normalized = [ + GoldenQuoteItemModel.model_validate(item).model_dump() for item in data_list + ] + return True, normalized, None + except ValidationError as e: + return False, None, str(e) + + +def validate_quality_review_item(data: dict) -> tuple[bool, dict | None, str | None]: + try: + normalized = QualityReviewModel.model_validate(data).model_dump() + return True, normalized, None + except ValidationError as e: + return False, None, str(e)