"""星火 / OpenAI 兼容 API 抽取器:JSON 模式、Schema 校验、失败重试 2 次、超时降级。 - 密钥只从环境变量读取(默认 SPARK_API_KEY),绝不写入代码或配置; - 无密钥时 get_extractor() 显式降级到离线抽取器; - 返回结果先过 ExtractionResult(pydantic)强校验,不合法即重试; - 重试耗尽或超时 → 降级到离线通道,保证链路永不断。 """ from __future__ import annotations import json import logging import os import httpx from app.extract.base import Extractor, load_lexicon from app.extract.offline import OfflineExtractor from app.extract.prompts import PROMPT_VERSION, SYSTEM_PROMPT, build_prompt from app.models import ExtractionResult, JDRecord logger = logging.getLogger(__name__) MAX_RETRIES = 2 # 失败重试 2 次 class LLMExtractor: """大模型抽取器(需配置密钥才启用)。""" name = "llm" def __init__( self, api_key: str, base_url: str, model: str, timeout: float = 20.0, fallback: Extractor | None = None, ): self.api_key = api_key self.base_url = base_url.rstrip("/") self.model = model self.timeout = timeout self.fallback = fallback or OfflineExtractor(load_lexicon()) def _call_api(self, jd_text: str) -> str: """调用 OpenAI 兼容 chat/completions,返回模型输出文本。""" resp = httpx.post( f"{self.base_url}/chat/completions", headers={"Authorization": f"Bearer {self.api_key}"}, json={ "model": self.model, "messages": [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": build_prompt(jd_text)}, ], "response_format": {"type": "json_object"}, "temperature": 0.1, }, timeout=self.timeout, ) resp.raise_for_status() return resp.json()["choices"][0]["message"]["content"] @staticmethod def _parse(content: str, record: JDRecord) -> ExtractionResult: """Schema 强校验:坏 JSON / 缺字段都会抛异常,由上层重试。 先剥离模型常见的 ```json 代码块包裹再解析。 """ text = content.strip() if text.startswith("```"): text = text.split("\n", 1)[-1] if "\n" in text else text.strip("`") if text.rstrip().endswith("```"): text = text.rstrip()[:-3] data = json.loads(text.strip()) result = ExtractionResult(**data) result.jid = record.jid result.extractor = "llm" result.prompt_version = PROMPT_VERSION result.confidence = 0.9 return result def extract(self, record: JDRecord) -> ExtractionResult: last_err: Exception | None = None for attempt in range(MAX_RETRIES + 1): try: return self._parse(self._call_api(record.text), record) except Exception as exc: # noqa: BLE001 - 网络/解析/校验任何失败都要可重试可降级 last_err = exc logger.warning("LLM 抽取第 %d 次失败(%s),%s", attempt + 1, type(exc).__name__, "重试" if attempt < MAX_RETRIES else "重试耗尽") logger.warning("LLM 不可用(%s),降级离线抽取 jid=%s", last_err, record.jid) return self.fallback.extract(record) def get_extractor(config: dict) -> Extractor: """按配置选择抽取通道:配置了密钥(config.json 或环境变量)→ LLM,否则显式降级离线。 密钥优先级:config.json 的 llm_api_key > 环境变量(默认 SPARK_API_KEY)。 LLM 通道失败时自动降级离线抽取(见 LLMExtractor.extract),链路永不断。 """ key_env = str(config.get("llm_api_key_env", "SPARK_API_KEY")) api_key = (str(config.get("llm_api_key") or "").strip() or os.environ.get(key_env, "").strip()) offline = OfflineExtractor(load_lexicon(config.get("lexicon_path") or None)) if api_key: logger.info("检测到 LLM 密钥,启用大模型抽取通道(模型 %s)", config.get("llm_model")) return LLMExtractor( api_key=api_key, base_url=str(config.get("llm_base_url")), model=str(config.get("llm_model")), timeout=float(config.get("llm_timeout", 20)), fallback=offline, ) logger.info("未配置 LLM 密钥,显式降级到离线确定性抽取器") return offline def llm_ready(config: dict) -> bool: """是否已配置 LLM 密钥(config.json 或环境变量)。""" key_env = str(config.get("llm_api_key_env", "SPARK_API_KEY")) return bool(str(config.get("llm_api_key") or "").strip() or os.environ.get(key_env, "").strip())