diff --git a/api/insurance/generation/celery_tasks.py b/api/insurance/generation/celery_tasks.py index f6e0b71..7b535e4 100644 --- a/api/insurance/generation/celery_tasks.py +++ b/api/insurance/generation/celery_tasks.py @@ -177,6 +177,10 @@ def _execute_ppt_parse(task_id: str): session_id, overall, stage_message, extractions=extractions ) + logger.info( + f"[PPT解析] 开始文件 {index}/{total}: {filename or filepath} " + f"plan_type={plan_type}" + ) try: result = asyncio.run(orchestrator.extract_plan( filepath, @@ -201,6 +205,12 @@ def _execute_ppt_parse(task_id: str): "yearCount": len(result.data.get("benefit_illustration", [])) if result.data else 0, "provenance": result.provenance, }) + logger.info( + f"[PPT解析] 完成文件 {index}/{total}: {filename or filepath} " + f"status={result.status} rows=" + f"{len(result.data.get('benefit_illustration', [])) if result.data else 0} " + f"error={result.error or '-'}" + ) except Exception as exc: logger.error(f"PDF 解析失败 [{filename}]: {exc}", exc_info=True) extractions.append({ @@ -220,6 +230,7 @@ def _execute_ppt_parse(task_id: str): return all_failed = all(e.get("status") == "error" for e in extractions) + needs_review = any(e.get("status") != "success" for e in extractions) first_error = next( (str(e.get("error")) for e in extractions if e.get("error")), "所有文件均处理失败", @@ -227,7 +238,11 @@ def _execute_ppt_parse(task_id: str): session.extractions_json = json.dumps(extractions, ensure_ascii=False) session.status = "error" if all_failed else "parsed" session.parse_progress = 100 - session.parse_message = "处理失败" if all_failed else "处理完成" + session.parse_message = ( + "处理失败" if all_failed + else "处理完成，部分文件需校对" if needs_review + else "处理完成" + ) session.parse_error = first_error[:1000] if all_failed else None session.parse_finished_at = datetime.now() # 成功时 workflow_step 推进到 review（数据核对），失败保持 parsing @@ -240,7 +255,11 @@ def _execute_ppt_parse(task_id: str): status="failed" if all_failed else "done", stage="completed", progress=100, - message="解析失败" if all_failed else "解析完成", + message=( + "解析失败" if all_failed + else "解析完成，部分文件需校对" if needs_review + else "解析完成" + ), error_code="all_failed" if all_failed else "", error_message=first_error[:1000] if all_failed else "", finished_at=datetime.now(), diff --git a/api/insurance/ppt/extraction.py b/api/insurance/ppt/extraction.py index ff5980b..937c2d9 100644 --- a/api/insurance/ppt/extraction.py +++ b/api/insurance/ppt/extraction.py @@ -13,7 +13,7 @@ from typing import Callable, Optional logger = logging.getLogger(__name__) -CACHE_VERSION = 7 +CACHE_VERSION = 8 def _format_exception(exc: Exception) -> str: @@ -107,6 +107,84 @@ def _format_pdf_pages(text_parts: list[str]) -> str: ) +def _split_marked_pages(pdf_text: str) -> list[tuple[int, str]]: + """拆分带 [PAGE N] 标记的文本；无标记时视为单页。""" + parts = re.split(r"\[PAGE\s+(\d+)\]", pdf_text or "", flags=re.IGNORECASE) + if len(parts) <= 1: + return [(1, (pdf_text or "").strip())] if (pdf_text or "").strip() else [] + + pages = [] + for index in range(1, len(parts), 2): + page_num = int(parts[index]) + content = parts[index + 1].strip() if index + 1 < len(parts) else "" + if content: + pages.append((page_num, content)) + return pages + + +def _select_page_chunks( + pdf_text: str, + keywords: tuple[str, ...] = (), + min_keyword_hits: int = 1, + pages_per_chunk: int = 2, + max_chars: int = 9000, +) -> list[str]: + """筛选指定页面并切成小块，避免单次 LLM 请求承载整份计划书。""" + pages = _split_marked_pages(pdf_text) + if keywords: + lowered_keywords = tuple(keyword.lower() for keyword in keywords) + pages = [ + (page_num, content) + for page_num, content in pages + if sum(keyword in content.lower() for keyword in lowered_keywords) >= min_keyword_hits + ] + + chunks: list[str] = [] + current: list[str] = [] + current_chars = 0 + for page_num, content in pages: + page_text = f"[PAGE {page_num}]\n{content}" + if len(page_text) > max_chars: + page_text = page_text[:max_chars] + if current and ( + len(current) >= pages_per_chunk + or current_chars + len(page_text) > max_chars + ): + chunks.append("\n\n".join(current)) + current = [] + current_chars = 0 + current.append(page_text) + current_chars += len(page_text) + if current: + chunks.append("\n\n".join(current)) + return chunks + + +def _merge_rows_by_policy_year(rows: list[dict]) -> list[dict]: + """按保单年度合并分块结果，重复年度保留字段更完整的一行。""" + by_year: dict[int, dict] = {} + without_year = [] + for row in rows: + if not isinstance(row, dict): + continue + try: + year = int(row.get("policy_year")) + except (TypeError, ValueError): + without_year.append(row) + continue + if year <= 0: + without_year.append(row) + continue + normalized = dict(row) + normalized["policy_year"] = year + existing = by_year.get(year) + if existing is None or sum(value is not None for value in normalized.values()) > sum( + value is not None for value in existing.values() + ): + by_year[year] = normalized + return [by_year[year] for year in sorted(by_year)] + without_year + + def _extract_pdf_text(pdf_path: str, max_chars: int = 120000) -> tuple[str, list[dict]]: """提取 PDF 文本，支持多种 PDF 解析库。 @@ -136,7 +214,7 @@ def _extract_pdf_text(pdf_path: str, max_chars: int = 120000) -> tuple[str, list page_qualities = _assess_page_qualities(text_parts) return truncated, page_qualities if text.strip(): - logger.warning("PyMuPDF 提取结果疑似乱码，将尝试 OCR") + logger.warning("PyMuPDF 提取结果疑似乱码，将尝试其他文本解析器，必要时使用 OCR") except ImportError: logger.debug("PyMuPDF 未安装，尝试下一个库") except Exception as e: @@ -669,6 +747,18 @@ def _apply_filename_hints(data: dict, pdf_path: str, plan_type: str) -> dict: return data +def _has_sufficient_benefit_rows(rows: list[dict]) -> bool: + years = set() + for row in rows: + try: + year = int((row or {}).get("policy_year")) + except (TypeError, ValueError): + continue + if year > 0: + years.add(year) + return len(rows) >= 20 or (len(years) >= 5 and {10, 20, 30}.issubset(years) and max(years) >= 30) + + def assess_extraction_payload(data: Optional[dict], plan_type: str) -> tuple[str, str]: """Return extraction status and a user-facing error when data is incomplete.""" if not isinstance(data, dict) or not data: @@ -725,6 +815,14 @@ def assess_extraction_payload(data: Optional[dict], plan_type: str) -> tuple[str else: if not has_positive_number(policy.get("annual_premium")): problems.append("年缴保费缺失") + if benefit_rows and not _has_sufficient_benefit_rows(benefit_rows): + problems.append(f"利益演示数据不足（当前 {len(benefit_rows)} 行）") + + split_benefit = ((data.get("_meta") or {}).get("split_extraction") or {}).get("benefit") or {} + if split_benefit.get("status") in ("failed", "partial") and not any( + "利益演示" in problem for problem in problems + ): + problems.append(f"利益演示分块提取未完成（当前 {len(benefit_rows)} 行）") if problems: return "partial", "；".join(problems[:3]) @@ -860,9 +958,50 @@ async def _llm_extract_split( """ from insurance.ppt.prompts import select_key_pages - extraction_text = select_key_pages(pdf_text, max_pages=10, max_chars=28000) + identity_text = select_key_pages(pdf_text, max_pages=3, max_chars=8000) + benefit_keywords = ( + "保单年度", "保單年度", "退保价值", "退保價值", "保证现金", "保證現金", + "policy year", "cash value", "surrender value", "account value", + ) + benefit_chunks = _select_page_chunks( + pdf_text, + benefit_keywords, + min_keyword_hits=2, + pages_per_chunk=2, + max_chars=9000, + ) + if not benefit_chunks: + benefit_fallback = select_key_pages(pdf_text, max_pages=6, max_chars=18000) + benefit_chunks = _select_page_chunks( + benefit_fallback, + pages_per_chunk=2, + max_chars=9000, + ) + merged_data = {} last_response = None + split_diagnostics = { + "tokens": {"input": 0, "output": 0}, + "identity": {"status": "pending"}, + "benefit": { + "status": "pending", + "chunkCount": len(benefit_chunks), + "completedChunks": 0, + "errors": [], + }, + "withdrawal": {"status": "pending"}, + } + + def record_response(block: str, response) -> None: + tokens = getattr(response, "tokens", None) or {} + split_diagnostics["tokens"]["input"] += int(tokens.get("input", 0) or 0) + split_diagnostics["tokens"]["output"] += int(tokens.get("output", 0) or 0) + latency_ms = getattr(response, "latency_ms", 0) or 0 + if latency_ms: + split_diagnostics[block]["latencyMs"] = round( + split_diagnostics[block].get("latencyMs", 0) + latency_ms, + 1, + ) normalized_type = (plan_type or "savings").lower() if normalized_type == "iul": @@ -901,7 +1040,7 @@ async def _llm_extract_split( "2. premium_payment_period 只输出数字（年数）\n" "3. 无法确定的字段填 null\n" "4. 只输出 JSON，无 markdown\n\n" - f"PDF 文本：\n{extraction_text}" + f"PDF 文本：\n{identity_text}" ) if progress_callback: @@ -933,7 +1072,13 @@ async def _llm_extract_split( ) if isinstance(identity_data, dict): merged_data.update(identity_data) + split_diagnostics["identity"]["status"] = "success" + record_response("identity", last_response) except Exception as e: + split_diagnostics["identity"] = { + "status": "failed", + "error": _format_exception(e), + } logger.warning(f"[SplitExtract] 身份字段提取失败: {e}") # ── 第二次调用：利益演示表（大输出，用更多 token）── @@ -961,73 +1106,37 @@ async def _llm_extract_split( '"source_page": 数字或null}]}' ) - benefit_prompt = ( + benefit_prompt_prefix = ( "从以下 PDF 文本中提取保险计划书的利益演示表。\n" "必须输出 JSON 对象，根对象只能包含 benefit_illustration 字段；禁止直接输出数组：\n" f"{benefit_shape}\n\n" "规则：\n" - "1. 扫描所有页面，提取全部保单年度\n" + "1. 提取当前分块中的全部保单年度，不要遗漏任何数据行\n" "2. 数值去逗号转数字，无法确定填 null，不要填 0\n" "3. 严禁编造数据\n" "4. 年龄和保单年度是不同列，年龄绝不能写入任何金额字段\n" "5. 只输出 JSON，无 markdown\n\n" - f"PDF 文本：\n{extraction_text}" ) if progress_callback: progress_callback(70, "正在提取利益演示表") - try: - benefit_data, benefit_resp = await llm_client.structured_output( - prompt=benefit_prompt, - system_prompt="你是保险计划书数据提取专家。只输出 JSON。", - schema={ - "type": "object", - "required": ["benefit_illustration"], - "properties": { - "benefit_illustration": { - "type": "array", - "items": {"type": "object"}, - }, - }, - "additionalProperties": False, - }, - temperature=0, + benefit_rows = [] + for chunk_index, benefit_text in enumerate(benefit_chunks, start=1): + benefit_prompt = ( + f"{benefit_prompt_prefix}" + f"当前分块：{chunk_index}/{len(benefit_chunks)}\n" + f"PDF 文本：\n{benefit_text}" ) - if isinstance(benefit_data, dict) and "benefit_illustration" in benefit_data: - merged_data["benefit_illustration"] = benefit_data["benefit_illustration"] - last_response = benefit_resp - except Exception as e: - logger.warning(f"[SplitExtract] 利益表提取失败: {e}") - - # ── 第三次调用：提领表（可选，仅储蓄险/IUL）── - if plan_type in ("savings", "iul"): - withdrawal_prompt = ( - "从以下 PDF 文本中提取保险计划书的提领/提款演示表。\n" - "如果文本中没有提领表，输出 {\"withdrawal_illustration\": []}。\n" - "必须输出 JSON 对象，根对象只能包含 withdrawal_illustration 字段；禁止直接输出数组：\n" - '{"withdrawal_illustration": [{"policy_year": 数字, "annual_withdrawal": 数字或null, ' - '"total_withdrawn": 数字或null, "surrender_value_before": 数字或null, ' - '"surrender_value_after": 数字或null, "source_page": 数字或null}]}\n\n' - "规则：\n" - "1. 只取\"总额/Total\"列，不取子列\n" - "2. 没有提领表时 withdrawal_illustration 必须为空数组 []\n" - "3. 只输出 JSON，无 markdown\n\n" - f"PDF 文本：\n{extraction_text}" - ) - - if progress_callback: - progress_callback(85, "正在提取提领表") - try: - withdrawal_data, withdrawal_resp = await llm_client.structured_output( - prompt=withdrawal_prompt, + benefit_data, benefit_resp = await llm_client.structured_output( + prompt=benefit_prompt, system_prompt="你是保险计划书数据提取专家。只输出 JSON。", schema={ "type": "object", - "required": ["withdrawal_illustration"], + "required": ["benefit_illustration"], "properties": { - "withdrawal_illustration": { + "benefit_illustration": { "type": "array", "items": {"type": "object"}, }, @@ -1036,15 +1145,117 @@ async def _llm_extract_split( }, temperature=0, ) - if isinstance(withdrawal_data, dict) and "withdrawal_illustration" in withdrawal_data: - merged_data["withdrawal_illustration"] = withdrawal_data["withdrawal_illustration"] - last_response = withdrawal_resp + chunk_rows = benefit_data.get("benefit_illustration", []) if isinstance(benefit_data, dict) else [] + if isinstance(chunk_rows, list): + benefit_rows.extend(row for row in chunk_rows if isinstance(row, dict)) + split_diagnostics["benefit"]["completedChunks"] += 1 + record_response("benefit", benefit_resp) + last_response = benefit_resp except Exception as e: - logger.warning(f"[SplitExtract] 提领表提取失败: {e}") + error_text = _format_exception(e) + split_diagnostics["benefit"]["errors"].append({ + "chunk": chunk_index, + "error": error_text, + }) + logger.warning( + f"[SplitExtract] 利益表分块 {chunk_index}/{len(benefit_chunks)} 提取失败: {error_text}" + ) + + merged_data["benefit_illustration"] = _merge_rows_by_policy_year(benefit_rows) + completed_benefit_chunks = split_diagnostics["benefit"]["completedChunks"] + if completed_benefit_chunks == len(benefit_chunks) and benefit_chunks: + split_diagnostics["benefit"]["status"] = "success" + elif completed_benefit_chunks > 0: + split_diagnostics["benefit"]["status"] = "partial" + else: + split_diagnostics["benefit"]["status"] = "failed" + split_diagnostics["benefit"]["rowCount"] = len(merged_data["benefit_illustration"]) + + # ── 第三次调用：提领表（可选，仅储蓄险/IUL）── + if normalized_type in ("savings", "iul"): + withdrawal_chunks = _select_page_chunks( + pdf_text, + ("提取", "提款", "提领", "提領", "领取", "領取", "withdrawal"), + min_keyword_hits=1, + pages_per_chunk=2, + max_chars=8000, + ) + split_diagnostics["withdrawal"].update({ + "chunkCount": len(withdrawal_chunks), + "completedChunks": 0, + "errors": [], + }) + withdrawal_rows = [] + + if progress_callback and withdrawal_chunks: + progress_callback(85, "正在提取提领表") + + for chunk_index, withdrawal_text in enumerate(withdrawal_chunks, start=1): + withdrawal_prompt = ( + "从以下 PDF 文本中提取保险计划书的提领/提款演示表。\n" + "如果文本中没有提领表，输出 {\"withdrawal_illustration\": []}。\n" + "必须输出 JSON 对象，根对象只能包含 withdrawal_illustration 字段；禁止直接输出数组：\n" + '{"withdrawal_illustration": [{"policy_year": 数字, "annual_withdrawal": 数字或null, ' + '"total_withdrawn": 数字或null, "surrender_value_before": 数字或null, ' + '"surrender_value_after": 数字或null, "source_page": 数字或null}]}\n\n' + "规则：\n" + "1. 只取\"总额/Total\"列，不取子列\n" + "2. 没有提领表时 withdrawal_illustration 必须为空数组 []\n" + "3. 只输出 JSON，无 markdown\n\n" + f"当前分块：{chunk_index}/{len(withdrawal_chunks)}\n" + f"PDF 文本：\n{withdrawal_text}" + ) + try: + withdrawal_data, withdrawal_resp = await llm_client.structured_output( + prompt=withdrawal_prompt, + system_prompt="你是保险计划书数据提取专家。只输出 JSON。", + schema={ + "type": "object", + "required": ["withdrawal_illustration"], + "properties": { + "withdrawal_illustration": { + "type": "array", + "items": {"type": "object"}, + }, + }, + "additionalProperties": False, + }, + temperature=0, + ) + chunk_rows = withdrawal_data.get("withdrawal_illustration", []) if isinstance(withdrawal_data, dict) else [] + if isinstance(chunk_rows, list): + withdrawal_rows.extend(row for row in chunk_rows if isinstance(row, dict)) + split_diagnostics["withdrawal"]["completedChunks"] += 1 + record_response("withdrawal", withdrawal_resp) + last_response = withdrawal_resp + except Exception as e: + error_text = _format_exception(e) + split_diagnostics["withdrawal"]["errors"].append({ + "chunk": chunk_index, + "error": error_text, + }) + logger.warning( + f"[SplitExtract] 提领表分块 {chunk_index}/{len(withdrawal_chunks)} 提取失败: {error_text}" + ) + + merged_data["withdrawal_illustration"] = _merge_rows_by_policy_year(withdrawal_rows) + completed_withdrawal_chunks = split_diagnostics["withdrawal"]["completedChunks"] + if not withdrawal_chunks: + split_diagnostics["withdrawal"]["status"] = "not_present" + elif completed_withdrawal_chunks == len(withdrawal_chunks): + split_diagnostics["withdrawal"]["status"] = "success" + elif completed_withdrawal_chunks > 0: + split_diagnostics["withdrawal"]["status"] = "partial" + else: + split_diagnostics["withdrawal"]["status"] = "failed" + split_diagnostics["withdrawal"]["rowCount"] = len(merged_data["withdrawal_illustration"]) + else: + split_diagnostics["withdrawal"]["status"] = "not_applicable" # 确保必要字段存在 merged_data.setdefault("benefit_illustration", []) merged_data.setdefault("withdrawal_illustration", []) + merged_data.setdefault("_meta", {})["split_extraction"] = split_diagnostics return merged_data, last_response @@ -1219,6 +1430,10 @@ class ExtractionOrchestrator: data, response = await _llm_extract_split( pdf_text, plan_type, llm_client, progress_callback, ) + split_meta = ((data.get("_meta") or {}).get("split_extraction") or {}) + extraction_stats["llm_blocks"] = { + key: value for key, value in split_meta.items() if key != "tokens" + } # 检查完整性，对缺失字段做针对性重试 initial_status, initial_error = assess_extraction_payload(data, plan_type) @@ -1237,10 +1452,9 @@ class ExtractionOrchestrator: if len(regex_benefit) > len(llm_benefit) and (quality_passed or not llm_benefit): data["benefit_illustration"] = regex_benefit - extraction_stats["llm_tokens"] = { - "input": response.tokens.get("input", 0) if response and response.tokens else 0, - "output": response.tokens.get("output", 0) if response and response.tokens else 0, - } + extraction_stats["llm_tokens"] = split_meta.get( + "tokens", {"input": 0, "output": 0} + ) if progress_callback: progress_callback(90, "AI 分块提取完成，正在校验数据") except Exception as e: @@ -1285,7 +1499,10 @@ class ExtractionOrchestrator: total_ms = (time.time() - start) * 1000 extraction_stats["total_ms"] = round(total_ms, 1) - logger.info(f"[ExtractionOrchestrator] 提取完成: {extraction_stats}") + logger.info( + f"[ExtractionOrchestrator] 提取完成 file={os.path.basename(abs_path)} " + f"status={status}: {extraction_stats}" + ) if progress_callback: progress_callback(100, "数据校验完成") @@ -1296,7 +1513,7 @@ class ExtractionOrchestrator: return ExtractionResult( pdf_path=abs_path, product_name=product_name, plan_type=detected_type, status=status, data=data, - usage={"input": response.tokens.get("input", 0), "output": response.tokens.get("output", 0)} if response and response.tokens else None, + usage=extraction_stats.get("llm_tokens"), error=extraction_error or None, duration_ms=total_ms, provenance=provenance, diff --git a/api/insurance/ppt/llm_client.py b/api/insurance/ppt/llm_client.py index a9b5f86..52d7de9 100644 --- a/api/insurance/ppt/llm_client.py +++ b/api/insurance/ppt/llm_client.py @@ -70,6 +70,14 @@ def _format_exception(exc: Exception) -> str: return f"{exc.__class__.__name__}: {message}" +def _is_retryable_llm_error(exc: Exception) -> bool: + if isinstance(exc, httpx.TransportError): + return True + if isinstance(exc, httpx.HTTPStatusError): + return exc.response.status_code in {408, 429, 500, 502, 503, 504} + return False + + def _parse_timeout_ms(value: Optional[str], default: int = DEFAULT_TIMEOUT_MS) -> int: try: timeout_ms = int(str(value or "").strip()) @@ -572,6 +580,7 @@ class LLMClient: start_idx = self._active_idx tried = set() + provider_errors = [] for i in range(len(self._configs)): idx = (start_idx + i) % len(self._configs) config, api_key = self._configs[idx] @@ -580,14 +589,6 @@ class LLMClient: continue tried.add(config.name) - # Dify 模式：通过 Dify Chat API 调用 - if config.name == "dify": - try: - return await _call_dify(config.model, messages, timeout_ms=self._timeout_ms) - except Exception as e: - logger.warning(f"[LLMClient] Dify 调用失败: {_format_exception(e)}") - continue - # 限流 limiter = self._limiters.get(config.name) if limiter: @@ -596,24 +597,54 @@ class LLMClient: logger.warning(f"[LLMClient] {config.name} 限流超时") continue - # 调用 - try: - response = await _call_provider( - config, - api_key, - messages, - timeout_ms=self._timeout_ms, - json_mode=json_mode, - temperature=temperature, - ) - self._active_idx = idx - return response - except Exception as e: - logger.warning(f"[LLMClient] {config.name} 失败: {_format_exception(e)}") - if attempt < 3 and i < len(self._configs) - 1: - self._active_idx = (idx + 1) % len(self._configs) + max_attempts = max(1, int(config.max_retries) + 1) + for retry_index in range(max_attempts): + try: + if config.name == "dify": + response = await _call_dify( + config.model, messages, timeout_ms=self._timeout_ms + ) + else: + response = await _call_provider( + config, + api_key, + messages, + timeout_ms=self._timeout_ms, + json_mode=json_mode, + temperature=temperature, + ) + self._active_idx = idx + return response + except Exception as e: + error_text = _format_exception(e) + provider_errors.append(f"{config.name}: {error_text}") + should_retry = ( + _is_retryable_llm_error(e) + and retry_index + 1 < max_attempts + ) + if should_retry: + delay = min(2 ** retry_index, 4) + logger.warning( + f"[LLMClient] {config.name} 传输失败，" + f"{delay}s 后重试 {retry_index + 2}/{max_attempts}: {error_text}" + ) + await asyncio.sleep(delay) + continue + logger.warning( + f"[LLMClient] {config.name} 失败 " + f"attempt={retry_index + 1}/{max_attempts}: {error_text}" + ) + break - raise RuntimeError(f"所有 LLM 供应商均失败，请检查 API Key、Base URL、模型名称或超时设置（当前 {self._timeout_ms}ms）") + if attempt < 3 and i < len(self._configs) - 1: + self._active_idx = (idx + 1) % len(self._configs) + + provider_count = len(self._configs) + last_error = provider_errors[-1] if provider_errors else "未知错误" + raise RuntimeError( + f"已配置的 {provider_count} 个 LLM 供应商均调用失败" + f"（当前超时 {self._timeout_ms}ms，最后错误: {last_error}）" + ) def get_status(self) -> dict: """获取当前供应商信息。""" diff --git a/tests/ppt_poster_optimization_test.py b/tests/ppt_poster_optimization_test.py index af782e3..233b1ff 100644 --- a/tests/ppt_poster_optimization_test.py +++ b/tests/ppt_poster_optimization_test.py @@ -412,7 +412,12 @@ def test_iul_split_fallback_requests_iul_specific_fields(): return {"benefit_illustration": []}, SimpleNamespace(tokens={}) return {"withdrawal_illustration": []}, SimpleNamespace(tokens={}) - asyncio.run(_llm_extract_split("[PAGE 1]\nIUL 计划书", "iul", FakeClient())) + asyncio.run(_llm_extract_split( + "[PAGE 1]\nIUL 计划书\n\n" + "[PAGE 2]\n提领方案 保单年度 提取金额 退保价值", + "iul", + FakeClient(), + )) assert "index_accounts" in prompts[0] assert "guaranteed_account_value" in prompts[1] @@ -422,6 +427,108 @@ def test_iul_split_fallback_requests_iul_specific_fields(): assert "根对象只能包含 withdrawal_illustration 字段" in prompts[2] +def test_split_extraction_chunks_benefit_pages_and_merges_rows(): + import asyncio + import re + from types import SimpleNamespace + from insurance.ppt.extraction import _llm_extract_split + + prompts = [] + + class FakeClient: + async def structured_output(self, **kwargs): + prompt = kwargs["prompt"] + prompts.append(prompt) + response = SimpleNamespace(tokens={"input": 10, "output": 5}, latency_ms=10) + if "身份和保单字段" in prompt: + return { + "product_name": "储蓄计划", + "insured": {"age": 35, "gender": "female"}, + "policy": {"currency": "USD", "sum_insured": None, + "annual_premium": 10000, "premium_payment_period": 5}, + }, response + page_numbers = [int(value) for value in re.findall(r"\[PAGE (\d+)\]", prompt)] + return { + "benefit_illustration": [ + {"policy_year": page, "total_surrender_value": page * 1000, "source_page": page} + for page in page_numbers + ], + }, response + + pdf_text = "[PAGE 1]\n产品名称：储蓄计划\n受保人年龄：35\n" + pdf_text += "\n\n".join( + f"[PAGE {page}]\n保单年度 保证现金价值 退保价值\n{page} 100 {page * 1000}" + for page in range(2, 7) + ) + + data, _response = asyncio.run(_llm_extract_split(pdf_text, "savings", FakeClient())) + + benefit_prompts = [ + prompt for prompt in prompts + if "利益演示表" in prompt and "当前分块" in prompt + ] + assert len(benefit_prompts) == 3 + assert [row["policy_year"] for row in data["benefit_illustration"]] == [2, 3, 4, 5, 6] + assert data["_meta"]["split_extraction"]["benefit"]["status"] == "success" + assert data["_meta"]["split_extraction"]["benefit"]["chunkCount"] == 3 + + +def test_incomplete_savings_benefit_rows_are_partial(): + from insurance.ppt.extraction import assess_extraction_payload + + status, message = assess_extraction_payload({ + "product_name": "储蓄计划", + "insured": {"age": 35}, + "policy": {"annual_premium": 10000}, + "benefit_illustration": [ + {"policy_year": year, "total_surrender_value": year * 1000} + for year in (1, 10, 20) + ], + "_meta": { + "split_extraction": { + "benefit": {"status": "failed", "errors": ["RemoteProtocolError"]}, + }, + }, + }, "savings") + + assert status == "partial" + assert "利益演示数据不足" in message + + +def test_llm_client_retries_retryable_transport_error(monkeypatch): + import asyncio + import httpx + from insurance.ppt import llm_client as llm_module + + client = llm_module.LLMClient.__new__(llm_module.LLMClient) + config = llm_module.LLMProviderConfig( + name="deepseek", base_url="https://example.com/v1", model="model", max_retries=2, + ) + client._configs = [(config, "secret")] + client._limiters = {} + client._active_idx = 0 + client._timeout_ms = 180000 + client._try_load_db_config = lambda: None + attempts = [] + + async def fake_provider(*_args, **_kwargs): + attempts.append(1) + if len(attempts) == 1: + raise httpx.RemoteProtocolError("incomplete chunked read") + return llm_module.LLMResponse(content='{"ok": true}', provider="deepseek") + + async def no_wait(_seconds): + return None + + monkeypatch.setattr(llm_module, "_call_provider", fake_provider) + monkeypatch.setattr(llm_module.asyncio, "sleep", no_wait) + + response = asyncio.run(client._call([{"role": "user", "content": "test"}], json_mode=True)) + + assert response.provider == "deepseek" + assert len(attempts) == 2 + + def test_single_required_array_schema_normalizes_safe_equivalent_roots(): from insurance.ppt.llm_client import _normalize_schema_root