From e8c122e54264d79dbbe0cc339a705f7d33951f02 Mon Sep 17 00:00:00 2001 From: kx_guo Date: Fri, 4 Sep 2026 18:02:39 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E5=96=84=E9=A3=8E=E9=99=A9=E8=AF=84?= =?UTF-8?q?=E5=88=86=E3=80=81LLM=E5=85=9C=E5=BA=95=E4=B8=8E=E4=BC=9A?= =?UTF-8?q?=E8=AF=9D=E5=8E=86=E5=8F=B2=E5=8A=A0=E8=BD=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 接入数据库驱动的七维风险评分第一阶段逻辑\n- 增加主模型失败后的 DeepSeek 语义、流式和预防建议兜底\n- 将 AI 预防建议改为基于评分、历史异常和标准规则动态生成\n- 增加无历史案例时的标准库兜底校验和通用规则匹配\n- 修复未知供应商识别及测试占位名称过滤\n- 增加后端滚动日志文件配置和 LLM 请求日志\n- 历史会话直接读取已保存结果，避免重新调用模型和重算风险\n- 补充评分、供应商识别和 LLM 协议测试 --- .env.example | 33 ++ .gitignore | 2 +- app/api/v1/risk_predictions.py | 5 +- app/core/config.py | 46 ++- app/core/logging.py | 50 ++- app/repositories/conversations.py | 1 - app/schemas/risk_prediction.py | 12 +- app/services/conversations.py | 487 ++++++++++++++++++++++---- app/services/database_risk_scoring.py | 406 +++++++++++++++++++++ app/services/llm_client.py | 398 +++++++++++++++++---- app/services/risk_prediction.py | 63 +++- tests/test_conversations.py | 59 ++++ tests/test_database_risk_scoring.py | 65 ++++ tests/test_llm_protocol.py | 84 ++++- 14 files changed, 1549 insertions(+), 162 deletions(-) create mode 100644 app/services/database_risk_scoring.py create mode 100644 tests/test_database_risk_scoring.py diff --git a/.env.example b/.env.example index ef2d8a7..89e8730 100644 --- a/.env.example +++ b/.env.example @@ -1,19 +1,52 @@ +# ==================== 应用基础配置 ==================== +# 应用名称、版本和运行环境，用于接口文档、日志和运行行为识别。 RISK_APP_NAME=Dieya Risk Prediction Backend RISK_APP_VERSION=0.1.0 RISK_ENVIRONMENT=development RISK_DEBUG=false + +# ==================== 数据库配置 ==================== +# 风险系统业务数据库连接地址，包含数据库用户、密码、主机、端口和库名。 RISK_DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/dieya_risk_prediction + +# ==================== 日志配置 ==================== +# 控制后端日志级别；可选 DEBUG、INFO、WARNING、ERROR。 RISK_LOG_LEVEL=INFO + +# ==================== JWT 登录鉴权配置 ==================== +# JWT 签名密钥、签名算法和登录态有效期（分钟）。生产环境必须更换签名密钥。 RISK_JWT_SECRET_KEY=CHANGE_THIS_IN_PRODUCTION RISK_JWT_ALGORITHM=HS256 RISK_JWT_EXPIRE_MINUTES=480 + +# ==================== 钉钉免登配置 ==================== +# 用于钉钉免登和组织身份校验；未接入钉钉时可以留空。 RISK_DINGTALK_APP_KEY= RISK_DINGTALK_APP_SECRET= RISK_DINGTALK_CORP_ID= RISK_DINGTALK_AGENT_ID= + +# ==================== 主 LLM 模型配置 ==================== +# 主模型 API 密钥、OpenAI 兼容接口地址、模型名、请求超时时间和生成温度。 RISK_LLM_API_KEY= RISK_LLM_BASE_URL=https://api.openai.com/v1 RISK_LLM_MODEL=gpt-4o-mini RISK_LLM_TIMEOUT_SECONDS=120 RISK_LLM_TEMPERATURE=0.2 + +# ==================== LLM 兜底模型配置 ==================== +# 主模型请求失败时使用 DeepSeek；填写独立的 DeepSeek API Key 才会真正启用兜底。 +# API_MODE=responses 时调用 /responses；如服务商只支持 Chat Completions，可改为 chat_completions。 +# MAX_RETRIES 是兜底端点内部的重试次数，REASONING_EFFORT 适用于 Responses 推理配置。 +RISK_LLM_FALLBACK_API_KEY= +RISK_LLM_FALLBACK_BASE_URL=https://api.deepseek.com +RISK_LLM_FALLBACK_MODEL=deepseek-v4-pro +RISK_LLM_FALLBACK_API_MODE=responses +RISK_LLM_FALLBACK_TIMEOUT_SECONDS=30 +RISK_LLM_FALLBACK_MAX_RETRIES=1 +RISK_LLM_FALLBACK_REASONING_EFFORT=high +RISK_LLM_FALLBACK_MAX_OUTPUT_TOKENS=16384 + +# ==================== 跨域配置 ==================== +# 允许访问后端接口的前端地址，多个地址用英文逗号分隔。 RISK_CORS_ALLOW_ORIGINS=http://localhost:39158,http://127.0.0.1:39158 diff --git a/.gitignore b/.gitignore index 4e156bf..2398d05 100644 --- a/.gitignore +++ b/.gitignore @@ -6,8 +6,8 @@ __pycache__/ .coverage htmlcov/ .env +logs/ *.egg-info/ dist/ build/ data/ - diff --git a/app/api/v1/risk_predictions.py b/app/api/v1/risk_predictions.py index 5344a1b..19fc252 100644 --- a/app/api/v1/risk_predictions.py +++ b/app/api/v1/risk_predictions.py @@ -4,7 +4,7 @@ from typing import Annotated from fastapi import APIRouter, Depends -from app.api.dependencies import require_permission +from app.api.dependencies import DatabaseSession, require_permission from app.models.entities import UserAccount from app.schemas.common import ApiResponse from app.schemas.risk_prediction import ModelInfo, RiskAssessmentRequest, RiskAssessmentResult @@ -23,6 +23,7 @@ async def model_info( @router.post("/preview", response_model=ApiResponse[RiskAssessmentResult]) async def preview_risk( request: RiskAssessmentRequest, + session: DatabaseSession, _: Annotated[UserAccount, Depends(require_permission("risk.assess", operate=True))], ) -> ApiResponse[RiskAssessmentResult]: - return ApiResponse(data=RiskPredictionService.preview(request)) + return ApiResponse(data=await RiskPredictionService.preview(request, session)) diff --git a/app/core/config.py b/app/core/config.py index 6bd7f70..2dbf7f8 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -1,7 +1,7 @@ """Environment-backed application settings.""" from functools import lru_cache -from typing import Annotated +from typing import Annotated, Literal from pydantic import AliasChoices, Field, SecretStr, field_validator from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict @@ -53,6 +53,45 @@ class Settings(BaseSettings): le=2, validation_alias=AliasChoices("RISK_LLM_TEMPERATURE", "LLM_TEMPERATURE"), ) + llm_fallback_api_key: SecretStr | None = Field( + default=None, + repr=False, + validation_alias=AliasChoices("RISK_LLM_FALLBACK_API_KEY", "DEEPSEEK_API_KEY"), + ) + llm_fallback_base_url: str = Field( + default="https://api.deepseek.com", + validation_alias=AliasChoices("RISK_LLM_FALLBACK_BASE_URL", "DEEPSEEK_BASE_URL"), + ) + llm_fallback_model: str = Field( + default="deepseek-v4-pro", + validation_alias=AliasChoices("RISK_LLM_FALLBACK_MODEL", "DEEPSEEK_MODEL"), + ) + llm_fallback_api_mode: Literal["chat_completions", "responses"] = Field( + default="responses", + validation_alias=AliasChoices("RISK_LLM_FALLBACK_API_MODE", "DEEPSEEK_API_MODE"), + ) + llm_fallback_timeout_seconds: int = Field( + default=30, + gt=0, + le=600, + validation_alias=AliasChoices("RISK_LLM_FALLBACK_TIMEOUT_SECONDS", "DEEPSEEK_TIMEOUT"), + ) + llm_fallback_max_retries: int = Field( + default=1, + ge=0, + le=10, + validation_alias=AliasChoices("RISK_LLM_FALLBACK_MAX_RETRIES", "DEEPSEEK_MAX_RETRIES"), + ) + llm_fallback_reasoning_effort: Literal["low", "high", "max"] = Field( + default="high", + validation_alias=AliasChoices("RISK_LLM_FALLBACK_REASONING_EFFORT", "DEEPSEEK_REASONING_EFFORT"), + ) + llm_fallback_max_output_tokens: int = Field( + default=16_384, + ge=1, + le=393_216, + validation_alias=AliasChoices("RISK_LLM_FALLBACK_MAX_OUTPUT_TOKENS", "DEEPSEEK_MAX_OUTPUT_TOKENS"), + ) cors_allow_origins: Annotated[tuple[str, ...], NoDecode] = ( "http://localhost:39158", "http://127.0.0.1:39158", @@ -77,6 +116,11 @@ class Settings(BaseSettings): def normalize_llm_base_url(cls, value: str) -> str: return value.rstrip("/") + @field_validator("llm_fallback_base_url") + @classmethod + def normalize_llm_fallback_base_url(cls, value: str) -> str: + return value.rstrip("/") + @field_validator("cors_allow_origins", mode="before") @classmethod def split_cors_origins(cls, value: object) -> object: diff --git a/app/core/logging.py b/app/core/logging.py index ec2a5e1..0c280eb 100644 --- a/app/core/logging.py +++ b/app/core/logging.py @@ -1,20 +1,60 @@ """Shared logging setup.""" import logging +from logging.handlers import RotatingFileHandler +from pathlib import Path LOGGER_NAME = "dieya_risk_prediction" +LOG_DIR = Path(__file__).resolve().parents[2] / "logs" +LOG_FILE = LOG_DIR / "backend.log" +_FORMAT = "%(asctime)s %(levelname)s %(name)s %(message)s" def configure_logging(level: str = "INFO") -> logging.Logger: logger = logging.getLogger(LOGGER_NAME) logger.setLevel(level.upper()) logger.propagate = False - if not logger.handlers: - handler = logging.StreamHandler() - handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s")) - logger.addHandler(handler) + LOG_DIR.mkdir(parents=True, exist_ok=True) + formatter = logging.Formatter(_FORMAT) + file_handler = next( + ( + handler + for handler in logger.handlers + if getattr(handler, "_dieya_file_handler", False) + ), + None, + ) + if file_handler is None: + file_handler = RotatingFileHandler( + LOG_FILE, + maxBytes=10 * 1024 * 1024, + backupCount=5, + encoding="utf-8", + ) + file_handler._dieya_file_handler = True + file_handler.setFormatter(formatter) + logger.addHandler(file_handler) + root_logger = logging.getLogger() + root_logger.setLevel(level.upper()) + if not any(getattr(handler, "_dieya_file_handler", False) for handler in root_logger.handlers): + root_file_handler = RotatingFileHandler( + LOG_FILE, + maxBytes=10 * 1024 * 1024, + backupCount=5, + encoding="utf-8", + ) + root_file_handler._dieya_file_handler = True + root_file_handler.setFormatter(formatter) + root_logger.addHandler(root_file_handler) + if not any( + isinstance(handler, logging.StreamHandler) + and not getattr(handler, "_dieya_file_handler", False) + for handler in logger.handlers + ): + stream_handler = logging.StreamHandler() + stream_handler.setFormatter(formatter) + logger.addHandler(stream_handler) return logger logger = configure_logging() - diff --git a/app/repositories/conversations.py b/app/repositories/conversations.py index 0f8299e..757d4d0 100644 --- a/app/repositories/conversations.py +++ b/app/repositories/conversations.py @@ -44,7 +44,6 @@ class ConversationRepository: Conversation.id == conversation_id, Conversation.user_id == user_id, ) - .options(selectinload(Conversation.messages)) ) if conversation is None: return None diff --git a/app/schemas/risk_prediction.py b/app/schemas/risk_prediction.py index 528cc72..a02dad4 100644 --- a/app/schemas/risk_prediction.py +++ b/app/schemas/risk_prediction.py @@ -14,7 +14,13 @@ class RiskIndicator(ApiSchema): class RiskAssessmentRequest(ApiSchema): subject_id: str = Field(min_length=1, max_length=128) - indicators: list[RiskIndicator] = Field(min_length=1, max_length=50) + indicators: list[RiskIndicator] = Field(default_factory=list, max_length=50) + product_category: str | None = Field(default=None, max_length=100) + material_type: str | None = Field(default=None, max_length=200) + supplier: str | None = Field(default=None, max_length=300) + factory: str | None = Field(default=None, max_length=300) + stage_new: str | None = Field(default=None, max_length=100) + defect_level: str | None = Field(default=None, max_length=10) class RiskFactorResult(ApiSchema): @@ -54,6 +60,10 @@ class RiskDimensionScore(ApiSchema): dimension: str score: float weight: float + status: str = "database" + reason: str | None = None + sample_size: int | None = None + raw_value: float | int | None = None class SimilarCase(ApiSchema): diff --git a/app/services/conversations.py b/app/services/conversations.py index 454bc75..03b57db 100644 --- a/app/services/conversations.py +++ b/app/services/conversations.py @@ -1,6 +1,7 @@ """Conversation use cases and the initial risk-assistant response adapter.""" import json +import logging import re from collections.abc import AsyncGenerator, Sequence from datetime import UTC, datetime @@ -25,8 +26,11 @@ from app.schemas.risk_prediction import ( StandardJudgment, SupplierFrequency, ) +from app.services.database_risk_scoring import DatabaseRiskScoringService from app.services.llm_client import LLMClient +logger = logging.getLogger("dieya_risk_prediction.conversations") + class ConversationService: _engine = BaselineRiskEngine() @@ -93,13 +97,11 @@ class ConversationService: if page is None: raise AppException("对话不存在或无权访问", code=404, status_code=404) conversation, messages, total = page - live_results = await self._refresh_risk_results(messages) return self._data( conversation, messages=messages, total=total, offset=offset, - risk_results=live_results, ) async def create(self, user: UserAccount, title: str | None = None) -> ConversationData: @@ -160,13 +162,11 @@ class ConversationService: if page is None: # pragma: no cover - the row already existed raise AppException("对话读取失败", code=500, status_code=500) persisted, messages, total = page - live_results = await self._refresh_risk_results(messages) return self._data( persisted, messages=messages, total=total, offset=0, - risk_results=live_results, ) async def stream( @@ -208,6 +208,11 @@ class ConversationService: yield {"type": "error", "message": str(exc)} return semantics = self._normalize_semantics(text, semantics) + semantics = await self._normalize_follow_up_semantics( + text, + semantics, + conversation, + ) yield {"type": "semantic", "data": semantics} context = self._llm_messages(conversation, text, semantics) @@ -228,6 +233,11 @@ class ConversationService: show_risk = self._is_risk_request(text, semantics) response_type = self._response_type(text, semantics) + response_type = await self._use_solution_card_without_history( + response_type, + text, + semantics, + ) risk_result = ( await self._enrich_risk_result(text, semantics, self._predict(text)) if show_risk and response_type == "risk_assessment" @@ -265,13 +275,11 @@ class ConversationService: yield {"type": "error", "message": "对话读取失败"} return persisted, messages, total = page - live_results = await self._refresh_risk_results(messages) result = self._data( persisted, messages=messages, total=total, offset=0, - risk_results=live_results, ) yield {"type": "done", "data": result.model_dump(mode="json", by_alias=True)} @@ -309,6 +317,8 @@ class ConversationService: def _response_type(cls, text: str, semantics: dict[str, Any]) -> str: """Choose the prototype card from the user's actual question.""" + if cls._is_placeholder_supplier_name(text): + return "general" intent = str(semantics.get("intent") or "").strip().lower() if intent == "supplier" or ( any(word in text for word in ("供应商", "加工厂", "工厂")) @@ -322,15 +332,138 @@ class ConversationService: return "standard" if intent == "compliance" or any(word in text for word in ("合规", "宣称", "备案", "法规")): return "compliance" - if intent == "solution" or ( - any(word in text for word in ("次氯酸", "消毒喷雾")) - and not any(word in text for word in ("评估", "风险评估", "风险预估")) - ): + if intent == "solution": return "solution" if intent == "assessment" or cls._is_explicit_risk_request(text): return "risk_assessment" return "general" + async def _normalize_follow_up_semantics( + self, + text: str, + semantics: dict[str, Any], + conversation: Conversation, + ) -> dict[str, Any]: + """Recover card intent for short follow-ups such as ``佳亿``. + + The LLM can answer a short follow-up using conversation history while + still labeling the current turn as ``general``. Card routing needs a + deterministic signal, so known supplier names are matched locally and + written back into the semantic payload before routing. + """ + + normalized = dict(semantics) + if self._is_placeholder_supplier_name(text): + normalized.update( + { + "intent": "general", + "risk_requested": False, + "semantic_source": "placeholder_filtered", + } + ) + normalized.pop("supplier", None) + return normalized + intent = str(normalized.get("intent") or "").strip().lower() + if intent not in {"", "general", "greet"}: + return normalized + supplier_names = list( + ( + await self.session.scalars( + select(AnomalyCase.supplier) + .where(AnomalyCase.supplier.is_not(None)) + .distinct() + ) + ).all() + ) + supplier_name = self._infer_supplier_follow_up(text, supplier_names) + supplier_name = supplier_name or self._infer_unknown_supplier_follow_up(text) + if not supplier_name: + return normalized + normalized.update( + { + "intent": "supplier", + "risk_requested": True, + "supplier": supplier_name, + "confidence": max(float(normalized.get("confidence") or 0), 0.75), + "semantic_source": "database_supplier_match" + if supplier_name in supplier_names + else "standalone_new_supplier_fallback", + } + ) + return normalized + + @staticmethod + def _infer_supplier_follow_up(text: str, supplier_names: Sequence[str | None]) -> str | None: + """Return a known supplier when the current message names it directly.""" + + cleaned = text.strip().lower() + if not cleaned: + return None + candidates = sorted( + { + " ".join(name.strip().split()) + for name in supplier_names + if name and name.strip() and name.strip() not in {"/", "-"} + }, + key=len, + reverse=True, + ) + for name in candidates: + if cleaned == name.lower(): + return name + return None + + @staticmethod + def _infer_unknown_supplier_follow_up(text: str) -> str | None: + """Treat a short standalone business name as a new supplier. + + This is deliberately conservative: common greetings, product terms, + workflow commands, and messages containing punctuation are excluded. + Unknown names are shown as a new supplier with zero historical data. + """ + + cleaned = " ".join(text.strip().split()) + if not 2 <= len(cleaned) <= 20 or " " in cleaned: + return None + if not re.fullmatch(r"[\u4e00-\u9fffA-Za-z][\u4e00-\u9fffA-Za-z0-9_-]*", cleaned): + return None + excluded = { + "你好", + "您好", + "谢谢", + "好的", + "继续", + "继续分析", + "洗发水", + "沐浴露", + "面霜", + "脱毛膏", + "清洁剂", + "牙膏", + "软管", + "泵头", + "瓶子", + "彩盒", + "标准库", + "风险评估", + "供应商", + "新供应商A", + "测试供应商", + } + if ( + cleaned in excluded + or cleaned.startswith("测试") + or cleaned.startswith("新供应商") + or any(term in cleaned for term in ("请", "帮我", "分析", "查询", "怎么样")) + ): + return None + return cleaned + + @staticmethod + def _is_placeholder_supplier_name(text: str) -> bool: + cleaned = " ".join(text.strip().split()) + return cleaned.startswith("测试") or cleaned.startswith("新供应商") + async def _build_response_card( self, response_type: str, @@ -347,6 +480,59 @@ class ConversationService: builder = builders.get(response_type) return await builder(text, semantics) if builder else None + async def _use_solution_card_without_history( + self, + response_type: str, + text: str, + semantics: dict[str, Any], + ) -> str: + """Use the solution card for a product risk question with no history.""" + + if response_type != "risk_assessment": + return response_type + category = self._semantic_text(semantics, "category") or self._first_match( + text, + ( + "洗发水", + "沐浴露", + "面霜", + "乳液", + "精华", + "面膜", + "防晒", + "护手霜", + "护发素", + "脱毛膏", + "清洁剂", + "牙膏", + "消毒喷雾", + "次氯酸", + ), + ) + has_product_context = bool( + category + or any( + key in text + for key in ( + "新品", + "新产品", + "产品开发", + "产品立项", + "规格", + "容量", + "包装", + "材质", + "配方", + "原料", + ) + ) + ) + if not has_product_context: + return response_type + if not category or await self._count_category_anomalies(category) == 0: + return "solution" + return response_type + async def _supplier_card(self, text: str, semantics: dict[str, Any]) -> dict[str, Any]: name = self._semantic_text(semantics, "supplier") or self._semantic_text(semantics, "factory") if not name: @@ -451,10 +637,29 @@ class ConversationService: async def _solution_card(self, text: str, semantics: dict[str, Any]) -> dict[str, Any]: category = self._semantic_text(semantics, "category") or self._first_match( - text, ("次氯酸消毒喷雾", "消毒喷雾", "清洁剂", "脱毛膏") + text, + ( + "次氯酸消毒喷雾", + "消毒喷雾", + "清洁剂", + "脱毛膏", + "洗发水", + "沐浴露", + "面霜", + "乳液", + "精华", + "面膜", + "防晒", + "护手霜", + "护发素", + "牙膏", + ), ) or "当前新品" direct_count = await self._count_category_anomalies(category) standards = await self._standards_for(category, text) + standard_fallback_enabled = direct_count == 0 + if standard_fallback_enabled and not standards: + standards = await self._fallback_standards_for(text) risks: list[dict[str, str]] = [] if "次氯酸" in text or "铝" in text: risks.append( @@ -480,6 +685,31 @@ class ConversationService: "detail": "供应商或加工厂的历史表现需要结合异常案例和现场审核确认。", } ) + fallback_suggestions = [ + "小样阶段完成内容物与包装材料相容性测试", + "增加倒置密封测试和加速老化测试", + "首单安排来料全检，并根据结果决定是否驻厂巡检", + "实际执行前由品控确认并安排小批量试产验证", + ] + try: + suggestions = await self.llm.generate_prevention_suggestions( + { + "user_input": text, + "category": category, + "historical_case_count": direct_count, + "standard_fallback_enabled": standard_fallback_enabled, + "matched_standards": [self._standard_dict(item) for item in standards[:5]], + "identified_risks": risks, + "rag_similar_cases": [], + } + ) + except Exception as exc: # noqa: BLE001 - the solution card must remain available + logger.warning( + "Solution card AI suggestions failed error=%s; using deterministic fallback", + exc.__class__.__name__, + ) + suggestions = fallback_suggestions + return { "type": "solution", "title": "AI 案例推理生成", @@ -488,19 +718,76 @@ class ConversationService: f"目标品类「{category}」当前命中历史异常 {direct_count} 条；" "方案结合已导入异常案例和标准规则生成。" if direct_count - else f"目标品类「{category}」暂无直接历史案例，当前按标准规则库进行兜底分析。" + else f"目标品类「{category}」暂无直接历史案例，当前按技术法规规则库进行兜底分析。" ), + "standardFallbackEnabled": standard_fallback_enabled, "standards": [self._standard_dict(item) for item in standards[:4]], "risks": risks, - "suggestions": [ - "小样阶段完成内容物与包装材料相容性测试", - "增加倒置密封测试和加速老化测试", - "首单安排来料全检，并根据结果决定是否驻厂巡检", - "实际执行前由品控确认并安排小批量试产验证", - ], + "suggestions": suggestions, "warning": "以上方案由 AI 推理生成，实际执行前请咨询品控部门确认。", } + async def _fallback_standards_for(self, text: str) -> Sequence[StandardJudgment]: + """Find generic safety rules when no direct product history is available.""" + + terms = [ + term + for term in ( + self._first_match(text, ("相容性", "密封", "稳定性", "消毒", "微生物", "氧化")), + self._first_match(text, ("铝", "金属", "喷头", "软管", "PET", "泵头")), + ) + if term + ] + if terms: + clauses = [ + or_( + StandardRule.category_lv1.ilike(f"%{term}%"), + StandardRule.category_lv2.ilike(f"%{term}%"), + StandardRule.category_lv3.ilike(f"%{term}%"), + StandardRule.risk_point.ilike(f"%{term}%"), + StandardRule.judge_standard.ilike(f"%{term}%"), + ) + for term in terms + ] + rows = list( + await self.session.scalars( + select(StandardRule) + .where(StandardRule.status == "启用", or_(*clauses)) + .order_by(StandardRule.id) + .limit(4) + ) + ) + if rows: + return [ + StandardJudgment( + category=row.category_lv1, + risk_point=row.risk_point, + rule=row.judge_standard, + source=row.regulation or "标准文件.xlsx", + status="待检测", + ) + for row in rows + ] + + rows = list( + await self.session.scalars( + select(StandardRule) + .where(StandardRule.status == "启用") + .order_by(StandardRule.id) + .limit(4) + ) + ) + return [ + StandardJudgment( + category=row.category_lv1, + risk_point=row.risk_point, + rule=row.judge_standard, + source=row.regulation or "标准文件.xlsx", + status="待检测", + ) + for row in rows + ] + async def _top_anomaly_types(self, column, value: str | None) -> Sequence[dict[str, Any]]: # noqa: ANN001 if not value: return [] @@ -603,20 +890,29 @@ class ConversationService: supplier = None supplier_or_factory = supplier or factory + database_score = await DatabaseRiskScoringService(self.session).score( + category=category, + material=self._first_match( + text, + ("软管", "PET瓶", "PET", "铝瓶", "泵头", "彩盒", "瓶子", "喷头", "盖子"), + ), + supplier=supplier, + factory=factory, + stage=self._semantic_text(semantics, "stage") or self._first_match( + text, ("需求发现", "产品立项", "产品开发", "量产上市", "产品生命管理") + ), + defect_level=self._first_match(text, ("A级", "B级", "C级")), + ) category_count = await self._count_category_anomalies(category) - supplier_count = await self._count_anomalies(AnomalyCase.supplier, supplier_or_factory) + supplier_count = await self._count_anomalies(AnomalyCase.supplier, supplier) factory_count = await self._count_anomalies(AnomalyCase.factory, factory) - if factory and not supplier: - anomaly_count = factory_count - elif supplier and not factory: - anomaly_count = supplier_count - else: - anomaly_count = supplier_count or factory_count + anomaly_count = supplier_count if supplier else factory_count standard_rows = await self._standards_for(category, text) - alpha = round(min(1.5, 1 + (anomaly_count / 100) * 0.3), 2) if anomaly_count else 1.0 - tier = self._supplier_tier(anomaly_count) - level_text = self._risk_level_text(result.level) + alpha_count = supplier_count if supplier else 0 + alpha = round(min(1.5, 1 + (alpha_count / 100) * 0.3), 2) if alpha_count else 1.0 + tier = self._supplier_tier(alpha_count) if supplier else None + level_text = self._risk_level_text(database_score.level) alert_text = self._alert_text( category, category_count, @@ -626,17 +922,45 @@ class ConversationService: ) dimensions = [ RiskDimensionScore( - dimension=self._dimension_name(factor.name), - score=round(max(0, 100 - factor.value), 2), - weight=factor.weight, + dimension=item.name, + score=item.score, + weight=item.weight, + status=item.status, + reason=item.reason, + sample_size=item.sample_size, + raw_value=item.raw_value, ) - for factor in result.factors + for item in database_score.dimensions ] - suggestions = self._suggestions(text, supplier_or_factory) + factors = [ + RiskFactorResult( + name=item.name, + value=round(100 - item.score, 2), + weight=item.weight, + contribution=round(item.score * item.weight / 100, 2), + ) + for item in database_score.dimensions + ] + suggestions = await self._generate_ai_suggestions( + text=text, + semantics=semantics, + category=category, + supplier_or_factory=supplier_or_factory, + category_count=category_count, + anomaly_count=anomaly_count, + alpha=alpha, + database_score=database_score, + dimensions=dimensions, + standard_rows=standard_rows, + ) return result.model_copy( update={ - "risk_badge": self._risk_badge(result.level), + "score": database_score.score, + "level": database_score.level, + "model_version": "database-seven-dimension-v1", + "factors": factors, + "risk_badge": self._risk_badge(database_score.level), "alert_text": alert_text, "category_warning": CategoryWarning( category=category, @@ -675,6 +999,62 @@ class ConversationService: ) return int(count or 0) + async def _generate_ai_suggestions( + self, + *, + text: str, + semantics: dict[str, Any], + category: str, + supplier_or_factory: str | None, + category_count: int, + anomaly_count: int, + alpha: float, + database_score, + dimensions: Sequence[RiskDimensionScore], + standard_rows: Sequence[StandardJudgment], + ) -> Sequence[str]: # noqa: ANN001 + """Generate suggestions from live scoring data, with a deterministic fallback.""" + + context = { + "user_input": text, + "semantic_fields": { + key: value + for key, value in semantics.items() + if key in {"category", "specification", "packaging", "supplier", "factory", "stage"} + and value + }, + "category": category, + "supplier_or_factory": supplier_or_factory, + "category_anomaly_count": category_count, + "supplier_or_factory_anomaly_count": anomaly_count, + "frequency_alpha": alpha, + "overall_score": database_score.score, + "risk_level": self._risk_level_text(database_score.level), + "dimensions": [ + { + "name": item.dimension, + "score": item.score, + "weight": item.weight, + "status": item.status, + "reason": item.reason, + "sample_size": item.sample_size, + } + for item in dimensions + ], + "matched_standards": [self._standard_dict(item) for item in standard_rows[:5]], + "rag_similar_cases": [], + } + try: + suggestions = await self.llm.generate_prevention_suggestions(context) + logger.info("AI prevention suggestions generated count=%s", len(suggestions)) + return suggestions + except Exception as exc: # noqa: BLE001 - suggestions must not break risk assessment + logger.warning( + "AI prevention suggestion generation failed error=%s; using deterministic fallback", + exc.__class__.__name__, + ) + return self._suggestions(text, supplier_or_factory) + async def _count_category_anomalies(self, category: str) -> int: """The imported workbook has sparse product_category values. @@ -878,9 +1258,7 @@ class ConversationService: id=conversation.id, title=conversation.title, message_count=resolved_total, - last_risk_level=ConversationService._last_risk_level_from_messages( - resolved_messages, - ) or ConversationService._last_risk_level(conversation), + last_risk_level=ConversationService._last_risk_level_from_messages(resolved_messages), updated_at=conversation.updated_at, messages=[ cls._message( @@ -898,41 +1276,6 @@ class ConversationService: has_more_messages=offset + len(resolved_messages) < resolved_total, ) - async def _refresh_risk_results( - self, - messages: Sequence[ConversationMessage], - ) -> dict[int, RiskAssessmentResult]: - """Refresh database-backed card fields for historical messages. - - Risk scores are intentionally kept as the persisted assessment result, - while fields sourced from imported cases and standards are re-read from - PostgreSQL. This prevents an old conversation from showing stale zero - counts after the reference workbooks have been imported. - """ - - refreshed: dict[int, RiskAssessmentResult] = {} - for index, message in enumerate(messages): - if message.role != "assistant": - continue - result = self._risk_result_for_message(messages, index) - if result is None: - continue - user_text = next( - ( - candidate.content - for candidate in reversed(messages[:index]) - if candidate.role == "user" - ), - result.subject_id, - ) - semantic = (message.structured_data or {}).get("semantic") - refreshed[message.id] = await self._enrich_risk_result( - user_text, - semantic if isinstance(semantic, dict) else {}, - result, - ) - return refreshed - @staticmethod def _message( message: ConversationMessage, diff --git a/app/services/database_risk_scoring.py b/app/services/database_risk_scoring.py new file mode 100644 index 0000000..0c36fb9 --- /dev/null +++ b/app/services/database_risk_scoring.py @@ -0,0 +1,406 @@ +"""First-phase database-backed seven-dimension risk scoring.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date, timedelta +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.domain.risk_engine import RiskLevel +from app.models.entities import AnomalyCase + +DEFAULT_WEIGHTS = { + "供应商表现": 25.0, + "物料品类": 20.0, + "产品品类": 15.0, + "加工厂质量": 15.0, + "标准符合性": 10.0, + "节点阶段": 10.0, + "缺陷等级": 5.0, +} + +STAGES = ("需求发现", "产品立项", "产品开发", "量产上市", "产品生命管理") + + +@dataclass(frozen=True, slots=True) +class DimensionScore: + name: str + score: float + weight: float + status: str + reason: str | None = None + sample_size: int | None = None + raw_value: float | int | None = None + + +@dataclass(frozen=True, slots=True) +class DatabaseRiskScore: + score: float + level: RiskLevel + dimensions: tuple[DimensionScore, ...] + + +class DatabaseRiskScoringService: + """Calculate the PRD first-phase model from anomaly_cases. + + Inspection results and standard test results are intentionally not guessed + here. Their dimensions use the PRD neutral score of 50 and expose the + reason so callers can distinguish a real score from a fallback. + """ + + def __init__(self, session: AsyncSession, *, as_of: date | None = None) -> None: + self.session = session + self.as_of = as_of or date.today() + + async def score( + self, + *, + category: str | None = None, + material: str | None = None, + supplier: str | None = None, + factory: str | None = None, + stage: str | None = None, + defect_level: str | None = None, + ) -> DatabaseRiskScore: + rows = await self._load_rows() + category_name = normalize_category(category) + material_name = normalize_material(material) + stage_name = normalize_stage(stage) + party_name = clean_value(supplier) + + dimensions = ( + self._supplier_dimension(rows, party_name), + self._material_dimension(rows, material_name), + self._category_dimension(rows, category_name), + DimensionScore( + name="加工厂质量", + score=50.0, + weight=DEFAULT_WEIGHTS["加工厂质量"], + status="default", + reason="当前数据库没有巡检合格率/批次数据，按 PRD 中性值 50 分计入", + ), + DimensionScore( + name="标准符合性", + score=50.0, + weight=DEFAULT_WEIGHTS["标准符合性"], + status="default", + reason="当前数据库没有检测结果与标准判定记录，按 PRD 中性值 50 分计入", + ), + self._stage_dimension(rows, stage_name), + self._defect_dimension(rows, category_name, material_name, party_name, defect_level), + ) + score = round(sum(item.score * item.weight / 100 for item in dimensions), 2) + return DatabaseRiskScore(score=score, level=classify_score(score), dimensions=dimensions) + + async def _load_rows(self) -> list[dict[str, Any]]: + start = self.as_of - timedelta(days=365 * 3) + statement = select( + AnomalyCase.date, + AnomalyCase.product_name, + AnomalyCase.product_category, + AnomalyCase.material_type, + AnomalyCase.supplier, + AnomalyCase.factory, + AnomalyCase.stage_new, + AnomalyCase.defect_level, + ).where( + AnomalyCase.date.is_not(None), + AnomalyCase.date >= start, + AnomalyCase.date <= self.as_of, + ) + return [dict(row) for row in (await self.session.execute(statement)).mappings().all()] + + def _supplier_dimension(self, rows: list[dict[str, Any]], party_name: str | None) -> DimensionScore: + weight = DEFAULT_WEIGHTS["供应商表现"] + if not party_name: + return DimensionScore( + name="供应商表现", + score=50.0, + weight=weight, + status="default", + reason="未提供供应商/加工厂，按 PRD 缺失维度默认 50 分", + ) + matched = [row for row in rows if party_matches(row, party_name)] + count = len(matched) + if count == 0: + base = 70.0 + elif count <= 20: + base = 90.0 + elif count <= 60: + base = 75.0 + elif count <= 100: + base = 65.0 + elif count <= 150: + base = 45.0 + else: + base = 30.0 + a_count = sum(1 for row in matched if normalize_defect(row.get("defect_level")) == "A") + base -= min(15, a_count * 5) + current_start = self.as_of - timedelta(days=183) + previous_start = current_start - timedelta(days=183) + current = sum(1 for row in matched if row.get("date") and row["date"] >= current_start) + previous = sum( + 1 + for row in matched + if row.get("date") and previous_start <= row["date"] < current_start + ) + trend_adjustment = ( + -10 + if current > previous * 1.2 and current >= 3 + else 5 + if previous > current * 1.2 + else 0 + ) + base = max(0.0, min(100.0, base + trend_adjustment)) + alpha = min(1.5, 1.0 + count / 100 * 0.3) + score = min(100.0, round(base * alpha, 2)) + return DimensionScore( + name="供应商表现", + score=score, + weight=weight, + status="database", + reason=f"近 3 年异常 {count} 条，基础分 {base:g}，α={alpha:.2f}", + sample_size=count, + raw_value=count, + ) + + @staticmethod + def _material_dimension(rows: list[dict[str, Any]], material: str | None) -> DimensionScore: + weight = DEFAULT_WEIGHTS["物料品类"] + if not material: + return DimensionScore( + name="物料品类", + score=50.0, + weight=weight, + status="default", + reason="未提供包装材质，按 PRD 缺失维度默认 50 分", + ) + counts: dict[str, int] = {} + for row in rows: + normalized = normalize_material(row.get("material_type")) + if normalized: + counts[normalized] = counts.get(normalized, 0) + 1 + count = counts.get(material, 0) + maximum = max(counts.values(), default=0) + score = 100.0 if maximum == 0 else round(100 - count / maximum * 100, 2) + return DimensionScore( + name="物料品类", + score=score, + weight=weight, + status="database", + reason=f"物料「{material}」异常 {count} 条，相对最大物料异常 {maximum} 条归一化", + sample_size=count, + raw_value=count, + ) + + @staticmethod + def _category_dimension(rows: list[dict[str, Any]], category: str | None) -> DimensionScore: + weight = DEFAULT_WEIGHTS["产品品类"] + if not category: + return DimensionScore( + name="产品品类", + score=50.0, + weight=weight, + status="default", + reason="无法从产品名称归类，按 PRD 缺失维度默认 50 分", + ) + counts: dict[str, int] = {} + for row in rows: + normalized = normalize_category(row.get("product_category")) or normalize_category(row.get("product_name")) + if normalized: + counts[normalized] = counts.get(normalized, 0) + 1 + count = counts.get(category, 0) + maximum = max(counts.values(), default=0) + score = 45.0 if count < 10 else round(100 - count / maximum * 100, 2) if maximum else 45.0 + return DimensionScore( + name="产品品类", + score=score, + weight=weight, + status="database", + reason=( + f"品类「{category}」异常 {count} 条" + + ( + "，样本少于 10 条按 45 分处理" + if count < 10 + else "，按品类最大异常数归一化" + ) + ), + sample_size=count, + raw_value=count, + ) + + @staticmethod + def _stage_dimension(rows: list[dict[str, Any]], stage: str | None) -> DimensionScore: + weight = DEFAULT_WEIGHTS["节点阶段"] + if not stage: + return DimensionScore( + name="节点阶段", + score=50.0, + weight=weight, + status="default", + reason="未提供产品阶段，按 PRD 缺失维度默认 50 分", + ) + total = len(rows) + count = sum(1 for row in rows if normalize_stage(row.get("stage_new")) == stage) + share = count / total if total else 0 + if share > 0.5: + score = 50.0 + elif share >= 0.3: + score = 60.0 + elif share >= 0.1: + score = 75.0 + elif share >= 0.01: + score = 80.0 + else: + score = 90.0 + return DimensionScore( + name="节点阶段", + score=score, + weight=weight, + status="database", + reason=f"阶段「{stage}」异常占比 {share:.1%}，按阶段档位取分", + sample_size=count, + raw_value=share * 100, + ) + + @staticmethod + def _defect_dimension( + rows: list[dict[str, Any]], + category: str | None, + material: str | None, + party_name: str | None, + defect_level: str | None, + ) -> DimensionScore: + weight = DEFAULT_WEIGHTS["缺陷等级"] + if not category and not material and not party_name and not defect_level: + return DimensionScore( + name="缺陷等级", + score=50.0, + weight=weight, + status="default", + reason="缺少品类、物料、供应商和缺陷等级组合，按 PRD 缺失维度默认 50 分", + ) + matched = [ + row + for row in rows + if ( + not category + or normalize_category(row.get("product_category")) == category + or normalize_category(row.get("product_name")) == category + ) + and (not material or normalize_material(row.get("material_type")) == material) + and (not party_name or party_matches(row, party_name)) + ] + counts = { + level: sum( + 1 + for row in matched + if normalize_defect(row.get("defect_level")) == level + ) + for level in ("A", "B", "C") + } + labeled = sum(counts.values()) + if not labeled: + return DimensionScore( + name="缺陷等级", + score=50.0, + weight=weight, + status="default", + reason="当前组合没有可用的 A/B/C 缺陷记录，按 PRD 缺失维度默认 50 分", + sample_size=0, + ) + penalty = counts["A"] * 40 + counts["B"] * 15 + counts["C"] * 5 + score = max(0.0, 100.0 - penalty) + return DimensionScore( + name="缺陷等级", + score=score, + weight=weight, + status="database", + reason=f"组合缺陷记录 A={counts['A']}、B={counts['B']}、C={counts['C']}，按严重度扣分", + sample_size=labeled, + raw_value=penalty, + ) + + +def clean_value(value: object) -> str | None: + if value is None: + return None + cleaned = " ".join(str(value).strip().split()) + return cleaned if cleaned and cleaned not in {"/", "-", "未填写", "未归类"} else None + + +def normalize_defect(value: object) -> str | None: + cleaned = clean_value(value) + return cleaned.upper() if cleaned and cleaned.upper() in {"A", "B", "C"} else None + + +def normalize_stage(value: object) -> str | None: + cleaned = clean_value(value) + if not cleaned: + return None + for stage in STAGES: + if stage in cleaned: + return stage + return None + + +def normalize_material(value: object) -> str | None: + cleaned = clean_value(value) + if not cleaned: + return None + aliases = ( + ("软管", "软管"), + ("泵头", "泵头"), + ("彩盒", "彩盒"), + ("标贴", "标贴"), + ("标签", "标贴"), + ("收缩膜", "收缩膜"), + ("瓶", "瓶子"), + ("PET", "瓶子"), + ("亚克力", "瓶子"), + ("盖", "盖子"), + ) + for keyword, normalized in aliases: + if keyword.lower() in cleaned.lower(): + return normalized + return cleaned + + +def normalize_category(value: object) -> str | None: + cleaned = clean_value(value) + if not cleaned: + return None + aliases = ( + (("洗发", "洗护", "洗发水", "洗发露", "洗发乳"), "洗发水/洗护类"), + (("沐浴", "沐浴露"), "沐浴露类"), + (("面霜", "乳液", "精华", "面膜", "护肤"), "面霜/护肤类"), + (("牙膏", "口腔"), "牙膏/口腔护理类"), + (("清洁剂", "清洁", "洗洁"), "清洁剂类"), + (("脱毛",), "脱毛类"), + (("彩妆", "口红", "粉底", "眼影"), "彩妆/唇部类"), + (("香氛", "香水", "止汗"), "香氛/止汗类"), + ) + for keywords, normalized in aliases: + if any(keyword.lower() in cleaned.lower() for keyword in keywords): + return normalized + return None + + +def party_matches(row: dict[str, Any], party_name: str) -> bool: + supplier = clean_value(row.get("supplier")) or "" + return party_name.lower() in supplier.lower() + + +def classify_score(score: float) -> RiskLevel: + if score >= 80: + return RiskLevel.LOW + if score >= 60: + return RiskLevel.MODERATELY_LOW + if score >= 40: + return RiskLevel.MEDIUM + if score >= 20: + return RiskLevel.HIGH + return RiskLevel.VERY_HIGH diff --git a/app/services/llm_client.py b/app/services/llm_client.py index 4d3266f..358b3a8 100644 --- a/app/services/llm_client.py +++ b/app/services/llm_client.py @@ -1,17 +1,21 @@ -"""OpenAI-compatible LLM client for semantic extraction and streaming replies.""" +"""Provider-neutral LLM client with a configurable DeepSeek fallback.""" import json +import logging from collections.abc import AsyncGenerator -from typing import Any +from typing import Any, Literal import httpx from app.core.config import get_settings from app.core.exceptions import AppException +logger = logging.getLogger("dieya_risk_prediction.llm_client") +ProviderName = Literal["primary", "fallback"] + class LLMClient: - """Small provider-neutral client using the OpenAI chat-completions protocol.""" + """Call the primary provider and fail over to DeepSeek when necessary.""" def __init__(self, transport: httpx.AsyncBaseTransport | None = None) -> None: settings = get_settings() @@ -20,22 +24,56 @@ class LLMClient: self.model = settings.llm_model self.timeout = settings.llm_timeout_seconds self.temperature = settings.llm_temperature + + self.fallback_api_key = ( + settings.llm_fallback_api_key.get_secret_value() if settings.llm_fallback_api_key else "" + ) + self.fallback_base_url = settings.llm_fallback_base_url.rstrip("/") + self.fallback_model = settings.llm_fallback_model + self.fallback_api_mode = settings.llm_fallback_api_mode + self.fallback_timeout = settings.llm_fallback_timeout_seconds + self.fallback_max_retries = settings.llm_fallback_max_retries + self.fallback_reasoning_effort = settings.llm_fallback_reasoning_effort + self.fallback_max_output_tokens = settings.llm_fallback_max_output_tokens self.transport = transport @property def configured(self) -> bool: - return bool(self.api_key) + return bool(self.api_key or self.fallback_api_key) + + @property + def fallback_configured(self) -> bool: + return bool(self.fallback_api_key and self.fallback_base_url and self.fallback_model) @property def endpoint(self) -> str: - suffix = "/chat/completions" - return f"{self.base_url}{suffix}" if self.base_url.endswith("/v1") else f"{self.base_url}/v1{suffix}" + return self._chat_endpoint(self.base_url) - def _headers(self) -> dict[str, str]: - if not self.configured: + @staticmethod + def _chat_endpoint(base_url: str) -> str: + suffix = "/chat/completions" + return f"{base_url}{suffix}" if base_url.endswith("/v1") else f"{base_url}/v1{suffix}" + + def _endpoint(self, provider: ProviderName) -> str: + if provider == "primary": + return self.endpoint + if self.fallback_api_mode == "responses": + return f"{self.fallback_base_url}/responses" + return self._chat_endpoint(self.fallback_base_url) + + def _provider_config(self, provider: ProviderName) -> tuple[str, str, int]: + if provider == "primary": + return self.api_key, self.model, self.timeout + return self.fallback_api_key, self.fallback_model, self.fallback_timeout + + def _headers(self, provider: ProviderName) -> dict[str, str]: + api_key, _, _ = self._provider_config(provider) + if not api_key: + if provider == "primary" and self.fallback_configured: + raise AppException("主 LLM 未配置 API Key，将尝试使用兜底模型", code=503, status_code=503) raise AppException("未配置 LLM_API_KEY，暂时无法使用 AI 对话", code=503, status_code=503) return { - "Authorization": f"Bearer {self.api_key}", + "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", "Accept": "text/event-stream", } @@ -45,14 +83,14 @@ class LLMClient: text: str, history: list[dict[str, str]] | None = None, ) -> dict[str, Any]: - """Extract structured product and intent fields before generating an answer.""" + """Extract structured fields, retrying with DeepSeek when the primary fails.""" messages = [ { "role": "system", "content": ( "你是蝶雅产品风险预估系统的语义理解模块。" - "只输出合法 JSON，不要输出 Markdown。抽取用户输入中的意图和风险评估参数。" + "只输出合法 JSON，不要输出 Markdown。请输出 json object。抽取用户输入中的意图和风险评估参数。" "intent 只能是 greet、general、assessment、supplier、material、cases、" "standard、compliance、solution 之一。" "字段必须包含 intent, risk_requested, category, specification, packaging, " @@ -67,87 +105,278 @@ class LLMClient: ] messages.extend(history or []) messages.append({"role": "user", "content": text}) - body = await self._request(messages, stream=False, json_mode=True) - try: - content = body["choices"][0]["message"]["content"] - except (KeyError, IndexError, TypeError) as exc: - raise AppException("LLM 语义结果缺少 content", code=503, status_code=503) from exc - if isinstance(content, list): - content = "".join(str(part.get("text", "")) for part in content if isinstance(part, dict)) - return self._parse_json(str(content)) - async def stream_reply( + try: + body = await self._request(messages, provider="primary", stream=False, json_mode=True) + content = self._response_text(body, provider="primary") + return self._parse_json(content) + except Exception as primary_exc: # noqa: BLE001 - fallback must cover provider and parse failures + logger.warning( + "LLM semantic request failed provider=primary model=%s error=%s", + self.model, + primary_exc.__class__.__name__, + ) + if not self.fallback_configured: + raise + + try: + body = await self._request(messages, provider="fallback", stream=False, json_mode=True) + content = self._response_text(body, provider="fallback") + logger.info("LLM semantic request succeeded provider=fallback model=%s", self.fallback_model) + return self._parse_json(content) + except Exception as fallback_exc: # noqa: BLE001 - normalize both provider failures + logger.exception( + "LLM semantic fallback failed provider=fallback model=%s", + self.fallback_model, + ) + raise AppException( + "主模型与兜底模型均调用失败，请检查网络、模型配置或额度", + code=503, + status_code=503, + ) from fallback_exc + + async def stream_reply(self, messages: list[dict[str, str]]) -> AsyncGenerator[str, None]: + """Yield reply deltas and fail over only before any primary text is emitted.""" + + emitted = False + try: + async for delta in self._stream_provider(messages, provider="primary"): + emitted = True + yield delta + return + except Exception as primary_exc: # noqa: BLE001 - fallback is the public behavior + logger.warning( + "LLM stream failed provider=primary model=%s error=%s emitted=%s", + self.model, + primary_exc.__class__.__name__, + emitted, + ) + if emitted or not self.fallback_configured: + raise + + logger.info("LLM stream switching provider=primary to provider=fallback model=%s", self.fallback_model) + async for delta in self._stream_provider(messages, provider="fallback"): + yield delta + + async def generate_prevention_suggestions(self, context: dict[str, Any]) -> list[str]: + """Generate executable prevention actions from the current risk context.""" + + messages = [ + { + "role": "system", + "content": ( + "你是蝶雅产品风险预估系统的预防方案生成模块。" + "请根据提供的风险评估数据生成可执行的质量预防建议。" + "只输出合法 json object，不要输出 Markdown，格式必须是 " + '{"suggestions":["建议1","建议2"]}。' + "输出 4 到 6 条建议，每条不超过 80 个汉字。" + "建议必须优先针对分数较低的风险维度、历史异常频发项和命中的标准规则，" + "明确检测或管控动作；可以包含责任角色、阶段和验收闭环。" + "不得虚构数据库没有提供的案例、检测结果、法规条款或 RAG 检索结果。" + "数据不足时明确写出需要补充的验证动作，不要把猜测写成事实。" + "不要重复表达，不要输出标题、解释、风险评分或额外字段。" + ), + }, + { + "role": "user", + "content": "以下是本次风险评估数据，请生成预防建议：\n" + json.dumps( + context, ensure_ascii=False, separators=(",", ":") + ), + }, + ] + try: + body = await self._request(messages, provider="primary", stream=False, json_mode=True) + raw = self._response_text(body, provider="primary") + return self._parse_suggestions(raw) + except Exception as primary_exc: # noqa: BLE001 - prevention suggestions have the same failover policy + logger.warning( + "LLM prevention suggestion request failed provider=primary model=%s error=%s", + self.model, + primary_exc.__class__.__name__, + ) + if not self.fallback_configured: + raise + + try: + body = await self._request(messages, provider="fallback", stream=False, json_mode=True) + raw = self._response_text(body, provider="fallback") + suggestions = self._parse_suggestions(raw) + logger.info( + "LLM prevention suggestions generated provider=fallback model=%s count=%s", + self.fallback_model, + len(suggestions), + ) + return suggestions + except Exception as fallback_exc: # noqa: BLE001 - caller owns deterministic fallback behavior + logger.exception( + "LLM prevention suggestion fallback failed provider=fallback model=%s", + self.fallback_model, + ) + raise AppException( + "主模型与兜底模型均无法生成预防建议", + code=503, + status_code=503, + ) from fallback_exc + + async def _stream_provider( self, messages: list[dict[str, str]], + *, + provider: ProviderName, ) -> AsyncGenerator[str, None]: - """Yield text deltas from an OpenAI-compatible streaming response.""" - - headers = self._headers() - payload = { - "model": self.model, - "messages": messages, - "temperature": self.temperature, - "stream": True, - } - timeout = httpx.Timeout(self.timeout, connect=min(15, self.timeout)) + headers = self._headers(provider) + _, model, timeout_seconds = self._provider_config(provider) + is_responses = provider == "fallback" and self.fallback_api_mode == "responses" + payload = self._payload(messages, provider=provider, stream=True, json_mode=False) + timeout = httpx.Timeout(timeout_seconds, connect=min(15, timeout_seconds)) async with httpx.AsyncClient(timeout=timeout, transport=self.transport) as client: - async with client.stream("POST", self.endpoint, headers=headers, json=payload) as response: - if response.status_code >= 400: - detail = await response.aread() - raise AppException( - f"LLM 请求失败（HTTP {response.status_code}）", - code=503, - status_code=503, - details=detail.decode("utf-8", errors="replace")[:500], - ) - async for line in response.aiter_lines(): - if not line.startswith("data:"): - continue - raw = line[5:].strip() - if not raw or raw == "[DONE]": - continue - try: - chunk = json.loads(raw) - except json.JSONDecodeError: - continue - choices = chunk.get("choices") or [] - if not choices: - continue - delta = (choices[0].get("delta") or {}).get("content") - if isinstance(delta, list): - delta = "".join( - str(part.get("text", "")) for part in delta if isinstance(part, dict) - ) - if delta: - yield str(delta) + try: + async with client.stream("POST", self._endpoint(provider), headers=headers, json=payload) as response: + if response.status_code >= 400: + detail = await response.aread() + raise self._http_failure(response.status_code, detail) + async for line in response.aiter_lines(): + if not line.startswith("data:"): + continue + raw = line[5:].strip() + if not raw or raw == "[DONE]": + continue + try: + chunk = json.loads(raw) + except json.JSONDecodeError: + continue + delta = self._stream_delta(chunk, responses_api=is_responses) + if delta: + yield delta + except AppException: + raise + except httpx.HTTPError as exc: + raise AppException( + "LLM 流式连接失败", + code=503, + status_code=503, + ) from exc + + logger.info("LLM stream completed provider=%s model=%s", provider, model) async def _request( self, messages: list[dict[str, str]], *, + provider: ProviderName, stream: bool, json_mode: bool = False, ) -> dict[str, Any]: - headers = self._headers() - payload: dict[str, Any] = { - "model": self.model, + headers = self._headers(provider) + payload = self._payload(messages, provider=provider, stream=stream, json_mode=json_mode) + _, model, timeout_seconds = self._provider_config(provider) + timeout = httpx.Timeout(timeout_seconds, connect=min(15, timeout_seconds)) + logger.info( + "LLM request start provider=%s model=%s endpoint=%s stream=%s", + provider, + model, + self._endpoint(provider), + stream, + ) + async with httpx.AsyncClient(timeout=timeout, transport=self.transport) as client: + try: + response = await client.post(self._endpoint(provider), headers=headers, json=payload) + if response.status_code >= 400: + detail = await response.aread() + raise self._http_failure(response.status_code, detail) + except AppException: + raise + except httpx.HTTPError as exc: + raise AppException("LLM 语义理解请求失败", code=503, status_code=503) from exc + try: + data = response.json() + except ValueError as exc: + raise AppException("LLM 返回格式无效", code=503, status_code=503) from exc + if not isinstance(data, dict): + raise AppException("LLM 返回格式无效", code=503, status_code=503) + logger.info("LLM request success provider=%s model=%s status=%s", provider, model, response.status_code) + return data + + def _payload( + self, + messages: list[dict[str, str]], + *, + provider: ProviderName, + stream: bool, + json_mode: bool, + ) -> dict[str, Any]: + if provider == "fallback" and self.fallback_api_mode == "responses": + payload: dict[str, Any] = { + "model": self.fallback_model, + "input": messages, + "stream": stream, + "max_output_tokens": self.fallback_max_output_tokens, + } + if self.fallback_reasoning_effort: + payload["reasoning_effort"] = self.fallback_reasoning_effort + return payload + + _, model, _ = self._provider_config(provider) + payload = { + "model": model, "messages": messages, "temperature": self.temperature, "stream": stream, } if json_mode: payload["response_format"] = {"type": "json_object"} - timeout = httpx.Timeout(self.timeout, connect=min(15, self.timeout)) - async with httpx.AsyncClient(timeout=timeout, transport=self.transport) as client: - try: - response = await client.post(self.endpoint, headers=headers, json=payload) - response.raise_for_status() - except httpx.HTTPError as exc: - raise AppException("LLM 语义理解请求失败", code=503, status_code=503) from exc + return payload + + @staticmethod + def _http_failure(status_code: int, detail: bytes) -> AppException: + return AppException( + f"LLM 请求失败（HTTP {status_code}）", + code=503, + status_code=503, + details=detail.decode("utf-8", errors="replace")[:500], + ) + + @staticmethod + def _response_text(body: dict[str, Any], *, provider: ProviderName) -> str: + if provider == "fallback": + output_text = body.get("output_text") + if isinstance(output_text, str) and output_text: + return output_text + parts: list[str] = [] + for item in body.get("output") or []: + if not isinstance(item, dict): + continue + content = item.get("content") + if isinstance(content, str): + parts.append(content) + for part in content or []: + if isinstance(part, dict) and part.get("type") in {"output_text", "text"}: + parts.append(str(part.get("text") or "")) + if parts: + return "".join(parts) + try: - return response.json() - except ValueError as exc: - raise AppException("LLM 返回格式无效", code=503, status_code=503) from exc + content = body["choices"][0]["message"]["content"] + except (KeyError, IndexError, TypeError) as exc: + raise AppException("LLM 语义结果缺少 content", code=503, status_code=503) from exc + if isinstance(content, list): + content = "".join(str(part.get("text", "")) for part in content if isinstance(part, dict)) + return str(content) + + @staticmethod + def _stream_delta(chunk: dict[str, Any], *, responses_api: bool) -> str: + if responses_api: + event_type = chunk.get("type") or chunk.get("event") + if event_type in {"response.output_text.delta", "output_text.delta"}: + return str(chunk.get("delta") or "") + return "" + choices = chunk.get("choices") or [] + if not choices: + return "" + delta = (choices[0].get("delta") or {}).get("content") + if isinstance(delta, list): + delta = "".join(str(part.get("text", "")) for part in delta if isinstance(part, dict)) + return str(delta or "") @staticmethod def _parse_json(content: str) -> dict[str, Any]: @@ -161,3 +390,28 @@ class LLMClient: if not isinstance(parsed, dict): raise AppException("LLM 语义结果格式无效", code=503, status_code=503) return parsed + + @staticmethod + def _parse_suggestions(content: str) -> list[str]: + """Parse and normalize the model's suggestion-only JSON response.""" + + cleaned = content.strip() + if cleaned.startswith("```"): + cleaned = cleaned.removeprefix("```").removeprefix("json").removesuffix("```").strip() + try: + parsed = json.loads(cleaned) + except json.JSONDecodeError as exc: + raise AppException("LLM 预防建议不是合法 JSON", code=503, status_code=503) from exc + values = parsed.get("suggestions") if isinstance(parsed, dict) else parsed + if not isinstance(values, list): + raise AppException("LLM 预防建议格式无效", code=503, status_code=503) + suggestions: list[str] = [] + for value in values: + if not isinstance(value, str): + continue + item = " ".join(value.strip().split()) + if item and item not in suggestions: + suggestions.append(item[:100]) + if not suggestions: + raise AppException("LLM 未返回有效预防建议", code=503, status_code=503) + return suggestions[:6] diff --git a/app/services/risk_prediction.py b/app/services/risk_prediction.py index 9c16855..63d0bc5 100644 --- a/app/services/risk_prediction.py +++ b/app/services/risk_prediction.py @@ -1,12 +1,17 @@ """Risk prediction use-case orchestration.""" +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.exceptions import AppException from app.domain.risk_engine import BaselineRiskEngine, IndicatorValue from app.schemas.risk_prediction import ( ModelInfo, RiskAssessmentRequest, RiskAssessmentResult, + RiskDimensionScore, RiskFactorResult, ) +from app.services.database_risk_scoring import DatabaseRiskScoringService class RiskPredictionService: @@ -14,10 +19,63 @@ class RiskPredictionService: @classmethod def model_info(cls) -> ModelInfo: - return ModelInfo(model_version=cls._engine.model_version) + return ModelInfo(model_version="database-seven-dimension-v1") @classmethod - def preview(cls, request: RiskAssessmentRequest) -> RiskAssessmentResult: + async def preview( + cls, + request: RiskAssessmentRequest, + session: AsyncSession | None = None, + ) -> RiskAssessmentResult: + if session and any( + value is not None + for value in ( + request.product_category, + request.material_type, + request.supplier, + request.factory, + request.stage_new, + request.defect_level, + ) + ): + database_score = await DatabaseRiskScoringService(session).score( + category=request.product_category, + material=request.material_type, + supplier=request.supplier, + factory=request.factory, + stage=request.stage_new, + defect_level=request.defect_level, + ) + dimensions = [ + RiskDimensionScore( + dimension=item.name, + score=item.score, + weight=item.weight, + status=item.status, + reason=item.reason, + sample_size=item.sample_size, + raw_value=item.raw_value, + ) + for item in database_score.dimensions + ] + return RiskAssessmentResult( + subject_id=request.subject_id, + score=database_score.score, + level=database_score.level, + model_version="database-seven-dimension-v1", + factors=[ + RiskFactorResult( + name=item.name, + value=round(100 - item.score, 2), + weight=item.weight, + contribution=round(item.score * item.weight / 100, 2), + ) + for item in database_score.dimensions + ], + dimension_scores=dimensions, + ) + if not request.indicators: + raise AppException("请填写产品评估参数", code=422, status_code=422) score = cls._engine.predict( [ IndicatorValue(name=item.name, value=item.value, weight=item.weight) @@ -39,4 +97,3 @@ class RiskPredictionService: for item in score.factors ], ) - diff --git a/tests/test_conversations.py b/tests/test_conversations.py index 165781a..aeec69f 100644 --- a/tests/test_conversations.py +++ b/tests/test_conversations.py @@ -1,5 +1,6 @@ from datetime import UTC, datetime +import pytest from app.models.entities import Conversation, ConversationMessage from app.services.conversations import ConversationService @@ -119,3 +120,61 @@ def test_product_assessment_semantics_override_nlu_miss() -> None: assert normalized["intent"] == "assessment" assert normalized["risk_requested"] is True + + +def test_short_known_supplier_follow_up_is_routed_to_supplier_card() -> None: + assert ConversationService._infer_supplier_follow_up( + "佳亿", + ["佳亿", "祥印", "瑞沣"], + ) == "佳亿" + assert ConversationService._infer_supplier_follow_up( + "这个供应商", + ["佳亿", "祥印"], + ) is None + + +def test_unknown_standalone_name_is_treated_as_new_supplier() -> None: + assert ConversationService._infer_unknown_supplier_follow_up("德盛") == "德盛" + assert ConversationService._infer_unknown_supplier_follow_up("新供应商A") is None + assert ConversationService._infer_unknown_supplier_follow_up("测试供应商") is None + assert ConversationService._infer_unknown_supplier_follow_up("新供应商B") is None + assert ConversationService._infer_unknown_supplier_follow_up("洗发水") is None + assert ConversationService._infer_unknown_supplier_follow_up("帮我查德盛") is None + assert ConversationService._response_type("测试供应商", {"intent": "supplier"}) == "general" + assert ConversationService._response_type("新供应商A", {"intent": "supplier"}) == "general" + + +@pytest.mark.asyncio +async def test_any_product_without_history_uses_solution_card(monkeypatch: pytest.MonkeyPatch) -> None: + service = object.__new__(ConversationService) + + async def no_history(_: str) -> int: + return 0 + + monkeypatch.setattr(service, "_count_category_anomalies", no_history) + + response_type = ConversationService._response_type( + "评估防晒霜新品，50g，软管包装", + {"intent": "assessment", "risk_requested": True, "category": "防晒霜"}, + ) + result = await service._use_solution_card_without_history( + response_type, + "评估防晒霜新品，50g，软管包装", + {"intent": "assessment", "risk_requested": True, "category": "防晒霜"}, + ) + + assert response_type == "risk_assessment" + assert result == "solution" + + +@pytest.mark.asyncio +async def test_unknown_product_without_category_uses_solution_card() -> None: + service = object.__new__(ConversationService) + + result = await service._use_solution_card_without_history( + "risk_assessment", + "评估这个新产品的风险", + {"intent": "assessment", "risk_requested": True}, + ) + + assert result == "solution" diff --git a/tests/test_database_risk_scoring.py b/tests/test_database_risk_scoring.py new file mode 100644 index 0000000..242b019 --- /dev/null +++ b/tests/test_database_risk_scoring.py @@ -0,0 +1,65 @@ +from datetime import date + +import pytest +from app.services.database_risk_scoring import DatabaseRiskScoringService + + +def _row(**values): + return { + "date": date(2026, 1, 10), + "product_name": "洗发水 500ml", + "product_category": None, + "material_type": "瓶子", + "supplier": "祥印", + "factory": "一一", + "stage_new": "第三阶段（产品开发）", + "defect_level": None, + **values, + } + + +@pytest.mark.asyncio +async def test_first_phase_score_uses_database_dimensions_and_prd_fallbacks() -> None: + service = DatabaseRiskScoringService(object(), as_of=date(2026, 9, 4)) + + async def load_rows(): + return [ + _row(), + _row(product_name="洗发水 250ml", defect_level="B"), + ] + + service._load_rows = load_rows # type: ignore[method-assign] + result = await service.score( + category="洗发水", + material="瓶子", + supplier="祥印", + stage="产品开发", + ) + + assert len(result.dimensions) == 7 + assert sum(item.weight for item in result.dimensions) == 100 + assert result.level.value == "medium" + assert result.dimensions[0].score == 95.57 + assert result.dimensions[0].raw_value == 2 + assert result.dimensions[1].score == 0 + assert result.dimensions[2].score == 45 + assert result.dimensions[3].status == "default" + assert result.dimensions[4].status == "default" + assert result.dimensions[5].score == 50 + assert result.dimensions[6].score == 85 + assert result.score == 52.39 + + +@pytest.mark.asyncio +async def test_supplier_dimension_does_not_count_factory_records() -> None: + service = DatabaseRiskScoringService(object(), as_of=date(2026, 9, 4)) + + async def load_rows(): + return [_row(supplier=None, factory="祥印")] + + service._load_rows = load_rows # type: ignore[method-assign] + result = await service.score(supplier="祥印") + + supplier_dimension = result.dimensions[0] + assert supplier_dimension.score == 70 + assert supplier_dimension.raw_value == 0 diff --git a/tests/test_llm_protocol.py b/tests/test_llm_protocol.py index c8ee269..387bf08 100644 --- a/tests/test_llm_protocol.py +++ b/tests/test_llm_protocol.py @@ -18,6 +18,9 @@ async def test_openai_compatible_semantic_and_stream_protocol(monkeypatch: pytes nonlocal calls calls += 1 if calls == 1: + request_body = json.loads(request.content) + assert request_body["response_format"] == {"type": "json_object"} + assert "json" in request_body["messages"][0]["content"] return httpx.Response( 200, json={ @@ -45,13 +48,86 @@ async def test_openai_compatible_semantic_and_stream_protocol(monkeypatch: pytes client = LLMClient(transport=httpx.MockTransport(handler)) semantic = await client.extract_semantics("评估洗发水新品") - chunks = [ - chunk - async for chunk in client.stream_reply([{"role": "user", "content": "继续分析"}]) - ] + chunks = [chunk async for chunk in client.stream_reply([{"role": "user", "content": "继续分析"}])] assert semantic["intent"] == "assessment" assert semantic["confidence"] == 0.92 assert "".join(chunks) == "第一段第二段" assert calls == 2 get_settings.cache_clear() + + +@pytest.mark.asyncio +async def test_semantic_request_fails_over_to_deepseek_responses(monkeypatch: pytest.MonkeyPatch) -> None: + get_settings.cache_clear() + monkeypatch.setenv("RISK_LLM_API_KEY", "primary-key") + monkeypatch.setenv("RISK_LLM_BASE_URL", "https://primary.test/v1") + monkeypatch.setenv("RISK_LLM_MODEL", "primary-model") + monkeypatch.setenv("RISK_LLM_FALLBACK_API_KEY", "fallback-key") + monkeypatch.setenv("RISK_LLM_FALLBACK_BASE_URL", "https://fallback.test") + monkeypatch.setenv("RISK_LLM_FALLBACK_MODEL", "deepseek-v4-pro") + monkeypatch.setenv("RISK_LLM_FALLBACK_API_MODE", "responses") + calls: list[str] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + calls.append(str(request.url)) + if request.url.path.endswith("/chat/completions"): + return httpx.Response(503, json={"error": {"message": "primary unavailable"}}) + body = json.loads(request.content) + assert body["model"] == "deepseek-v4-pro" + assert body["input"][-1]["content"] == "评估洗发水新品" + return httpx.Response( + 200, + json={ + "output_text": json.dumps( + {"intent": "assessment", "risk_requested": True, "confidence": 0.8}, + ensure_ascii=False, + ) + }, + ) + + client = LLMClient(transport=httpx.MockTransport(handler)) + result = await client.extract_semantics("评估洗发水新品") + + assert result["intent"] == "assessment" + assert calls == [ + "https://primary.test/v1/chat/completions", + "https://fallback.test/responses", + ] + get_settings.cache_clear() + + +@pytest.mark.asyncio +async def test_stream_request_fails_over_before_primary_emits_text(monkeypatch: pytest.MonkeyPatch) -> None: + get_settings.cache_clear() + monkeypatch.setenv("RISK_LLM_API_KEY", "primary-key") + monkeypatch.setenv("RISK_LLM_BASE_URL", "https://primary.test/v1") + monkeypatch.setenv("RISK_LLM_FALLBACK_API_KEY", "fallback-key") + monkeypatch.setenv("RISK_LLM_FALLBACK_BASE_URL", "https://fallback.test") + monkeypatch.setenv("RISK_LLM_FALLBACK_MODEL", "deepseek-v4-pro") + monkeypatch.setenv("RISK_LLM_FALLBACK_API_MODE", "responses") + calls: list[str] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + calls.append(str(request.url)) + if request.url.path.endswith("/chat/completions"): + raise httpx.ConnectError("primary connection failed", request=request) + return httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + content=( + 'data: {"type":"response.output_text.delta","delta":"兜底"}\n\n' + 'data: {"type":"response.output_text.delta","delta":"回复"}\n\n' + "data: [DONE]\n\n" + ).encode(), + ) + + client = LLMClient(transport=httpx.MockTransport(handler)) + chunks = [chunk async for chunk in client.stream_reply([{"role": "user", "content": "继续分析"}])] + + assert "".join(chunks) == "兜底回复" + assert calls == [ + "https://primary.test/v1/chat/completions", + "https://fallback.test/responses", + ] + get_settings.cache_clear()