diff --git a/docs/en/guides/01-configuration.md b/docs/en/guides/01-configuration.md index f27398883c..ad10a71d6a 100644 --- a/docs/en/guides/01-configuration.md +++ b/docs/en/guides/01-configuration.md @@ -882,6 +882,34 @@ Use `github_domains`, `gitlab_domains`, or `azure_devops_domains` when the host needs those platform-specific URL semantics. Add other Git hosts to `code_hosting_domains`. +### pdf + +PDF parsing configuration. Three strategies are supported: `local` (local pdfplumber), `mineru` (remote MinerU API), and `auto` (try local first, fall back to MinerU). + +```json +{ + "pdf": { + "strategy": "auto", + "mineru_endpoint": "http://127.0.0.1:8000", + "mineru_timeout": 300.0, + "mineru_bodys": { + "backend": "hybrid-auto-engine", + "lang_list": ["ch"], + "parse_method": "auto" + } + } +} +``` + +| Parameter | Type | Description | +|-----------|------|-------------| +| `strategy` | str | Parsing strategy: `local` / `mineru` / `auto` (default `auto`) | +| `mineru_endpoint` | str | MinerU API **base URL** (e.g. `http://127.0.0.1:8000`) | +| `mineru_timeout` | float | Request timeout in seconds (default `300.0`) | +| `mineru_bodys` | dict | MinerU API multipart form fields | + +**MinerU protocol**: a synchronous `POST {mineru_endpoint}/file_parse` request with the PDF as the multipart `files` field; form parameters are passed through from `mineru_bodys`. + ### rerank Reranking model for search result refinement. Supports VikingDB (Volcengine), Cohere, and OpenAI-compatible APIs. diff --git a/docs/zh/guides/01-configuration.md b/docs/zh/guides/01-configuration.md index dd41165361..a3ada0f23a 100644 --- a/docs/zh/guides/01-configuration.md +++ b/docs/zh/guides/01-configuration.md @@ -850,6 +850,34 @@ ollama pull guoxuter/ov_intent_analysis_sft:v7_q8 需要 GitHub、GitLab 或 Azure DevOps 专属 URL 语义时,应配置到对应的平台字段; 其他 Git 主机统一添加到 `code_hosting_domains`。 +### pdf + +PDF 解析配置。支持三种策略:`local`(本地 pdfplumber)、`mineru`(远程 MinerU API)、`auto`(先本地、失败回退 MinerU)。 + +```json +{ + "pdf": { + "strategy": "auto", + "mineru_endpoint": "http://127.0.0.1:8000", + "mineru_timeout": 300.0, + "mineru_bodys": { + "backend": "hybrid-auto-engine", + "lang_list": ["ch"], + "parse_method": "auto" + } + } +} +``` + +| 参数 | 类型 | 说明 | +|------|------|------| +| `strategy` | str | 解析策略:`local` / `mineru` / `auto`(默认 `auto`) | +| `mineru_endpoint` | str | MinerU API **base URL**(如 `http://127.0.0.1:8000`) | +| `mineru_timeout` | float | 请求超时秒数(默认 `300.0`) | +| `mineru_bodys` | dict | MinerU API multipart form 参数 | + +**MinerU 协议**:同步调用 `POST {mineru_endpoint}/file_parse`,multipart 文件字段为 `files`,form 参数由 `mineru_bodys` 透传。 + ### rerank 用于搜索结果精排的 Rerank 模型。支持 VikingDB (火山引擎)、Cohere 和 OpenAI 兼容接口。 diff --git a/examples/ov.conf.example b/examples/ov.conf.example index 9056a72865..f1f0b2ac0b 100644 --- a/examples/ov.conf.example +++ b/examples/ov.conf.example @@ -228,9 +228,13 @@ "max_section_size": 4000, "section_size_flexibility": 0.3, "max_section_chars": 6000, - "mineru_endpoint": "https://mineru.example.com/api/v1", - "mineru_api_key": "{your-mineru-api-key}", + "mineru_endpoint": "http://127.0.0.1:8000", "mineru_timeout": 300.0, + "mineru_bodys": { + "backend": "hybrid-auto-engine", + "lang_list": ["ch"], + "parse_method": "auto" + }, }, "code": { "github_raw_domain": "raw.githubusercontent.com", diff --git a/openviking/parse/parsers/pdf.py b/openviking/parse/parsers/pdf.py index 4d7acca204..f299a0680d 100644 --- a/openviking/parse/parsers/pdf.py +++ b/openviking/parse/parsers/pdf.py @@ -13,6 +13,7 @@ """ import asyncio +import base64 import hashlib import io import re @@ -29,6 +30,7 @@ lazy_import, ) from openviking.parse.parsers.base_parser import BaseParser +from openviking.utils.time_utils import parse_iso_datetime from openviking_cli.utils import get_logger from openviking_cli.utils.config.parser_config import PDFConfig @@ -57,8 +59,7 @@ class PDFParser(BaseParser): >>> # Remote API parsing >>> config = PDFConfig( ... strategy="mineru", - ... mineru_endpoint="https://api.example.com/convert", - ... mineru_api_key="key" + ... mineru_endpoint="http://127.0.0.1:8000" ... ) >>> parser = PDFParser(config) >>> result = await parser.parse("document.pdf") @@ -682,27 +683,38 @@ def _extract_image_from_page(self, page, img_info: dict) -> Optional[bytes]: async def _convert_mineru( self, pdf_path: Path, + storage=None, resource_name: Optional[str] = None, ) -> tuple[str, Dict[str, Any]]: """ - Convert PDF to Markdown using MinerU API. + Convert PDF to Markdown using the MinerU /file_parse API. Args: pdf_path: Path to PDF file - resource_name: Optional resource name (unused in MinerU conversion) + storage: Media storage used to persist extracted images; defaults to + the configured storage if None + resource_name: Resource name under which extracted images are saved; + defaults to the PDF stem Returns: - Tuple of (markdown_content, metadata) + Tuple of (markdown_content, metadata) where metadata includes + strategy, endpoint, api_version, backend, task_id, processing_time + (seconds) and images_saved Raises: - ImportError: If httpx not installed - Exception: If API call fails + ValueError: If MinerU endpoint is not configured, or the task does + not complete + Exception: If the API call fails """ httpx = lazy_import("httpx") if not self.config.mineru_endpoint: raise ValueError("MinerU endpoint not configured") + # mineru_endpoint is a base URL; append /file_parse unless it already ends with it + base = self.config.mineru_endpoint.rstrip("/") + url = base if base.endswith("/file_parse") else f"{base}/file_parse" + meta = { "strategy": "mineru", "endpoint": self.config.mineru_endpoint, @@ -713,44 +725,86 @@ async def _convert_mineru( async with httpx.AsyncClient(timeout=self.config.mineru_timeout) as client: # Prepare file upload with open(pdf_path, "rb") as f: - files = {"file": (pdf_path.name, f, "application/pdf")} + files = {"files": (pdf_path.name, f, "application/pdf")} - # Prepare headers - headers = {} - if self.config.mineru_api_key: - headers["Authorization"] = f"Bearer {self.config.mineru_api_key}" + # Prepare Form fields + data: Dict[str, Any] = dict(self.config.mineru_bodys or {}) - # Prepare request params - params = self.config.mineru_params or {} + # MinerU must return extracted images for the markdown refs. + data["return_images"] = True # Make API request - logger.info(f"Calling MinerU API: {self.config.mineru_endpoint}") + logger.info(f"Calling MinerU API: {url}") response = await client.post( - self.config.mineru_endpoint, + url, files=files, - headers=headers, - params=params, + data=data, ) response.raise_for_status() - # Parse response - result = response.json() - markdown_content = result.get("markdown", "") + # Parse response + result = response.json() + if result.get("status") != "completed": + raise ValueError(f"MinerU task not completed: {result.get('status')}") + + results = result.get("results") or {} + file_result = results.get(pdf_path.name) or next(iter(results.values()), {}) + markdown_content = file_result.get("md_content") or "" + + # Extract metadata from response + meta["api_version"] = result.get("version") + meta["backend"] = result.get("backend") + meta["task_id"] = result.get("task_id") + started_at, completed_at = result.get("started_at"), result.get("completed_at") + if started_at and completed_at: + meta["processing_time"] = ( + parse_iso_datetime(completed_at) - parse_iso_datetime(started_at) + ).total_seconds() + + if not markdown_content: + logger.warning(f"MinerU returned empty content for {pdf_path}") + return markdown_content, meta - # Extract metadata from response - meta["api_version"] = result.get("version") - meta["processing_time"] = result.get("processing_time") - meta["total_pages"] = result.get("total_pages") + if storage is None: + from openviking_cli.utils.storage import get_storage - if not markdown_content: - logger.warning(f"MinerU returned empty content for {pdf_path}") + storage = get_storage() - logger.info( - f"MinerU conversion: {meta.get('total_pages', '?')} pages → " - f"{len(markdown_content)} chars" + if resource_name is None: + resource_name = pdf_path.stem + + # MinerU embeds images as base64 data-URLs, referenced from markdown + # as `images/`; save them into the media store and rewrite + # the references to the stored relative paths. + repl: Dict[str, str] = {} + media_dir = storage.media_dir + for img_name, data_url in (file_result.get("images") or {}).items(): + try: + # data URL form: "data:image/jpeg;base64," + image_bytes = base64.b64decode(data_url.split(",", 1)[-1]) + img_path = Path(img_name) + image_path = storage.save_image( + resource_name, + image_bytes, + filename=img_path.stem, + extension=img_path.suffix or ".png", + ) + repl[f"images/{img_name}"] = image_path.relative_to(media_dir).as_posix() + except Exception as img_err: + logger.warning(f"Failed to save MinerU image {img_name}: {img_err}") + + if repl: + # Single pass over the markdown replaces every known reference; + # unknown ``images/...`` text is left untouched. + ref_pattern = re.compile( + "|".join(re.escape(ref) for ref in sorted(repl, key=len, reverse=True)) ) + markdown_content = ref_pattern.sub(lambda m: repl[m.group(0)], markdown_content) + meta["images_saved"] = len(repl) - return markdown_content, meta + logger.info(f"MinerU conversion: {len(markdown_content)} chars") + + return markdown_content, meta except Exception as e: logger.error(f"MinerU API call failed: {e}") diff --git a/openviking/service/core.py b/openviking/service/core.py index 8a4abf6b0d..039d801f50 100644 --- a/openviking/service/core.py +++ b/openviking/service/core.py @@ -19,6 +19,7 @@ from openviking.service.agent_evolution_service import AgentEvolutionService from openviking.service.debug_service import DebugService from openviking.service.fs_service import FSService +from openviking.service.mineru_preflight import wait_for_mineru_ready from openviking.service.pack_service import PackService from openviking.service.relation_service import RelationService from openviking.service.resource_memory_link_service import ResourceMemoryLinkService @@ -498,6 +499,25 @@ async def initialize(self) -> None: self._queue_manager.start() logger.info("QueueManager workers started") + # Preflight the MinerU endpoint when it will be used, so endpoint + # misconfiguration or a stopped service surfaces now instead of on the + # first PDF import. Required for strategy="mineru"; advisory for "auto". + pdf_config = self._config.pdf + should_preflight_mineru = pdf_config.strategy == "mineru" or ( + pdf_config.strategy == "auto" and pdf_config.mineru_endpoint is not None + ) + + if should_preflight_mineru and pdf_config.mineru_endpoint: + try: + await wait_for_mineru_ready(pdf_config.mineru_endpoint) + logger.info("MinerU preflight passed: %s", pdf_config.mineru_endpoint) + except RuntimeError as exc: + if pdf_config.strategy == "mineru": + raise + logger.warning( + "MinerU preflight failed (fallback will retry on first parse): %s", exc + ) + self._initialized = True logger.info("OpenVikingService initialized") diff --git a/openviking/service/mineru_preflight.py b/openviking/service/mineru_preflight.py new file mode 100644 index 0000000000..1b12693e82 --- /dev/null +++ b/openviking/service/mineru_preflight.py @@ -0,0 +1,68 @@ +# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd. +# SPDX-License-Identifier: AGPL-3.0 +""" +Startup preflight for the MinerU API. + +Polls the MinerU ``/health`` endpoint so that endpoint misconfiguration, a +stopped service, or a protocol incompatibility is surfaced at service +initialization instead of on the first PDF import. +""" + +import asyncio +import time + +import httpx + +HEALTH_TIMEOUT = 5.0 # Total polling timeout in seconds. +HEALTH_POLL_INTERVAL = 0.2 # Delay between polls in seconds. +HEALTH_REQUEST_TIMEOUT = 0.5 # Per-request timeout in seconds. + + +async def wait_for_mineru_ready(endpoint: str, timeout: float = HEALTH_TIMEOUT) -> None: + """ + Poll the MinerU ``/health`` endpoint until it reports ready. + + A healthy response is HTTP 200 with a JSON object carrying a non-empty + ``protocol_version`` and ``status == "healthy"``. + + Args: + endpoint: MinerU base URL (e.g. ``http://127.0.0.1:8000``). + timeout: Total time to poll before giving up, in seconds. + + Raises: + RuntimeError: If the endpoint does not report ready within ``timeout`` + seconds. The message includes the health URL and the last error. + """ + health_url = f"{endpoint.rstrip('/')}/health" + deadline = time.monotonic() + timeout + last_error = "no response" + + async with httpx.AsyncClient(timeout=HEALTH_REQUEST_TIMEOUT) as client: + while time.monotonic() < deadline: + try: + response = await client.get(health_url) + payload = response.json() + except (httpx.HTTPError, ValueError) as exc: + last_error = str(exc) or exc.__class__.__name__ + else: + if isinstance(payload, dict): + protocol = payload.get("protocol_version") + status = payload.get("status") + else: + protocol = status = None + + if response.status_code == 200 and protocol and status == "healthy": + return + + last_error = ( + f"unexpected response: HTTP {response.status_code}, " + f"protocol_version={protocol!r}, status={status!r}" + ) + + await asyncio.sleep(HEALTH_POLL_INTERVAL) + + raise RuntimeError( + f"MinerU startup preflight failed for {health_url} after {timeout}s. " + f"Last error: {last_error}. " + "Start a compatible mineru-api service or correct pdf.mineru_endpoint." + ) diff --git a/openviking_cli/utils/config/parser_config.py b/openviking_cli/utils/config/parser_config.py index 83fd271c20..1ac59cef63 100644 --- a/openviking_cli/utils/config/parser_config.py +++ b/openviking_cli/utils/config/parser_config.py @@ -148,18 +148,16 @@ class PDFConfig(ParserConfig): Attributes: strategy: Parsing strategy ("local" | "mineru" | "auto") mineru_endpoint: MinerU API endpoint URL - mineru_api_key: MinerU API authentication key mineru_timeout: MinerU request timeout in seconds - mineru_params: Additional MinerU API parameters + mineru_bodys: Additional MinerU API multipart form fields """ strategy: str = "auto" # "local" | "mineru" | "auto" # MinerU API configuration mineru_endpoint: Optional[str] = None # API endpoint URL - mineru_api_key: Optional[str] = None # API authentication key mineru_timeout: float = 300.0 # Request timeout in seconds (5 minutes) - mineru_params: Optional[dict] = None # Additional API parameters + mineru_bodys: Optional[dict] = None # Additional API multipart form fields # Heading detection configuration heading_detection: str = "auto" # "bookmarks" | "font" | "auto" | "none" diff --git a/tests/unit/test_mineru_preflight.py b/tests/unit/test_mineru_preflight.py new file mode 100644 index 0000000000..0e7d081520 --- /dev/null +++ b/tests/unit/test_mineru_preflight.py @@ -0,0 +1,117 @@ +# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd. +# SPDX-License-Identifier: AGPL-3.0 +"""Unit tests for the MinerU startup preflight (wait_for_mineru_ready).""" + +import httpx +import pytest + +from openviking.service.mineru_preflight import wait_for_mineru_ready + +ENDPOINT = "http://127.0.0.1:8000" +HEALTH_URL = "http://127.0.0.1:8000/health" + + +class _FakeResponse: + def __init__(self, status_code=200, payload=None, json_error=None): + self.status_code = status_code + self._payload = payload + self._json_error = json_error + + def json(self): + if self._json_error is not None: + raise self._json_error + return self._payload + + +def _make_fake_client(get_handler): + class _FakeClient: + def __init__(self, **kwargs): + self.kwargs = kwargs + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def get(self, url): + return get_handler(url) + + return _FakeClient + + +@pytest.fixture +def patch_httpx(monkeypatch): + def _apply(get_handler): + monkeypatch.setattr(httpx, "AsyncClient", _make_fake_client(get_handler)) + + return _apply + + +@pytest.mark.asyncio +async def test_ready_immediately(patch_httpx): + patch_httpx(lambda url: _FakeResponse(payload={"protocol_version": 2, "status": "healthy"})) + + # Should return without raising. + await wait_for_mineru_ready(ENDPOINT) + + +@pytest.mark.asyncio +async def test_ready_after_unhealthy(patch_httpx): + calls = {"count": 0} + + def handler(url): + calls["count"] += 1 + if calls["count"] == 1: + return _FakeResponse(status_code=503, payload={"status": "starting"}) + return _FakeResponse(payload={"protocol_version": 2, "status": "healthy"}) + + patch_httpx(handler) + + await wait_for_mineru_ready(ENDPOINT) + assert calls["count"] == 2 + + +@pytest.mark.asyncio +async def test_timeout_raises(patch_httpx): + patch_httpx( + lambda url: _FakeResponse(payload={"protocol_version": 2, "status": "unhealthy"}) + ) + + with pytest.raises(RuntimeError) as exc_info: + await wait_for_mineru_ready(ENDPOINT, timeout=0.05) + + message = str(exc_info.value) + assert HEALTH_URL in message + assert ENDPOINT in message + assert "status='unhealthy'" in message + + +@pytest.mark.asyncio +async def test_network_error_raises(patch_httpx): + def handler(url): + raise httpx.ConnectError("connection refused") + + patch_httpx(handler) + + with pytest.raises(RuntimeError) as exc_info: + await wait_for_mineru_ready(ENDPOINT, timeout=0.05) + + message = str(exc_info.value) + assert HEALTH_URL in message + assert "connection refused" in message + + +@pytest.mark.asyncio +async def test_empty_error_message_falls_back_to_class_name(patch_httpx): + def handler(url): + raise httpx.ConnectTimeout("") + + patch_httpx(handler) + + with pytest.raises(RuntimeError) as exc_info: + await wait_for_mineru_ready(ENDPOINT, timeout=0.05) + + message = str(exc_info.value) + assert HEALTH_URL in message + assert "ConnectTimeout" in message