feat(response_format): 使用 OpenAI 格式的 response_format 参数约束 json 格式化数据生成，并且兼容旧版本逻辑，等待本体合并相关参数的传入

_conf_schema.json

@@ -185,30 +185,6 @@
                 "description": "LLM 请求重试退避基值（秒）",
                 "default": 2,
                 "hint": "重试之间的基准等待时间（秒），实际等待时间为基值乘以尝试次数。"
-            },
-            "topic_max_tokens": {
-                "type": "int",
-                "description": "（兼容部分提供商）话题分析最大 Token 数",
-                "default": 12288,
-                "hint": "（兼容部分提供商，实测大部分模型调整后没有明显效果）话题分析时 LLM 能生成的最大 token 数量。当分析内容较多或者分析提示词复杂时，建议适当调大此值以保证输出质量。"
-            },
-            "golden_quote_max_tokens": {
-                "type": "int",
-                "description": "（兼容部分提供商）金句分析最大 Token 数",
-                "default": 4096,
-                "hint": "（兼容部分提供商，实测大部分模型调整后没有明显效果）金句分析时 LLM 能生成的最大 token 数量。当分析内容较多或者分析提示词复杂时，建议适当调大此值以保证输出质量。"
-            },
-            "user_title_max_tokens": {
-                "type": "int",
-                "description": "（兼容部分提供商）用户称号分析最大 Token 数",
-                "default": 4096,
-                "hint": "（兼容部分提供商，实测大部分模型调整后没有明显效果）用户称号分析时 LLM 能生成的最大 token 数量。当分析内容较多或者分析提示词复杂时，建议适当调大此值以保证输出质量。"
-            },
-            "quality_max_tokens": {
-                "type": "int",
-                "description": "（兼容部分提供商）聊天质量分析最大 Token 数",
-                "default": 4096,
-                "hint": "聊天质量分析时 LLM 能生成的最大 token 数量。"
             }
         }
     },

src/infrastructure/analysis/analyzers/base_analyzer.py

@@ -4,7 +4,8 @@
 """
 
 from abc import ABC, abstractmethod
-from typing import Any
+from collections.abc import Sized
+from typing import Generic, TypeVar
 
 from ....domain.models.data_models import TokenUsage
 from ....utils.logger import logger
@@ -13,10 +14,14 @@ from ..utils.llm_utils import (
     call_provider_with_retry,
     extract_response_text,
     extract_token_usage,
+    get_provider_id_with_fallback,
 )
+from ..utils.structured_output_schema import JSONObject, build_response_format
+
+TDataObject = TypeVar("TDataObject")
 
 
-class BaseAnalyzer(ABC):
+class BaseAnalyzer(ABC, Generic[TDataObject]):
     """
     基础分析器抽象类
     定义所有分析器的通用接口 and 流程
@@ -66,7 +71,7 @@ class BaseAnalyzer(ABC):
         pass
 
     @abstractmethod
-    def build_prompt(self, data: Any) -> str:
+    def build_prompt(self, data: object) -> str:
         """
         构建LLM提示词
 
@@ -93,7 +98,7 @@ class BaseAnalyzer(ABC):
         pass
 
     @abstractmethod
-    def create_data_objects(self, data_list: list[dict]) -> list[Any]:
+    def create_data_objects(self, data_list: list[dict]) -> list[TDataObject]:
         """
         创建数据对象列表
 
@@ -105,6 +110,147 @@ class BaseAnalyzer(ABC):
         """
         pass
 
+    def get_response_schema_name(self) -> str:
+        return f"{self.get_data_type()}_output"
+
+    def get_response_schema(self) -> JSONObject | None:
+        return None
+
+    def get_response_format(self) -> JSONObject | None:
+        schema = self.get_response_schema()
+        if not schema:
+            return None
+        return build_response_format(self.get_response_schema_name(), schema)
+
+    def get_schema_retry_max_attempts(self) -> int:
+        """
+        schema 解析失败后的最大重试次数（不含首轮请求）。
+        """
+        return 2
+
+    def get_schema_retry_temperatures(
+        self, base_temperature: float | None
+    ) -> tuple[float, ...]:
+        """
+        schema 解析失败后的温度重试序列（不含首轮请求）。
+        采用动态降温，提高结构化稳定性。
+        """
+        attempts = max(0, self.get_schema_retry_max_attempts())
+        if attempts == 0:
+            return ()
+
+        base = base_temperature if base_temperature is not None else 0.7
+        first_retry = max(0.1, min(2.0, base * 0.5))
+
+        temperatures: list[float] = [round(first_retry, 2), 0.0]
+        if attempts < len(temperatures):
+            temperatures = temperatures[:attempts]
+
+        deduped: list[float] = []
+        for temp in temperatures:
+            if not deduped or deduped[-1] != temp:
+                deduped.append(temp)
+        return tuple(deduped)
+
+    async def _resolve_provider_temperature(
+        self,
+        provider_id_key: str | None,
+        umo: str | None,
+    ) -> float | None:
+        """
+        尝试从当前将要调用的 Provider 配置中解析基础 temperature。
+        """
+        provider_id = await get_provider_id_with_fallback(
+            self.context,
+            self.config_manager,
+            provider_id_key,
+            umo,
+        )
+        if not provider_id:
+            return None
+
+        provider = self.context.get_provider_by_id(provider_id=provider_id)
+        if provider is None:
+            return None
+
+        provider_config_obj = getattr(provider, "provider_config", None)
+        if not isinstance(provider_config_obj, dict):
+            return None
+
+        raw_temperature = provider_config_obj.get("temperature")
+        if raw_temperature is None:
+            custom_extra_body = provider_config_obj.get("custom_extra_body")
+            if isinstance(custom_extra_body, dict):
+                raw_temperature = custom_extra_body.get("temperature")
+
+        if isinstance(raw_temperature, bool):
+            return None
+
+        parsed_temperature: float | None = None
+        if isinstance(raw_temperature, (int, float)):
+            parsed_temperature = float(raw_temperature)
+        elif isinstance(raw_temperature, str):
+            try:
+                parsed_temperature = float(raw_temperature.strip())
+            except ValueError:
+                return None
+
+        if parsed_temperature is None:
+            return None
+
+        return max(0.0, min(2.0, parsed_temperature))
+
+    def parse_structured_response(
+        self, result_text: str
+    ) -> tuple[bool, list[dict] | None, str | None]:
+        """
+        解析结构化响应（默认 JSON 数组解析）。
+        子类可重写此方法定制对象解析逻辑。
+        """
+        return parse_json_response(result_text, self.get_data_type())
+
+    def build_schema_retry_prompt(
+        self,
+        original_prompt: str,
+        previous_output: str,
+        parse_error: str | None,
+        attempt_index: int,
+    ) -> str:
+        """
+        构建结构化失败后的修复重试提示词。
+        """
+        err_text = parse_error or "unknown_parse_error"
+        return (
+            f"{original_prompt}\n\n"
+            "[STRUCTURED OUTPUT RETRY]\n"
+            f"Attempt: {attempt_index}\n"
+            "Your previous output did not satisfy the required strict JSON schema.\n"
+            "Return ONLY valid JSON that strictly matches the schema. "
+            "Do not include markdown, explanation, or extra text.\n"
+            f"Parse error: {err_text}\n"
+            "Previous invalid output:\n"
+            f"{previous_output}"
+        )
+
+    def _try_parse_with_fallback(
+        self, result_text: str
+    ) -> tuple[bool, list[dict] | None, str | None]:
+        """
+        先尝试结构化 JSON 解析（含修复逻辑），失败后立即尝试正则降级。
+        """
+        success, parsed_data, error_msg = self.parse_structured_response(result_text)
+        if success and parsed_data:
+            return True, parsed_data, None
+
+        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)} 条数据"
+            )
+            return True, regex_data, None
+
+        return False, None, error_msg
+
     def _save_debug_data(self, prompt: str, session_id: str):
         """
         保存调试数据到文件
@@ -139,8 +285,8 @@ class BaseAnalyzer(ABC):
             logger.error(f"保存调试数据失败: {e}", exc_info=True)
 
     async def analyze(
-        self, data: Any, umo: str | None = None, session_id: str | None = None
-    ) -> tuple[list[Any], TokenUsage]:
+        self, data: object, umo: str | None = None, session_id: str | None = None
+    ) -> tuple[list[TDataObject], TokenUsage]:
         """
         统一的分析流程
 
@@ -157,9 +303,8 @@ class BaseAnalyzer(ABC):
             logger.debug(
                 f"{self.get_data_type()}分析开始构建prompt，输入数据类型: {type(data)}"
             )
-            logger.debug(
-                f"{self.get_data_type()}分析输入数据长度: {len(data) if hasattr(data, '__len__') else 'N/A'}"
-            )
+            data_length = len(data) if isinstance(data, Sized) else "N/A"
+            logger.debug(f"{self.get_data_type()}分析输入数据长度: {data_length}")
 
             prompt = self.build_prompt(data)
             logger.info(f"开始{self.get_data_type()}分析，构建提示词完成")
@@ -185,9 +330,10 @@ class BaseAnalyzer(ABC):
                 return [], TokenUsage()
 
             # 2. 调用LLM（使用配置的 provider）
-            max_tokens = self.get_max_tokens()
-            temperature = self.get_temperature()
             provider_id_key = self.get_provider_id_key()
+            base_temperature = await self._resolve_provider_temperature(
+                provider_id_key, umo
+            )
 
             # 获取人格设定
             system_prompt = await self._build_system_prompt(umo)
@@ -215,11 +361,10 @@ class BaseAnalyzer(ABC):
                 self.context,
                 self.config_manager,
                 prompt=prompt,
-                max_tokens=max_tokens,
-                temperature=temperature,
                 umo=umo,
                 provider_id_key=provider_id_key,
                 system_prompt=system_prompt,
+                response_format=self.get_response_format(),
             )
 
             if response is None:
@@ -240,10 +385,50 @@ class BaseAnalyzer(ABC):
             result_text = extract_response_text(response)
             logger.debug(f"{self.get_data_type()}分析原始响应: {result_text[:500]}...")
 
-            # 5. 尝试JSON解析
-            success, parsed_data, error_msg = parse_json_response(
-                result_text, self.get_data_type()
-            )
+            # 5. 尝试结构化解析 + 正则降级解析
+            success, parsed_data, error_msg = self._try_parse_with_fallback(result_text)
+
+            # 5.1 仅在两种解析方式都失败时，进入 schema 修复重试（温度递减）
+            if not success and self.get_response_format() is not None:
+                temperatures = self.get_schema_retry_temperatures(base_temperature)
+                for idx, temperature in enumerate(temperatures, start=1):
+                    retry_prompt = self.build_schema_retry_prompt(
+                        original_prompt=prompt,
+                        previous_output=result_text,
+                        parse_error=error_msg,
+                        attempt_index=idx,
+                    )
+                    logger.warning(
+                        f"{self.get_data_type()}结构化解析失败，触发 schema 修复重试 "
+                        f"(attempt={idx}, temperature={temperature:.1f})"
+                    )
+                    retry_response = await call_provider_with_retry(
+                        self.context,
+                        self.config_manager,
+                        prompt=retry_prompt,
+                        umo=umo,
+                        provider_id_key=provider_id_key,
+                        system_prompt=system_prompt,
+                        response_format=self.get_response_format(),
+                        extra_generate_kwargs={"temperature": temperature},
+                    )
+                    if retry_response is None:
+                        continue
+
+                    retry_result_text = extract_response_text(retry_response)
+                    if not retry_result_text:
+                        continue
+
+                    result_text = retry_result_text
+                    retry_success, retry_parsed_data, retry_error_msg = (
+                        self._try_parse_with_fallback(retry_result_text)
+                    )
+                    if retry_success:
+                        success = True
+                        parsed_data = retry_parsed_data
+                        error_msg = None
+                        break
+                    error_msg = retry_error_msg
 
             if success and parsed_data:
                 # JSON解析成功，创建数据对象
@@ -253,47 +438,16 @@ class BaseAnalyzer(ABC):
                 )
                 return data_objects, token_usage
 
-            # 6. JSON解析失败，使用正则表达式降级
-            logger.warning(
-                f"{self.get_data_type()}JSON解析失败，尝试正则表达式提取: {error_msg}"
+            # 6. 全部尝试失败
+            logger.error(
+                f"{self.get_data_type()}分析失败: JSON解析与正则降级均未成功: {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)} 条数据"
-                )
-                data_objects = self.create_data_objects(regex_data)
-                return data_objects, token_usage
-            else:
-                # 最后的降级方案 - 两种方法都失败
-                logger.error(
-                    f"{self.get_data_type()}分析失败: JSON解析和正则表达式提取均未成功，返回空列表"
-                )
-                return [], token_usage
+            return [], token_usage
 
         except Exception as e:
             logger.error(f"{self.get_data_type()}分析失败: {e}", exc_info=True)
             return [], TokenUsage()
 
-    def get_max_tokens(self) -> int:
-        """
-        获取最大token数，子类可重写
-
-        Returns:
-            最大token数
-        """
-        return 10000
-
-    def get_temperature(self) -> float:
-        """
-        获取温度参数，子类可重写
-
-        Returns:
-            温度参数
-        """
-        return 0.6
-
     async def _build_system_prompt(self, umo: str | None) -> str | None:
         """
         构建带有会话人格的系统提示词

src/infrastructure/analysis/analyzers/chat_quality_analyzer.py

@@ -14,10 +14,11 @@ from ..utils.llm_utils import (
     extract_response_text,
     extract_token_usage,
 )
+from ..utils.structured_output_schema import JSONObject, build_chat_quality_schema
 from .base_analyzer import BaseAnalyzer
 
 
-class ChatQualityAnalyzer(BaseAnalyzer):
+class ChatQualityAnalyzer(BaseAnalyzer[QualityReview]):
     """
     聊天质量分析器
     专门处理群聊质量的锐评和多维度分析
@@ -39,13 +40,11 @@ class ChatQualityAnalyzer(BaseAnalyzer):
         """获取最大维度数量"""
         return 8
 
-    def get_max_tokens(self) -> int:
-        """获取最大token数"""
-        return self.config_manager.get_quality_max_tokens()
+    def get_response_schema_name(self) -> str:
+        return "daily_chat_quality_review"
 
-    def get_temperature(self) -> float:
-        """获取温度参数"""
-        return 0.8
+    def get_response_schema(self) -> JSONObject:
+        return build_chat_quality_schema(self.get_max_count())
 
     def build_prompt(self, data: list[dict]) -> str:
         """
@@ -197,6 +196,62 @@ class ChatQualityAnalyzer(BaseAnalyzer):
             summary=data.get("summary", "今天也是充满活力的一天。"),
         )
 
+    async def _retry_parse_quality_object(
+        self,
+        *,
+        original_prompt: str,
+        previous_output: str,
+        parse_error: str | None,
+        umo: str | None,
+        system_prompt: str | None,
+        base_temperature: float | None,
+    ) -> dict | None:
+        response_format = self.get_response_format()
+        if response_format is None:
+            return None
+
+        for idx, temperature in enumerate(
+            self.get_schema_retry_temperatures(base_temperature), start=1
+        ):
+            retry_prompt = self.build_schema_retry_prompt(
+                original_prompt=original_prompt,
+                previous_output=previous_output,
+                parse_error=parse_error,
+                attempt_index=idx,
+            )
+            logger.warning(
+                f"聊天质量结构化解析失败，触发 schema 修复重试 "
+                f"(attempt={idx}, temperature={temperature:.1f})"
+            )
+            retry_response = await call_provider_with_retry(
+                self.context,
+                self.config_manager,
+                prompt=retry_prompt,
+                umo=umo,
+                provider_id_key=self.get_provider_id_key(),
+                system_prompt=system_prompt,
+                response_format=response_format,
+                extra_generate_kwargs={"temperature": temperature},
+            )
+            if retry_response is None:
+                continue
+
+            retry_text = extract_response_text(retry_response)
+            if not retry_text:
+                continue
+
+            retry_success, retry_parsed_data, _ = parse_json_object_response(
+                retry_text, self.get_data_type()
+            )
+            if retry_success and retry_parsed_data:
+                return retry_parsed_data
+
+            retry_regex_data = extract_quality_with_regex(retry_text)
+            if retry_regex_data:
+                return retry_regex_data
+
+        return None
+
     async def summarize_batch_reviews(
         self,
         batch_reviews: list[dict],
@@ -266,16 +321,18 @@ class ChatQualityAnalyzer(BaseAnalyzer):
 
             # 调用 LLM 进行汇总
             system_prompt = await self._build_system_prompt(umo)
+            base_temperature = await self._resolve_provider_temperature(
+                self.get_provider_id_key(), umo
+            )
 
             response = await call_provider_with_retry(
                 self.context,
                 self.config_manager,
                 prompt=prompt,
-                max_tokens=self.get_max_tokens(),
-                temperature=0.7,
                 umo=umo,
                 provider_id_key=self.get_provider_id_key(),
                 system_prompt=system_prompt,
+                response_format=self.get_response_format(),
             )
 
             if response is None:
@@ -303,6 +360,21 @@ class ChatQualityAnalyzer(BaseAnalyzer):
                 )
                 return review, usage
 
+            repaired_data = await self._retry_parse_quality_object(
+                original_prompt=prompt,
+                previous_output=result_text,
+                parse_error=error_msg,
+                umo=umo,
+                system_prompt=system_prompt,
+                base_temperature=base_temperature,
+            )
+            if repaired_data:
+                review = self._build_review_from_dict(repaired_data)
+                logger.info(
+                    f"聊天质量汇总 schema 修复重试成功，解析到 {len(review.dimensions)} 个汇总维度"
+                )
+                return review, usage
+
             # 降级：如果汇总失败，返回最新的一个
             logger.warning(f"聊天质量汇总分析失败，降级使用最新批次: {error_msg}")
             return self._build_review_from_dict(batch_reviews[-1]), usage
@@ -330,6 +402,9 @@ class ChatQualityAnalyzer(BaseAnalyzer):
         try:
             # 1. 获取人格设定
             system_prompt = await self._build_system_prompt(umo)
+            base_temperature = await self._resolve_provider_temperature(
+                self.get_provider_id_key(), umo
+            )
 
             # 2. 构建 prompt
             prompt = self.build_prompt(messages)
@@ -341,11 +416,10 @@ class ChatQualityAnalyzer(BaseAnalyzer):
                 self.context,
                 self.config_manager,
                 prompt=prompt,
-                max_tokens=self.get_max_tokens(),
-                temperature=self.get_temperature(),
                 umo=umo,
                 provider_id_key=self.get_provider_id_key(),
                 system_prompt=system_prompt,
+                response_format=self.get_response_format(),
             )
 
             if response is None:
@@ -376,19 +450,31 @@ class ChatQualityAnalyzer(BaseAnalyzer):
                 )
                 return review, usage
 
-            # 7. 正则降级（使用 extract_quality_with_regex）
-            logger.warning(f"聊天质量JSON解析失败，尝试正则表达式提取: {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)} 个维度"
+                    f"聊天质量首轮结构化失败后，正则提取成功，获得 {len(review.dimensions)} 个维度"
                 )
                 return review, usage
 
-            # 8. 全部失败
-            logger.error("聊天质量分析失败: JSON解析和正则表达式提取均未成功")
+            repaired_data = await self._retry_parse_quality_object(
+                original_prompt=prompt,
+                previous_output=result_text,
+                parse_error=error_msg,
+                umo=umo,
+                system_prompt=system_prompt,
+                base_temperature=base_temperature,
+            )
+            if repaired_data:
+                review = self._build_review_from_dict(repaired_data)
+                logger.debug(
+                    f"聊天质量 schema 修复重试成功，解析到 {len(review.dimensions)} 个维度"
+                )
+                return review, usage
+
+            # 7. 全部失败
+            logger.error(f"聊天质量分析失败: JSON解析和正则提取均未成功: {error_msg}")
             return None, usage
 
         except Exception as e:

src/infrastructure/analysis/analyzers/golden_quote_analyzer.py

@@ -9,10 +9,11 @@ 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.structured_output_schema import JSONObject, build_golden_quotes_schema
 from .base_analyzer import BaseAnalyzer
 
 
-class GoldenQuoteAnalyzer(BaseAnalyzer):
+class GoldenQuoteAnalyzer(BaseAnalyzer[GoldenQuote]):
     """
     金句分析器
     专门处理群聊金句的提取和分析
@@ -32,13 +33,11 @@ class GoldenQuoteAnalyzer(BaseAnalyzer):
             return self._incremental_max_count
         return self.config_manager.get_max_golden_quotes()
 
-    def get_max_tokens(self) -> int:
-        """获取最大token数"""
-        return self.config_manager.get_golden_quote_max_tokens()
+    def get_response_schema_name(self) -> str:
+        return "daily_golden_quotes"
 
-    def get_temperature(self) -> float:
-        """获取温度参数"""
-        return 0.7
+    def get_response_schema(self) -> JSONObject:
+        return build_golden_quotes_schema(self.get_max_count())
 
     def build_prompt(self, data: list[dict]) -> str:
         """

src/infrastructure/analysis/analyzers/topic_analyzer.py

@@ -10,10 +10,11 @@ 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.structured_output_schema import JSONObject, build_topics_schema
 from .base_analyzer import BaseAnalyzer
 
 
-class TopicAnalyzer(BaseAnalyzer):
+class TopicAnalyzer(BaseAnalyzer[SummaryTopic]):
     """
     话题分析器
     专门处理群聊话题的提取和分析
@@ -33,13 +34,11 @@ class TopicAnalyzer(BaseAnalyzer):
             return self._incremental_max_count
         return self.config_manager.get_max_topics()
 
-    def get_max_tokens(self) -> int:
-        """获取最大token数"""
-        return self.config_manager.get_topic_max_tokens()
+    def get_response_schema_name(self) -> str:
+        return "daily_topics"
 
-    def get_temperature(self) -> float:
-        """获取温度参数"""
-        return 0.6
+    def get_response_schema(self) -> JSONObject:
+        return build_topics_schema(self.get_max_count())
 
     def build_prompt(self, data: list[dict]) -> str:
         """

src/infrastructure/analysis/analyzers/user_title_analyzer.py

@@ -6,10 +6,11 @@
 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.structured_output_schema import JSONObject, build_user_titles_schema
 from .base_analyzer import BaseAnalyzer
 
 
-class UserTitleAnalyzer(BaseAnalyzer):
+class UserTitleAnalyzer(BaseAnalyzer[UserTitle]):
     """
     用户称号分析器
     专门处理用户称号分配和MBTI类型分析
@@ -27,13 +28,11 @@ class UserTitleAnalyzer(BaseAnalyzer):
         """获取最大用户称号数量"""
         return self.config_manager.get_max_user_titles()
 
-    def get_max_tokens(self) -> int:
-        """获取最大token数"""
-        return self.config_manager.get_user_title_max_tokens()
+    def get_response_schema_name(self) -> str:
+        return "daily_user_titles"
 
-    def get_temperature(self) -> float:
-        """获取温度参数"""
-        return 0.5
+    def get_response_schema(self) -> JSONObject:
+        return build_user_titles_schema(self.get_max_count())
 
     def build_prompt(self, data: dict) -> str:
         """

src/infrastructure/analysis/llm_analyzer.py

@@ -486,8 +486,6 @@ class LLMAnalyzer(IAnalysisProvider):
         self,
         provider,
         prompt: str,
-        max_tokens: int,
-        temperature: float,
         umo: str | None = None,
         provider_id_key: str | None = None,
     ):
@@ -498,8 +496,6 @@ class LLMAnalyzer(IAnalysisProvider):
         Args:
             provider: LLM服务商实例或None（已弃用，现在使用 provider_id_key）
             prompt: 输入的提示语
-            max_tokens: 最大生成token数
-            temperature: 采样温度
             umo: 指定使用的模型唯一标识符
             provider_id_key: 配置中的 provider_id 键名（可选）
 
@@ -510,8 +506,6 @@ class LLMAnalyzer(IAnalysisProvider):
             self.context,
             self.config_manager,
             prompt,
-            max_tokens,
-            temperature,
             umo,
             provider_id_key,
         )

src/infrastructure/analysis/utils/llm_utils.py

@@ -4,14 +4,32 @@ LLM API请求处理工具模块
 """
 
 import asyncio
-from typing import Any
 
 from ....utils.logger import logger
 from ....utils.resilience import CircuitBreaker, global_llm_rate_limiter
+from .structured_output_schema import JSONObject
 
 _circuit_breakers = {}
 
 
+def _is_response_format_unsupported_error(error: Exception) -> bool:
+    """
+    判断是否为 Provider/网关不支持 response_format 的兼容性错误。
+    """
+    text = str(error).lower()
+    patterns = [
+        "response_format",
+        "json_schema",
+        "unexpected keyword argument",
+        "extra fields not permitted",
+        "unknown field",
+        "not support",
+        "not supported",
+        "invalid request",
+    ]
+    return any(pattern in text for pattern in patterns)
+
+
 def _get_circuit_breaker(provider_id: str) -> CircuitBreaker:
     if provider_id not in _circuit_breakers:
         _circuit_breakers[provider_id] = CircuitBreaker(name=f"provider_{provider_id}")
@@ -184,12 +202,12 @@ async def call_provider_with_retry(
     context,
     config_manager,
     prompt: str,
-    max_tokens: int,
-    temperature: float,
     umo: str | None = None,
     provider_id_key: str | None = None,
     system_prompt: str | None = None,
-) -> Any | None:
+    response_format: JSONObject | None = None,
+    extra_generate_kwargs: dict[str, object] | None = None,
+) -> object | None:
     """
     调用LLM提供者，带超时、重试与退避。支持自定义服务商和配置化 Provider 选择。
 
@@ -197,11 +215,11 @@ async def call_provider_with_retry(
         context: AstrBot上下文对象
         config_manager: 配置管理器
         prompt: 输入的提示语
-        max_tokens: 最大生成token数
-        temperature: 采样温度
         umo: 指定使用的模型唯一标识符
         provider_id_key: 配置中的 provider_id 键名（如 'topic_provider_id'），用于选择特定的 Provider
         system_prompt: 系统提示词
+        response_format: 结构化输出约束（OpenAI 风格）
+        extra_generate_kwargs: 传递给 context.llm_generate 的附加参数（用于内部高级重试策略）
 
     Returns:
         LLM生成的结果，失败时返回None
@@ -225,7 +243,8 @@ async def call_provider_with_retry(
 
             logger.info(
                 f"[LLM 调用] 使用 Provider ID: {provider_id} | "
-                f"max_tokens={max_tokens} | temperature={temperature} | "
+                "max_tokens=provider-default | "
+                "temperature=provider-default | "
                 f"prompt长度={len(prompt) if prompt else 0}字符"
             )
 
@@ -255,13 +274,36 @@ async def call_provider_with_retry(
             # 超时由 Provider 内部控制，无需外层 wait_for
             try:
                 async with global_llm_rate_limiter:
-                    llm_resp = await context.llm_generate(
-                        chat_provider_id=provider_id,
-                        prompt=prompt,
-                        max_tokens=max_tokens,
-                        temperature=temperature,
-                        system_prompt=system_prompt,
-                    )
+                    llm_kwargs: dict[str, object] = {
+                        "chat_provider_id": provider_id,
+                        "prompt": prompt,
+                    }
+                    if system_prompt is not None:
+                        llm_kwargs["system_prompt"] = system_prompt
+                    if response_format is not None:
+                        llm_kwargs["response_format"] = response_format
+                    if extra_generate_kwargs:
+                        llm_kwargs.update(extra_generate_kwargs)
+
+                    try:
+                        llm_resp = await context.llm_generate(
+                            **llm_kwargs,
+                        )
+                    except Exception as e:
+                        if (
+                            response_format is not None
+                            and _is_response_format_unsupported_error(e)
+                        ):
+                            logger.warning(
+                                "[LLM 调用] 当前 Provider 可能不支持 response_format，"
+                                "已自动降级为无 schema 约束重试本次请求。"
+                            )
+                            llm_kwargs.pop("response_format", None)
+                            llm_resp = await context.llm_generate(
+                                **llm_kwargs,
+                            )
+                        else:
+                            raise
 
                 # 成功记录
                 cb.record_success()

src/infrastructure/analysis/utils/structured_output_schema.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+from typing import TypeAlias
+
+JSONScalar: TypeAlias = str | int | float | bool | None
+JSONValue: TypeAlias = JSONScalar | dict[str, "JSONValue"] | list["JSONValue"]
+JSONObject: TypeAlias = dict[str, JSONValue]
+
+
+def build_response_format(name: str, schema: JSONObject) -> JSONObject:
+    return {
+        "type": "json_schema",
+        "json_schema": {
+            "name": name,
+            "strict": True,
+            "schema": schema,
+        },
+    }
+
+
+def build_topics_schema(max_items: int) -> JSONObject:
+    return {
+        "type": "array",
+        "maxItems": max(1, int(max_items)),
+        "items": {
+            "type": "object",
+            "properties": {
+                "topic": {"type": "string"},
+                "contributors": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                },
+                "detail": {"type": "string"},
+            },
+            "required": ["topic", "contributors", "detail"],
+            "additionalProperties": False,
+        },
+    }
+
+
+def build_user_titles_schema(max_items: int) -> JSONObject:
+    return {
+        "type": "array",
+        "maxItems": max(1, int(max_items)),
+        "items": {
+            "type": "object",
+            "properties": {
+                "name": {"type": "string"},
+                "user_id": {"type": "string"},
+                "title": {"type": "string"},
+                "mbti": {"type": "string"},
+                "reason": {"type": "string"},
+            },
+            "required": ["name", "user_id", "title", "mbti", "reason"],
+            "additionalProperties": False,
+        },
+    }
+
+
+def build_golden_quotes_schema(max_items: int) -> JSONObject:
+    return {
+        "type": "array",
+        "maxItems": max(1, int(max_items)),
+        "items": {
+            "type": "object",
+            "properties": {
+                "content": {"type": "string"},
+                "sender": {"type": "string"},
+                "reason": {"type": "string"},
+            },
+            "required": ["content", "sender", "reason"],
+            "additionalProperties": False,
+        },
+    }
+
+
+def build_chat_quality_schema(max_dimensions: int) -> JSONObject:
+    return {
+        "type": "object",
+        "properties": {
+            "title": {"type": "string"},
+            "subtitle": {"type": "string"},
+            "dimensions": {
+                "type": "array",
+                "maxItems": max(1, int(max_dimensions)),
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "name": {"type": "string"},
+                        "percentage": {"type": "number"},
+                        "comment": {"type": "string"},
+                    },
+                    "required": ["name", "percentage", "comment"],
+                    "additionalProperties": False,
+                },
+            },
+            "summary": {"type": "string"},
+        },
+        "required": ["title", "subtitle", "dimensions", "summary"],
+        "additionalProperties": False,
+    }

src/infrastructure/config/config_manager.py

@@ -195,22 +195,6 @@ class ConfigManager:
         """获取LLM请求重试退避基值（秒），实际退避会乘以尝试次数"""
         return self._get_group("llm").get("llm_backoff", 2)
 
-    def get_topic_max_tokens(self) -> int:
-        """获取话题分析最大token数"""
-        return self._get_group("llm").get("topic_max_tokens", 12288)
-
-    def get_golden_quote_max_tokens(self) -> int:
-        """获取金句分析最大token数"""
-        return self._get_group("llm").get("golden_quote_max_tokens", 4096)
-
-    def get_user_title_max_tokens(self) -> int:
-        """获取用户称号分析最大token数"""
-        return self._get_group("llm").get("user_title_max_tokens", 4096)
-
-    def get_quality_max_tokens(self) -> int:
-        """获取聊天质量分析最大token数"""
-        return self._get_group("llm").get("quality_max_tokens", 4096)
-
     def get_debug_mode(self) -> bool:
         """获取是否启用调试模式"""
         return self._get_group("basic").get("debug_mode", False)