"""Provider-neutral LLM client with a configurable DeepSeek fallback.""" import json import logging from collections.abc import AsyncGenerator 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: """Call the primary provider and fail over to DeepSeek when necessary.""" def __init__(self, transport: httpx.AsyncBaseTransport | None = None) -> None: settings = get_settings() self.api_key = settings.llm_api_key.get_secret_value() if settings.llm_api_key else "" self.base_url = settings.llm_base_url.rstrip("/") 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 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: return self._chat_endpoint(self.base_url) @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 {api_key}", "Content-Type": "application/json", "Accept": "text/event-stream", } async def extract_semantics( self, text: str, history: list[dict[str, str]] | None = None, ) -> dict[str, Any]: """Extract structured fields, retrying with DeepSeek when the primary fails.""" messages = [ { "role": "system", "content": ( "你是蝶雅产品风险预警系统的语义理解模块。" "只输出合法 JSON，不要输出 Markdown。请输出 json object。抽取用户输入中的意图和风险评估参数。" "intent 只能是 greet、general、assessment、supplier、material、cases、" "standard、compliance、solution 之一。" "字段必须包含 intent, risk_requested, category, specification, packaging, " "supplier, factory, stage, confidence。" "只有用户明确要求风险评估、风险查询、标准判定、案例检索或预防方案时 risk_requested 才为 true；" "用户描述要开发新品并提供规格、容量、包装材质、供应商或加工厂等参数时，" "即使没有出现‘风险’二字，也必须将 intent 设为 assessment、risk_requested 设为 true；" "问候、闲聊、测试回复、要求原样回复等情况必须为 false。" "缺失字段使用 null，confidence 为 0 到 1 的数字。" ), }, ] messages.extend(history or []) messages.append({"role": "user", "content": text}) 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 extract_ocr_fields(self, text: str) -> dict[str, Any]: """Turn OCR text into an editable risk-item draft without inventing absent values.""" messages = [ { "role": "system", "content": ( "你是蝶雅产品风险预警系统的OCR字段抽取模块，只输出合法JSON对象。" "只能提取输入文字明确出现的信息，不得推测或补造。输出字段：product_name、product_category、" "supplier、factory、material_type、stage_new、description、defect_level、occurred_on、confidence、" "measurements。缺失标量用null。defect_level只能是A、B、C或null；occurred_on使用YYYY-MM-DD或null。" "measurements为数组，每项包含item、raw_value、value、unit、confidence；value必须是数字或null。" "measurements只允许质量检测指标，例如pH、色差、重金属含量、力矩、喷出量、附着力、" "不合格率；数量、抽检数量、产品规格、容量、日期、单号和批次不得作为检测指标。" "confidence为0到1。description应保留异常现象和影响比例，但不得加入原文没有的结论。" ), }, {"role": "user", "content": text[:20_000]}, ] last_error: Exception | None = None providers: list[ProviderName] = ["primary"] if self.fallback_configured: providers.append("fallback") for provider in providers: try: body = await self._request(messages, provider=provider, stream=False, json_mode=True) return self._parse_json(self._response_text(body, provider=provider)) except Exception as exc: # noqa: BLE001 - deterministic extraction handles provider failures last_error = exc logger.warning("LLM OCR extraction failed provider=%s error=%s", provider, exc.__class__.__name__) if last_error: raise last_error return {} 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]: 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: 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(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"} 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: 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]: 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 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]