feat(response_validation): 新增 Pydantic 校验层，健壮化处理 LLM 生成内容

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):
         """
         保存调试数据到文件

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,

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],

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]:
         """
         从已清理的消息中提取文本消息用于话题分析。

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],

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)