diff --git a/README.md b/README.md
index 81eacc358..2b25fded4 100644
--- a/README.md
+++ b/README.md
@@ -1,5 +1,3 @@
-
-
Prepare unstructured data for AI Agents
@@ -45,42 +43,33 @@ Knowhere is the open-source infrastructure for unstructured data processing. It
## How it Works
> [!TIP]
-> **TL;DR**: Knowhere parses documents into structured units, maps them in a graph, and lets agents navigate that context to find and cite reliable evidence.
+> **TL;DR**: Knowhere builds navigable memory from messy documents, then lets agents retrieve and cite evidence from that memory.
-Knowhere turns raw documents into a structured memory store that AI agents can navigate and cite. The process follows a three-stage pipeline:
+Knowhere turns raw documents into a structured memory store that AI agents can navigate and cite. The process follows two steps:
-```mermaid
-flowchart LR
- A[📄 Document Parsing] --> B[🕸️ Graph Construction]
- B --> C[🤖 Agentic Retrieval]
- B --> D[🔍 Vector-based RAG]
- C --> E[✅ Cited Results]
- D --> E
-```
+### Step 1: Parse and Build Memory
+
+
+ 
+
+
+Parsing, chunking, hierarchy extraction, and graph construction are unified into one outcome: a navigable memory layer for AI agents.
+
+- **Parse**: Route PDFs, Office files, images, tables, Markdown, and text to specialized parsers.
+- **Structure**: Preserve headings, section paths, multi-modal assets, and chunk relationships.
+- **Build Memory**: Store chunks, navigation trees, summaries, and graph links as agent-ready context.
+
+### Step 2: Agentic Retrieval
+
+
+ 
+
+
+Agents retrieve by navigating memory instead of depending on a single flat vector lookup.
-### 1. Document Parsing
-Knowhere routes files to specialized parsers for PDFs, Office docs, images, and more. We don't just extract text; we preserve the document's hierarchy:
-- **Hierarchical Paths**: Every chunk knows its exact location (e.g., `Section 2.1 > Table 4`).
-- **Multi-modal Units**: Tables and images are treated as distinct assets with their own metadata.
-- **Structural Awareness**: Heading levels and section boundaries are maintained to keep context intact.
-
-### 2. Memory Graph
-Parsed content is organized into a lightweight graph. It’s designed as a practical map for agents, not a complex ontology.
-- **Nodes**: Represent documents, sections, and chunks.
-- **Edges**: Map semantic relationships (keyword overlap, summaries) and structural links.
-This graph helps agents quickly understand what a document is about and which neighboring files might be relevant.
-
-### 3a. Agentic Retrieval
-An agent navigates the memory graph to find evidence rather than relying on a single vector lookup:
-- **Hybrid Discovery**: Fuses keyword and semantic search (RRF) for broad first-pass coverage.
-- **Agent Navigation**: The agent "walks" the graph, reviewing section previews to drill down into the most relevant paths.
-- **Cited Evidence**: Results are returned as traceable evidence — source document, section, chunk, and any linked image or table assets.
-
-### 3b. Vector-based RAG
-For teams that prefer a pure retrieval pipeline without agent overhead, Knowhere's parsed chunks plug directly into standard vector stacks:
-- **Dense Search**: Chunk embeddings stored in Qdrant, pgvector, or Milvus for fast ANN lookup.
-- **Sparse Search**: BM25 term index for keyword-sensitive queries.
-- **Multi-channel Fusion**: Dense and sparse results are fused with RRF before being returned, giving you the best of both signals.
+- **Discover**: Fuse keyword, path, content, and semantic signals for broad first-pass coverage.
+- **Navigate**: Walk section trees and graph links to drill into the most relevant document regions.
+- **Cite Evidence**: Return traceable results with source document, section, chunk, and linked image or table assets.
## Ecosystem
@@ -140,8 +129,16 @@ cp apps/worker/.env.example apps/worker/.env
- database and Redis connection settings
- S3-compatible storage credentials
-- `DS_KEY`
-- any optional LLM, billing, or webhook providers you want to enable
+- at least one LLM provider key: `DS_KEY`, `ALI_API_KEYS`, `GPT_API_KEY`, or `GLM_API_KEY`
+- `MINERU_API_KEYS` if you need PDF parsing
+- a vision-capable model provider if you need image summaries, OCR, atlas classification, or image-aware retrieval
+- any optional billing or webhook providers you want to enable
+
+Most parser and retrieval tuning values have code defaults. Start with the
+required external services first, then override model names, provider URLs,
+budgets, or concurrency limits only when your deployment needs different
+behavior. See [docs/external-services.md](docs/external-services.md) for the
+full dependency matrix.
4. Start the local infrastructure stack:
diff --git a/apps/api/.env.example b/apps/api/.env.example
index 7f51ce287..bee1948c7 100644
--- a/apps/api/.env.example
+++ b/apps/api/.env.example
@@ -66,37 +66,26 @@ OSS_ENDPOINT=
OSS_EVENT_CALLBACK_KEY=
OSS_EVENT_VERIFY_SIGNATURE=true
-# Required for local startup: one OpenAI-compatible provider
-DS_KEY=replace-with-your-deepseek-key
-DS_URL=https://api.deepseek.com/v1
+# Required for parsing/retrieval LLM calls: configure at least one provider key.
+# Provider URLs and model names have code defaults; override them only when needed.
+DS_KEY=
GPT_API_KEY=
GLM_API_KEY=
-GLM_URL=https://open.bigmodel.cn/api/paas/v4
ALI_API_KEYS=
-ALI_TOKEN_RPM_LIMIT=300
-ALI_TOKEN_DAILY_LIMIT=10000
-ALI_TOKEN_COOLDOWN_SECONDS=60
-ALI_INLINE_MAX_RETRIES=3
-ALI_SDK_MAX_RETRIES=3
-ALI_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
ARK_API_KEY=
-ARK_URL=https://ark.cn-beijing.volces.com/api/v3/chat/completions
-NORMOL_MODEL=deepseek-chat
-HIERARCHY_LLM_MODEL=qwen3.6-flash
-IMAGE_MODEL=qwen3.5-flash
-IMAGE_MODEL_MAX=qwen3.5-flash
-RETRIEVAL_DECOMPOSITION_ENABLED=false
-RETRIEVAL_PLANNER_MODEL=
-RETRIEVAL_PLANNER_THINKING_BUDGET=4000
-RETRIEVAL_DECOMPOSITION_MAX_STEPS=5
-RETRIEVAL_WALLET_TOTAL_BUDGET=200000
-RETRIEVAL_WALLET_PER_RETRIEVE_STEP_BUDGET=40000
-RETRIEVAL_WALLET_PER_SYNTHESIZE_STEP_BUDGET=6000
-RETRIEVAL_WORKFLOW_PARALLEL_MAX=3
-# Agentic retrieval (LLM-driven hierarchical navigation).
-# Set to false to fall back to legacy 3-channel RRF mode.
-RETRIEVAL_AGENTIC_ENABLED=true
+# Optional overrides:
+# DS_URL=https://api.deepseek.com/v1
+# GLM_URL=https://open.bigmodel.cn/api/paas/v4
+# ALI_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
+# ARK_URL=https://ark.cn-beijing.volces.com/api/v3/chat/completions
+# NORMOL_MODEL=deepseek-chat
+# HIERARCHY_LLM_MODEL=
+# IMAGE_MODEL=qwen3.5-flash
+# IMAGE_MODEL_MAX=qwen3.5-flash
+
+# Optional retrieval overrides have code defaults. Set RETRIEVAL_AGENTIC_ENABLED=false
+# only when you need to fall back to legacy 3-channel RRF mode.
# File handling defaults
SUPPORTED_EXTENSIONS=.doc,.docx,.pdf,.txt,.xls,.xlsx,.csv,.pptx,.jpg,.jpeg,.png,.md
@@ -120,16 +109,16 @@ FRONTEND_URL=http://localhost:3000
# Required for specific features: parsing providers
MINERU_API_KEYS=
-MINERU_URL=https://mineru.net/api/v4
-MINERU_TOKEN_RPM_LIMIT=300
-MINERU_TOKEN_DAILY_LIMIT=10000
-MINERU_TOKEN_COOLDOWN_SECONDS=60
-FORCE_MINERU_UPLOAD_ENABLED=false
ILOVEAPI_PUBLIC_KEY=
ILOVEAPI_SECRET_KEY=
-ILOVEAPI_BASE_URL=https://api.ilovepdf.com/v1
-ILOVEAPI_TIMEOUT=120
+# MINERU_URL=https://mineru.net/api/v4
+# MINERU_TOKEN_RPM_LIMIT=300
+# MINERU_TOKEN_DAILY_LIMIT=10000
+# MINERU_TOKEN_COOLDOWN_SECONDS=60
+# FORCE_MINERU_UPLOAD_ENABLED=false
+# ILOVEAPI_BASE_URL=https://api.ilovepdf.com/v1
+# ILOVEAPI_TIMEOUT=120
# Legacy parser compatibility fields.
-ALL_DF_COLS=content,path,type,length,keywords,summary,know_id,tokens,connectto,addtime,page_nums
-SPLIT_CHAR=/
+# ALL_DF_COLS=content,path,type,length,keywords,summary,know_id,tokens,connectto,addtime,page_nums
+# SPLIT_CHAR=/
diff --git a/apps/worker/.env.example b/apps/worker/.env.example
index a57b458fc..82cf92088 100644
--- a/apps/worker/.env.example
+++ b/apps/worker/.env.example
@@ -71,32 +71,26 @@ QSTASH_MAX_RETRIES=5
# QSTASH_CURRENT_SIGNING_KEY=
# QSTASH_NEXT_SIGNING_KEY=
-# Required for local startup: one OpenAI-compatible provider
-DS_KEY=replace-with-your-deepseek-key
-DS_URL=https://api.deepseek.com/v1
+# Required for parsing/retrieval LLM calls: configure at least one provider key.
+# Provider URLs and model names have code defaults; override them only when needed.
+DS_KEY=
GPT_API_KEY=
GLM_API_KEY=
-GLM_URL=https://open.bigmodel.cn/api/paas/v4
ALI_API_KEYS=
-ALI_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
ARK_API_KEY=
-ARK_URL=https://ark.cn-beijing.volces.com/api/v3/chat/completions
-NORMOL_MODEL=deepseek-chat
-HIERARCHY_LLM_MODEL=deepseek-chat
-IMAGE_MODEL=qwen3.5-flash
-IMAGE_MODEL_MAX=qwen3.5-flash
-RETRIEVAL_DECOMPOSITION_ENABLED=false
-RETRIEVAL_PLANNER_MODEL=
-RETRIEVAL_PLANNER_THINKING_BUDGET=4000
-RETRIEVAL_DECOMPOSITION_MAX_STEPS=5
-RETRIEVAL_WALLET_TOTAL_BUDGET=200000
-RETRIEVAL_WALLET_PER_RETRIEVE_STEP_BUDGET=40000
-RETRIEVAL_WALLET_PER_SYNTHESIZE_STEP_BUDGET=6000
-RETRIEVAL_WORKFLOW_PARALLEL_MAX=3
-# Agentic retrieval (LLM-driven hierarchical navigation).
-# Set to false to fall back to legacy 3-channel RRF mode.
-RETRIEVAL_AGENTIC_ENABLED=true
+# Optional overrides:
+# DS_URL=https://api.deepseek.com/v1
+# GLM_URL=https://open.bigmodel.cn/api/paas/v4
+# ALI_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
+# ARK_URL=https://ark.cn-beijing.volces.com/api/v3/chat/completions
+# NORMOL_MODEL=deepseek-chat
+# HIERARCHY_LLM_MODEL=
+# IMAGE_MODEL=qwen3.5-flash
+# IMAGE_MODEL_MAX=qwen3.5-flash
+
+# Optional retrieval overrides have code defaults. Set RETRIEVAL_AGENTIC_ENABLED=false
+# only when you need to fall back to legacy 3-channel RRF mode.
# Required for specific features: billing and analytics
BILLING_ENABLED=false
@@ -106,21 +100,21 @@ MOESIF_APPLICATION_ID=
# Required for specific features: parsing providers
MINERU_API_KEYS=
-MINERU_URL=https://mineru.net/api/v4
-MINERU_TOKEN_RPM_LIMIT=300
-MINERU_TOKEN_DAILY_LIMIT=10000
-MINERU_TOKEN_COOLDOWN_SECONDS=60
-FORCE_MINERU_UPLOAD_ENABLED=false
ILOVEAPI_PUBLIC_KEY=
ILOVEAPI_SECRET_KEY=
-ILOVEAPI_BASE_URL=https://api.ilovepdf.com/v1
-ILOVEAPI_TIMEOUT=120
+# MINERU_URL=https://mineru.net/api/v4
+# MINERU_TOKEN_RPM_LIMIT=300
+# MINERU_TOKEN_DAILY_LIMIT=10000
+# MINERU_TOKEN_COOLDOWN_SECONDS=60
+# FORCE_MINERU_UPLOAD_ENABLED=false
+# ILOVEAPI_BASE_URL=https://api.ilovepdf.com/v1
+# ILOVEAPI_TIMEOUT=120
# File handling defaults
SUPPORTED_EXTENSIONS=.doc,.docx,.pdf,.txt,.xls,.xlsx,.csv,.pptx,.jpg,.jpeg,.png,.md
MAX_FILE_SIZE=104857600
# Legacy parser compatibility fields.
-ALL_DF_COLS=content,path,type,length,keywords,summary,know_id,tokens,connectto,addtime,page_nums
-SPLIT_CHAR=/
+# ALL_DF_COLS=content,path,type,length,keywords,summary,know_id,tokens,connectto,addtime,page_nums
+# SPLIT_CHAR=/
diff --git a/apps/worker/app/services/document_agent/__init__.py b/apps/worker/app/services/document_agent/__init__.py
new file mode 100644
index 000000000..2e36f6f2e
--- /dev/null
+++ b/apps/worker/app/services/document_agent/__init__.py
@@ -0,0 +1,15 @@
+"""Phase 1 document split-agent utilities."""
+
+from app.services.document_agent.manifest import (
+ GlobalSignals,
+ ShardManifest,
+ ShardSignal,
+ SpecialPage,
+)
+
+__all__ = [
+ "GlobalSignals",
+ "ShardManifest",
+ "ShardSignal",
+ "SpecialPage",
+]
diff --git a/apps/worker/app/services/document_agent/manifest.py b/apps/worker/app/services/document_agent/manifest.py
new file mode 100644
index 000000000..7ed3f9cba
--- /dev/null
+++ b/apps/worker/app/services/document_agent/manifest.py
@@ -0,0 +1,137 @@
+"""Stable Phase 1 shard manifest contract.
+
+The manifest is intentionally independent from existing parser internals so it
+can be produced and inspected before parser integration work begins.
+"""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
+from typing import Any, Literal
+
+
+SpecialKind = Literal[
+ "toc",
+ "blank",
+ "sparse",
+ "table_heavy",
+ "image_heavy",
+ "landscape",
+ "single_image",
+ "normal",
+]
+
+PredominantKind = Literal[
+ "text_dense",
+ "table_heavy",
+ "image_heavy",
+ "mixed",
+ "landscape_block",
+ "toc",
+ "sparse",
+]
+
+
+@dataclass
+class SpecialPage:
+ page: int
+ kind: SpecialKind
+ confidence: float
+ note: str = ""
+
+ def to_dict(self) -> dict[str, Any]:
+ return asdict(self)
+
+
+@dataclass
+class ShardSignal:
+ page_start: int
+ page_end: int
+ page_offset: int
+ predominant_kind: PredominantKind
+ special_pages: list[SpecialPage] = field(default_factory=list)
+ estimated_difficulty: str | None = None
+ parser_hint: str | None = None
+ cut_rationale: str = ""
+
+ def to_dict(self) -> dict[str, Any]:
+ data = asdict(self)
+ data["special_pages"] = [page.to_dict() for page in self.special_pages]
+ return data
+
+
+@dataclass
+class GlobalSignals:
+ has_toc: bool
+ toc_pages: list[int]
+ landscape_ratio: float
+ table_page_ratio: float
+ image_page_ratio: float
+ sample_size: int
+ notes: str = ""
+
+ def to_dict(self) -> dict[str, Any]:
+ return asdict(self)
+
+
+@dataclass
+class ShardManifest:
+ job_id: str
+ file_uri: str
+ file_sha: str
+ page_count: int
+ shard_count: int
+ shards: list[ShardSignal]
+ global_signals: GlobalSignals
+ decision_log_ref: str
+ created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+ version: str = "1.0"
+
+ def validate(self) -> None:
+ """Enforce the downstream coverage contract."""
+ if self.page_count < 0:
+ raise ValueError("page_count must be non-negative")
+ if self.shard_count != len(self.shards):
+ raise ValueError("shard_count must match shards length")
+ if self.page_count == 0:
+ if self.shards:
+ raise ValueError("empty documents cannot contain shards")
+ return
+
+ expected_start = 1
+ for shard in self.shards:
+ if shard.page_start != expected_start:
+ raise ValueError(
+ f"non-contiguous shard coverage at page {expected_start}: "
+ f"got start={shard.page_start}"
+ )
+ if shard.page_end < shard.page_start:
+ raise ValueError(
+ f"invalid shard range {shard.page_start}-{shard.page_end}"
+ )
+ if shard.page_offset != shard.page_start - 1:
+ raise ValueError(
+ f"invalid page_offset for shard {shard.page_start}-{shard.page_end}"
+ )
+ expected_start = shard.page_end + 1
+
+ if expected_start != self.page_count + 1:
+ raise ValueError(
+ f"shards must cover 1..{self.page_count}, stopped at {expected_start - 1}"
+ )
+
+ def to_dict(self) -> dict[str, Any]:
+ self.validate()
+ return {
+ "version": self.version,
+ "job_id": self.job_id,
+ "file_uri": self.file_uri,
+ "file_sha": self.file_sha,
+ "page_count": self.page_count,
+ "shard_count": self.shard_count,
+ "shards": [shard.to_dict() for shard in self.shards],
+ "global_signals": self.global_signals.to_dict(),
+ "decision_log_ref": self.decision_log_ref,
+ "created_at": self.created_at.isoformat(),
+ }
diff --git a/apps/worker/app/services/document_agent/tools/__init__.py b/apps/worker/app/services/document_agent/tools/__init__.py
new file mode 100644
index 000000000..8f54e85fa
--- /dev/null
+++ b/apps/worker/app/services/document_agent/tools/__init__.py
@@ -0,0 +1,15 @@
+"""Document split-agent Phase 1 tools."""
+
+from app.services.document_agent.tools.classify_special_pages import (
+ classify_special_pages,
+)
+from app.services.document_agent.tools.probe_sample_pages import sample_pages
+from app.services.document_agent.tools.probe_vlm_inspect import vlm_inspect_pages
+from app.services.document_agent.tools.propose_shard_plan import propose_shard_plan
+
+__all__ = [
+ "classify_special_pages",
+ "propose_shard_plan",
+ "sample_pages",
+ "vlm_inspect_pages",
+]
diff --git a/apps/worker/app/services/document_agent/tools/classify_special_pages.py b/apps/worker/app/services/document_agent/tools/classify_special_pages.py
new file mode 100644
index 000000000..564307b21
--- /dev/null
+++ b/apps/worker/app/services/document_agent/tools/classify_special_pages.py
@@ -0,0 +1,172 @@
+"""Classify sampled pages into Phase 1 special page kinds."""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from app.services.document_agent.manifest import SpecialKind
+from app.services.document_agent.tools.llm_json import extract_json_object
+from loguru import logger
+
+ALLOWED_KINDS: set[str] = {
+ "toc",
+ "blank",
+ "sparse",
+ "table_heavy",
+ "image_heavy",
+ "landscape",
+ "single_image",
+ "normal",
+}
+
+PROMPT = """You are classifying sampled PDF pages for a document split planner.
+Return ONLY valid json.
+
+Allowed special_kind values:
+- toc: table of contents / contents pages
+- blank: blank page
+- sparse: very little useful content
+- table_heavy: mostly tables or dense tabular rules
+- image_heavy: many charts/photos/figures
+- landscape: landscape page that should not be split through a landscape block
+- single_image: one large screenshot/scanned image dominates the page
+- normal: ordinary text page
+
+Use structural features first, and use VLM observations when present. Be conservative:
+only mark toc/table/image/landscape/single_image when there is clear evidence.
+
+JSON schema:
+{
+ "pages": [
+ {"page": 1, "special_kind": "normal", "confidence": 0.75, "note": "short reason"}
+ ],
+ "global_notes": "short summary"
+}
+"""
+
+
+def _heuristic_kind(page: dict[str, Any]) -> tuple[SpecialKind, float, str]:
+ text = str(page.get("text_preview") or "")
+ text_len = int(page.get("text_length") or 0)
+ image_coverage = float(page.get("image_coverage") or 0.0)
+ table_count = int(page.get("table_count") or 0)
+ drawings_count = int(page.get("drawings_count") or 0)
+ orientation = str(page.get("orientation") or "")
+ is_blank_like = bool(page.get("is_blank_like"))
+
+ toc_markers = ["目录", "contents", "table of contents"]
+ if any(marker.lower() in text.lower() for marker in toc_markers):
+ return "toc", 0.82, "text preview contains TOC marker"
+ if is_blank_like:
+ return "blank", 0.9, "very low text/image/drawing signal"
+ if image_coverage >= 0.72 and text_len < 250:
+ return "single_image", 0.82, "one or more images dominate the page"
+ if table_count > 0 or drawings_count >= 80:
+ return "table_heavy", 0.72, "table detector or dense ruled drawings fired"
+ if image_coverage >= 0.35:
+ return "image_heavy", 0.72, "high image coverage"
+ if orientation == "landscape":
+ return "landscape", 0.75, "page is landscape"
+ if text_len < 80:
+ return "sparse", 0.68, "short text and no stronger special signal"
+ return "normal", 0.65, "no special signal"
+
+
+def heuristic_classify_special_pages(
+ sampled_pages: list[dict[str, Any]],
+) -> dict[str, Any]:
+ pages = []
+ for page in sampled_pages:
+ kind, confidence, note = _heuristic_kind(page)
+ pages.append(
+ {
+ "page": int(page.get("page_number") or 0),
+ "special_kind": kind,
+ "confidence": confidence,
+ "note": note,
+ }
+ )
+ return {"pages": pages, "global_notes": "heuristic classification"}
+
+
+def _normalize_llm_pages(
+ data: dict[str, Any],
+ sampled_pages: list[dict[str, Any]],
+) -> dict[str, Any]:
+ by_page = {int(page.get("page_number") or 0): page for page in sampled_pages}
+ pages = []
+ for item in data.get("pages", []) or []:
+ if not isinstance(item, dict):
+ continue
+ page_number = int(item.get("page") or item.get("page_number") or 0)
+ if page_number not in by_page:
+ continue
+ kind = str(item.get("special_kind") or item.get("kind") or "normal")
+ if kind not in ALLOWED_KINDS:
+ kind = "normal"
+ confidence = max(0.0, min(float(item.get("confidence") or 0.0), 1.0))
+ if confidence <= 0:
+ confidence = 0.5
+ pages.append(
+ {
+ "page": page_number,
+ "special_kind": kind,
+ "confidence": confidence,
+ "note": str(item.get("note") or "")[:300],
+ }
+ )
+
+ seen = {item["page"] for item in pages}
+ for fallback in heuristic_classify_special_pages(sampled_pages)["pages"]:
+ if fallback["page"] not in seen:
+ pages.append(fallback)
+ pages.sort(key=lambda item: item["page"])
+ return {
+ "pages": pages,
+ "global_notes": str(data.get("global_notes") or data.get("notes") or "")[:1000],
+ }
+
+
+def classify_special_pages(
+ sampled_pages: list[dict[str, Any]],
+ *,
+ vlm_observations: list[dict[str, Any]] | None = None,
+ model: str | None = None,
+ use_llm: bool = True,
+) -> dict[str, Any]:
+ """Classify special pages with LLM, falling back to deterministic heuristics."""
+ if not use_llm:
+ return heuristic_classify_special_pages(sampled_pages)
+
+ try:
+ from shared.core.config import settings
+ from shared.utils.OpenAICompatibleClientSync import get_openai_client
+
+ effective_model = model or settings.HIERARCHY_LLM_MODEL or settings.NORMOL_MODEL
+ client = get_openai_client(model=effective_model)
+ payload = {
+ "sampled_pages": sampled_pages,
+ "vlm_observations": vlm_observations or [],
+ }
+ response = client.chat_completion(
+ messages=[
+ {"role": "system", "content": PROMPT},
+ {
+ "role": "user",
+ "content": "Classify these pages as json:\n"
+ + json.dumps(payload, ensure_ascii=False),
+ },
+ ],
+ model=effective_model,
+ temperature=0.0,
+ max_tokens=1800,
+ response_format={"type": "json_object"},
+ )
+ return _normalize_llm_pages(extract_json_object(response), sampled_pages)
+ except Exception as exc:
+ logger.warning(
+ f"[document_agent.classify_special_pages] LLM classification failed, "
+ f"using heuristics: {exc}"
+ )
+ return heuristic_classify_special_pages(sampled_pages)
diff --git a/apps/worker/app/services/document_agent/tools/llm_json.py b/apps/worker/app/services/document_agent/tools/llm_json.py
new file mode 100644
index 000000000..9531e3690
--- /dev/null
+++ b/apps/worker/app/services/document_agent/tools/llm_json.py
@@ -0,0 +1,28 @@
+"""Small JSON helpers for split-agent LLM tools."""
+
+from __future__ import annotations
+
+import json
+import re
+from typing import Any
+
+
+def extract_json_object(text: str) -> dict[str, Any]:
+ """Parse JSON from a model response that may include light prose/fences."""
+ raw = (text or "").strip()
+ if not raw:
+ raise ValueError("empty JSON response")
+ if raw.startswith("```"):
+ raw = re.sub(r"^```(?:json)?\s*", "", raw, flags=re.IGNORECASE)
+ raw = re.sub(r"\s*```$", "", raw)
+ try:
+ data = json.loads(raw)
+ except json.JSONDecodeError:
+ start = raw.find("{")
+ end = raw.rfind("}")
+ if start < 0 or end <= start:
+ raise
+ data = json.loads(raw[start : end + 1])
+ if not isinstance(data, dict):
+ raise ValueError("expected JSON object")
+ return data
diff --git a/apps/worker/app/services/document_agent/tools/probe_sample_pages.py b/apps/worker/app/services/document_agent/tools/probe_sample_pages.py
new file mode 100644
index 000000000..e0f21ac88
--- /dev/null
+++ b/apps/worker/app/services/document_agent/tools/probe_sample_pages.py
@@ -0,0 +1,208 @@
+"""Feature-page sampling for the Phase 1 split agent."""
+
+from __future__ import annotations
+
+import gc
+import statistics
+from typing import Any, Literal
+
+from app.services.document_parser.pymupdf_subprocess import run_in_child_process, worker
+
+SampleStrategy = Literal["stratified", "uniform", "key_pages"]
+
+
+def _choose_sample_indices(
+ page_count: int,
+ *,
+ strategy: SampleStrategy = "stratified",
+ max_samples: int = 25,
+) -> list[int]:
+ if page_count <= 0:
+ return []
+ if max_samples <= 0:
+ return []
+ if page_count <= max_samples:
+ return list(range(page_count))
+
+ if strategy == "key_pages":
+ candidates = [0, 1, 2, 3, 4, page_count - 5, page_count - 4, page_count - 3, page_count - 2, page_count - 1]
+ return sorted({idx for idx in candidates if 0 <= idx < page_count})[:max_samples]
+
+ if strategy == "uniform":
+ if max_samples == 1:
+ return [0]
+ return sorted(
+ {
+ round(i * (page_count - 1) / (max_samples - 1))
+ for i in range(max_samples)
+ }
+ )
+
+ edge_each = min(5, max_samples // 3)
+ edge_indices = list(range(edge_each)) + list(range(page_count - edge_each, page_count))
+ remaining = max_samples - len(set(edge_indices))
+ middle_start = edge_each
+ middle_end = page_count - edge_each - 1
+ middle_indices: list[int] = []
+ if remaining > 0 and middle_start <= middle_end:
+ if remaining == 1:
+ middle_indices = [(middle_start + middle_end) // 2]
+ else:
+ middle_indices = [
+ round(middle_start + i * (middle_end - middle_start) / (remaining - 1))
+ for i in range(remaining)
+ ]
+ return sorted({idx for idx in edge_indices + middle_indices if 0 <= idx < page_count})
+
+
+def _rect_area(rect: Any) -> float:
+ return max(float(getattr(rect, "width", 0.0) or 0.0), 0.0) * max(
+ float(getattr(rect, "height", 0.0) or 0.0),
+ 0.0,
+ )
+
+
+def _measure_image_coverage(page: Any, page_area: float) -> tuple[float, int]:
+ if page_area <= 0:
+ return 0.0, 0
+ image_area = 0.0
+ images = page.get_images(full=True) or []
+ seen_rects: set[tuple[float, float, float, float]] = set()
+ for image in images:
+ if not image:
+ continue
+ xref = image[0]
+ try:
+ rects = page.get_image_rects(xref) or []
+ except Exception:
+ rects = []
+ for rect in rects:
+ key = (
+ round(float(getattr(rect, "x0", 0.0) or 0.0), 2),
+ round(float(getattr(rect, "y0", 0.0) or 0.0), 2),
+ round(float(getattr(rect, "x1", 0.0) or 0.0), 2),
+ round(float(getattr(rect, "y1", 0.0) or 0.0), 2),
+ )
+ if key in seen_rects:
+ continue
+ seen_rects.add(key)
+ image_area += _rect_area(rect)
+ return min(image_area / page_area, 1.0), len(images)
+
+
+def _font_stats(page: Any) -> dict[str, float | int]:
+ sizes: list[float] = []
+ try:
+ text_dict = page.get_text("dict") or {}
+ except Exception:
+ text_dict = {}
+ for block in text_dict.get("blocks", []) or []:
+ for line in block.get("lines", []) or []:
+ for span in line.get("spans", []) or []:
+ size = float(span.get("size") or 0.0)
+ if size > 0:
+ sizes.append(size)
+ if not sizes:
+ return {"min": 0.0, "max": 0.0, "mean": 0.0, "median": 0.0, "count": 0}
+ return {
+ "min": min(sizes),
+ "max": max(sizes),
+ "mean": statistics.fmean(sizes),
+ "median": statistics.median(sizes),
+ "count": len(sizes),
+ }
+
+
+def _table_count(page: Any) -> int:
+ try:
+ finder = page.find_tables()
+ return len(getattr(finder, "tables", []) or [])
+ except Exception:
+ return 0
+
+
+def _extract_page_features(page: Any, page_index: int) -> dict[str, Any]:
+ rect = page.rect
+ page_area = max(_rect_area(rect), 1.0)
+ text = page.get_text() or ""
+ text_len = len(text.strip())
+ image_coverage, image_count = _measure_image_coverage(page, page_area)
+ try:
+ drawings_count = len(page.get_drawings() or [])
+ except Exception:
+ drawings_count = 0
+ orientation = "landscape" if float(rect.width) > float(rect.height) else "portrait"
+ text_density = text_len / page_area * 10000
+ table_count = _table_count(page)
+ is_blank_like = text_len < 20 and image_coverage < 0.02 and drawings_count < 5
+
+ return {
+ "page_index": page_index,
+ "page_number": page_index + 1,
+ "width": float(rect.width),
+ "height": float(rect.height),
+ "orientation": orientation,
+ "text_length": text_len,
+ "text_density": round(text_density, 4),
+ "image_count": image_count,
+ "image_coverage": round(image_coverage, 4),
+ "table_count": table_count,
+ "drawings_count": drawings_count,
+ "font_size_stats": _font_stats(page),
+ "is_blank_like": is_blank_like,
+ "text_preview": " ".join(text.split())[:500],
+ }
+
+
+@worker
+def _sample_pages_worker(
+ queue,
+ pdf_path: str,
+ strategy: str,
+ max_samples: int,
+) -> None:
+ import pymupdf
+
+ doc = pymupdf.open(pdf_path)
+ try:
+ page_count = int(doc.page_count)
+ indices = _choose_sample_indices(
+ page_count,
+ strategy=strategy if strategy in {"stratified", "uniform", "key_pages"} else "stratified",
+ max_samples=max_samples,
+ )
+ sampled_pages = [_extract_page_features(doc[idx], idx) for idx in indices]
+ finally:
+ doc.close()
+ gc.collect()
+
+ queue.put(
+ {
+ "ok": True,
+ "page_count": page_count,
+ "sample_indices": indices,
+ "sampled_pages": sampled_pages,
+ }
+ )
+
+
+def sample_pages(
+ pdf_path: str,
+ *,
+ strategy: SampleStrategy = "stratified",
+ max_samples: int = 25,
+ timeout: int = 120,
+) -> dict[str, Any]:
+ """Sample structural page features in an isolated PyMuPDF child process."""
+ result = run_in_child_process(
+ _sample_pages_worker,
+ pdf_path,
+ strategy,
+ max_samples,
+ timeout=timeout,
+ )
+ return {
+ "page_count": int(result.get("page_count") or 0),
+ "sample_indices": list(result.get("sample_indices") or []),
+ "sampled_pages": list(result.get("sampled_pages") or []),
+ }
diff --git a/apps/worker/app/services/document_agent/tools/probe_vlm_inspect.py b/apps/worker/app/services/document_agent/tools/probe_vlm_inspect.py
new file mode 100644
index 000000000..7b4de6396
--- /dev/null
+++ b/apps/worker/app/services/document_agent/tools/probe_vlm_inspect.py
@@ -0,0 +1,161 @@
+"""Selective VLM inspection for ambiguous sampled PDF pages."""
+
+from __future__ import annotations
+
+import base64
+import os
+import tempfile
+from typing import Any
+
+from app.services.document_parser.pymupdf_subprocess import run_in_child_process, worker
+from loguru import logger
+from openai.types.chat import (
+ ChatCompletionContentPartImageParam,
+ ChatCompletionContentPartParam,
+ ChatCompletionContentPartTextParam,
+ ChatCompletionMessageParam,
+)
+
+DEFAULT_QUESTION = (
+ "Inspect these PDF page screenshots. For each page, decide whether it is a "
+ "table-heavy page, image-heavy page, table of contents, blank/sparse page, "
+ "landscape page, single-image page, or normal page. Return compact JSON with "
+ "items: [{page, judgement, confidence, note}]."
+)
+
+
+@worker
+def _render_vlm_pages_worker(
+ queue,
+ pdf_path: str,
+ page_indices: list[int],
+ dpi: int,
+ out_dir: str,
+) -> None:
+ import pymupdf
+
+ doc = pymupdf.open(pdf_path)
+ rendered: list[dict[str, Any]] = []
+ try:
+ mat = pymupdf.Matrix(dpi / 72, dpi / 72)
+ for idx in page_indices:
+ if idx < 0 or idx >= doc.page_count:
+ continue
+ page = doc[idx]
+ pix = page.get_pixmap(matrix=mat, alpha=False)
+ out_path = os.path.join(out_dir, f"vlm_probe_p{idx + 1}.png")
+ pix.save(out_path)
+ rendered.append({"page_index": idx, "page_number": idx + 1, "path": out_path})
+ pix = None
+ page = None
+ finally:
+ doc.close()
+ queue.put({"ok": True, "rendered": rendered})
+
+
+def _png_to_data_url(path: str) -> str | None:
+ try:
+ with open(path, "rb") as file:
+ data = base64.b64encode(file.read()).decode("utf-8")
+ return f"data:image/png;base64,{data}"
+ except Exception as exc:
+ logger.warning(f"[document_agent.vlm_inspect] failed to encode {path}: {exc}")
+ return None
+
+
+def _call_vlm(
+ *,
+ image_items: list[dict[str, Any]],
+ question: str,
+ model: str | None = None,
+ max_tokens: int = 900,
+) -> tuple[str, dict[str, int]]:
+ from shared.core.config import settings
+ from shared.utils.OpenAICompatibleClientSync import get_openai_client
+
+ effective_model = model or settings.IMAGE_MODEL or "qwen3.5-flash"
+ client = get_openai_client(model=effective_model)
+ content: list[ChatCompletionContentPartParam] = [
+ ChatCompletionContentPartTextParam(
+ type="text",
+ text=question,
+ )
+ ]
+
+ for item in image_items:
+ url = item.get("data_url")
+ if not url:
+ continue
+ content.append(
+ ChatCompletionContentPartTextParam(
+ type="text",
+ text=f"Page {item['page_number']}:",
+ )
+ )
+ content.append(
+ ChatCompletionContentPartImageParam(
+ type="image_url",
+ image_url={"url": url},
+ )
+ )
+
+ messages: list[ChatCompletionMessageParam] = [{"role": "user", "content": content}]
+ return client.chat_completion_with_usage(
+ messages=messages,
+ model=effective_model,
+ temperature=0.0,
+ max_tokens=max_tokens,
+ )
+
+
+def vlm_inspect_pages(
+ pdf_path: str,
+ *,
+ page_indices: list[int],
+ question: str = DEFAULT_QUESTION,
+ dpi: int = 120,
+ model: str | None = None,
+ max_tokens: int = 900,
+ timeout: int = 60,
+) -> dict[str, Any]:
+ """Render selected 0-based pages and ask the configured VLM to inspect them."""
+ if not page_indices:
+ return {"observations": [], "raw_response": "", "usage": {}}
+
+ with tempfile.TemporaryDirectory(prefix="doc_agent_vlm_") as tmp_dir:
+ result = run_in_child_process(
+ _render_vlm_pages_worker,
+ pdf_path,
+ sorted(set(page_indices)),
+ dpi,
+ tmp_dir,
+ timeout=timeout,
+ )
+ image_items = []
+ for item in result.get("rendered", []) or []:
+ data_url = _png_to_data_url(item["path"])
+ if data_url is not None:
+ image_items.append({**item, "data_url": data_url})
+
+ if not image_items:
+ return {"observations": [], "raw_response": "", "usage": {}}
+
+ response, usage = _call_vlm(
+ image_items=image_items,
+ question=question,
+ model=model,
+ max_tokens=max_tokens,
+ )
+ return {
+ "observations": [
+ {
+ "page": item["page_number"],
+ "page_index": item["page_index"],
+ "vlm_judgement": response,
+ "confidence": None,
+ }
+ for item in image_items
+ ],
+ "raw_response": response,
+ "usage": usage,
+ }
diff --git a/apps/worker/app/services/document_agent/tools/propose_shard_plan.py b/apps/worker/app/services/document_agent/tools/propose_shard_plan.py
new file mode 100644
index 000000000..2eb834ff4
--- /dev/null
+++ b/apps/worker/app/services/document_agent/tools/propose_shard_plan.py
@@ -0,0 +1,361 @@
+"""Shard planning for the Phase 1 split agent."""
+
+from __future__ import annotations
+
+import json
+from collections import Counter
+from hashlib import sha256
+from typing import Any
+
+from app.services.document_agent.manifest import (
+ GlobalSignals,
+ ShardManifest,
+ ShardSignal,
+ SpecialPage,
+)
+from app.services.document_agent.tools.llm_json import extract_json_object
+from loguru import logger
+
+PROMPT = """You are planning PDF shards for a downstream parser.
+Return ONLY valid json.
+
+Goal:
+- Cover every page from 1 to page_count exactly once.
+- Prefer shards as close to max_pages_per_shard pages as possible without exceeding it.
+- Do not cut through obvious table-heavy ranges, continuous image/landscape blocks, or likely TOC pages.
+- Align cuts near safer normal/sparse pages when possible.
+
+JSON schema:
+{
+ "cuts": [
+ {"start": 1, "end": 199, "predominant_kind": "text_dense", "rationale": "short reason"}
+ ],
+ "global_notes": "short summary"
+}
+
+Allowed predominant_kind values: text_dense, table_heavy, image_heavy, mixed, landscape_block, toc, sparse.
+"""
+
+ALLOWED_PREDOMINANT = {
+ "text_dense",
+ "table_heavy",
+ "image_heavy",
+ "mixed",
+ "landscape_block",
+ "toc",
+ "sparse",
+}
+
+
+def _special_by_page(classifications: dict[str, Any]) -> dict[int, dict[str, Any]]:
+ by_page = {}
+ for item in classifications.get("pages", []) or []:
+ if not isinstance(item, dict):
+ continue
+ page = int(item.get("page") or 0)
+ if page > 0:
+ by_page[page] = item
+ return by_page
+
+
+def _kind_for_range(start: int, end: int, by_page: dict[int, dict[str, Any]]) -> str:
+ kinds = [
+ str(item.get("special_kind") or item.get("kind") or "normal")
+ for page, item in by_page.items()
+ if start <= page <= end
+ ]
+ if not kinds:
+ return "text_dense"
+ counts = Counter(kinds)
+ special_total = sum(count for kind, count in counts.items() if kind != "normal")
+ if special_total == 0:
+ return "text_dense"
+ top_kind, top_count = counts.most_common(1)[0]
+ if top_kind in {"table_heavy"}:
+ return "table_heavy"
+ if top_kind in {"image_heavy", "single_image"}:
+ return "image_heavy"
+ if top_kind == "landscape":
+ return "landscape_block"
+ if top_kind == "toc":
+ return "toc"
+ if top_kind in {"blank", "sparse"} and top_count >= max(1, (end - start + 1) // 2):
+ return "sparse"
+ return "mixed"
+
+
+def _fallback_cuts(
+ page_count: int,
+ *,
+ max_pages_per_shard: int,
+ classifications: dict[str, Any],
+) -> list[dict[str, Any]]:
+ by_page = _special_by_page(classifications)
+ cuts = []
+ start = 1
+ while start <= page_count:
+ target_end = min(start + max_pages_per_shard - 1, page_count)
+ end = target_end
+ if target_end < page_count:
+ # Prefer a safe boundary close to the shard limit without exceeding it.
+ target_kind = str(by_page.get(target_end, {}).get("special_kind") or "normal")
+ if target_kind not in {"blank", "sparse", "normal"}:
+ window_start = max(start, target_end - 5)
+ candidates = []
+ for page in range(window_start, target_end):
+ kind = str(by_page.get(page, {}).get("special_kind") or "normal")
+ priority = {"blank": 0, "sparse": 1, "normal": 2}.get(kind)
+ if priority is not None:
+ candidates.append((abs(page - target_end), priority, page))
+ if candidates:
+ end = min(candidates)[2]
+ cuts.append(
+ {
+ "start": start,
+ "end": end,
+ "predominant_kind": _kind_for_range(start, end, by_page),
+ "rationale": "deterministic fallback cut near max shard size",
+ }
+ )
+ start = end + 1
+ return cuts
+
+
+def _normalize_cuts(
+ cuts: list[dict[str, Any]],
+ *,
+ page_count: int,
+ max_pages_per_shard: int,
+ classifications: dict[str, Any],
+) -> list[dict[str, Any]]:
+ normalized = []
+ expected = 1
+ for raw in cuts:
+ if not isinstance(raw, dict):
+ continue
+ start = int(raw.get("start") or raw.get("page_start") or 0)
+ end = int(raw.get("end") or raw.get("page_end") or 0)
+ if start != expected or end < start or end > page_count:
+ raise ValueError("LLM shard cuts are not contiguous")
+ kind = str(raw.get("predominant_kind") or "mixed")
+ if kind not in ALLOWED_PREDOMINANT:
+ kind = "mixed"
+ normalized.append(
+ {
+ "start": start,
+ "end": end,
+ "predominant_kind": kind,
+ "rationale": str(raw.get("rationale") or "")[:500],
+ }
+ )
+ expected = end + 1
+ if expected != page_count + 1:
+ raise ValueError("LLM shard cuts do not cover all pages")
+ if not normalized:
+ raise ValueError("empty LLM shard cuts")
+ # Avoid accepting pathological single giant cuts except naturally small docs.
+ if page_count > max_pages_per_shard * 2 and any(
+ cut["end"] - cut["start"] + 1 > max_pages_per_shard * 2
+ for cut in normalized
+ ):
+ raise ValueError("LLM shard cut exceeds hard tolerance")
+ return normalized
+
+
+def _build_global_signals(
+ *,
+ sampled_pages: list[dict[str, Any]],
+ classifications: dict[str, Any],
+) -> GlobalSignals:
+ pages = classifications.get("pages", []) or []
+ toc_pages = [
+ int(page.get("page") or 0)
+ for page in pages
+ if str(page.get("special_kind") or page.get("kind")) == "toc"
+ ]
+ sample_size = len(sampled_pages)
+ if sample_size <= 0:
+ return GlobalSignals(
+ has_toc=bool(toc_pages),
+ toc_pages=toc_pages,
+ landscape_ratio=0.0,
+ table_page_ratio=0.0,
+ image_page_ratio=0.0,
+ sample_size=0,
+ notes=str(classifications.get("global_notes") or ""),
+ )
+ landscape_count = sum(1 for page in sampled_pages if page.get("orientation") == "landscape")
+ table_count = sum(
+ 1
+ for page in pages
+ if str(page.get("special_kind") or page.get("kind")) == "table_heavy"
+ )
+ image_count = sum(
+ 1
+ for page in pages
+ if str(page.get("special_kind") or page.get("kind")) in {"image_heavy", "single_image"}
+ )
+ return GlobalSignals(
+ has_toc=bool(toc_pages),
+ toc_pages=toc_pages,
+ landscape_ratio=landscape_count / sample_size,
+ table_page_ratio=table_count / sample_size,
+ image_page_ratio=image_count / sample_size,
+ sample_size=sample_size,
+ notes=str(classifications.get("global_notes") or ""),
+ )
+
+
+def build_manifest_from_cuts(
+ *,
+ file_uri: str,
+ job_id: str,
+ page_count: int,
+ sampled_pages: list[dict[str, Any]],
+ classifications: dict[str, Any],
+ cuts: list[dict[str, Any]],
+ decision_log_ref: str = "local-debug",
+) -> ShardManifest:
+ by_page = _special_by_page(classifications)
+ shards = []
+ for cut in cuts:
+ start = int(cut["start"])
+ end = int(cut["end"])
+ special_pages = []
+ for page in range(start, end + 1):
+ item = by_page.get(page)
+ if not item:
+ continue
+ kind = str(item.get("special_kind") or item.get("kind") or "normal")
+ if kind == "normal":
+ continue
+ special_pages.append(
+ SpecialPage(
+ page=page,
+ kind=kind, # type: ignore[arg-type]
+ confidence=float(item.get("confidence") or 0.0),
+ note=str(item.get("note") or ""),
+ )
+ )
+ shards.append(
+ ShardSignal(
+ page_start=start,
+ page_end=end,
+ page_offset=start - 1,
+ predominant_kind=cut["predominant_kind"],
+ special_pages=special_pages,
+ cut_rationale=str(cut.get("rationale") or ""),
+ )
+ )
+
+ manifest = ShardManifest(
+ job_id=job_id,
+ file_uri=file_uri,
+ file_sha=_hash_file(file_uri),
+ page_count=page_count,
+ shard_count=len(shards),
+ shards=shards,
+ global_signals=_build_global_signals(
+ sampled_pages=sampled_pages,
+ classifications=classifications,
+ ),
+ decision_log_ref=decision_log_ref,
+ )
+ manifest.validate()
+ return manifest
+
+
+def _hash_file(file_uri: str) -> str:
+ digest = sha256()
+ try:
+ with open(file_uri, "rb") as file:
+ for chunk in iter(lambda: file.read(1024 * 1024), b""):
+ digest.update(chunk)
+ return digest.hexdigest()
+ except OSError:
+ return sha256(file_uri.encode("utf-8")).hexdigest()
+
+
+def propose_shard_plan(
+ *,
+ file_uri: str,
+ job_id: str,
+ page_count: int,
+ sampled_pages: list[dict[str, Any]],
+ classifications: dict[str, Any],
+ max_pages_per_shard: int = 199,
+ model: str | None = None,
+ use_llm: bool = True,
+) -> dict[str, Any]:
+ """Produce a validated shard proposal and manifest."""
+ max_pages_per_shard = max(1, int(max_pages_per_shard))
+
+ cuts: list[dict[str, Any]]
+ raw_response = ""
+ if use_llm and page_count > max_pages_per_shard:
+ try:
+ from shared.core.config import settings
+ from shared.utils.OpenAICompatibleClientSync import get_openai_client
+
+ effective_model = model or settings.HIERARCHY_LLM_MODEL or settings.NORMOL_MODEL
+ client = get_openai_client(model=effective_model)
+ payload = {
+ "page_count": page_count,
+ "max_pages_per_shard": max_pages_per_shard,
+ "sampled_pages": sampled_pages,
+ "page_classifications": classifications,
+ }
+ raw_response = client.chat_completion(
+ messages=[
+ {"role": "system", "content": PROMPT},
+ {
+ "role": "user",
+ "content": "Propose a shard plan as json:\n"
+ + json.dumps(payload, ensure_ascii=False),
+ },
+ ],
+ model=effective_model,
+ temperature=0.0,
+ max_tokens=2200,
+ response_format={"type": "json_object"},
+ )
+ data = extract_json_object(raw_response)
+ cuts = _normalize_cuts(
+ list(data.get("cuts") or []),
+ page_count=page_count,
+ max_pages_per_shard=max_pages_per_shard,
+ classifications=classifications,
+ )
+ if data.get("global_notes") and not classifications.get("global_notes"):
+ classifications = {**classifications, "global_notes": data.get("global_notes")}
+ except Exception as exc:
+ logger.warning(
+ f"[document_agent.propose_shard_plan] LLM planning failed, "
+ f"using fallback: {exc}"
+ )
+ cuts = _fallback_cuts(
+ page_count,
+ max_pages_per_shard=max_pages_per_shard,
+ classifications=classifications,
+ )
+ else:
+ cuts = _fallback_cuts(
+ page_count,
+ max_pages_per_shard=max_pages_per_shard,
+ classifications=classifications,
+ )
+
+ manifest = build_manifest_from_cuts(
+ file_uri=file_uri,
+ job_id=job_id,
+ page_count=page_count,
+ sampled_pages=sampled_pages,
+ classifications=classifications,
+ cuts=cuts,
+ )
+ return {
+ "cuts": cuts,
+ "manifest": manifest,
+ "manifest_dict": manifest.to_dict(),
+ "raw_response": raw_response,
+ }
diff --git a/docs/assets/step-1-parse-build-memory.png b/docs/assets/step-1-parse-build-memory.png
new file mode 100644
index 000000000..66e6d63a5
Binary files /dev/null and b/docs/assets/step-1-parse-build-memory.png differ
diff --git a/docs/assets/step-2-agentic-retrieval.png b/docs/assets/step-2-agentic-retrieval.png
new file mode 100644
index 000000000..c15d5be15
Binary files /dev/null and b/docs/assets/step-2-agentic-retrieval.png differ
diff --git a/docs/external-services.md b/docs/external-services.md
new file mode 100644
index 000000000..72b15f107
--- /dev/null
+++ b/docs/external-services.md
@@ -0,0 +1,59 @@
+# External Services
+
+Knowhere can run with a small set of required infrastructure services. Most
+parser and retrieval tuning knobs have code defaults; configure only the
+external services and provider keys your deployment actually uses.
+
+## Required For API And Worker Startup
+
+- `DATABASE_URL`: PostgreSQL connection URL.
+- Redis: configure `REDIS_HOST` / `REDIS_PORT` or use the local defaults, plus
+ `CELERY_REDIS_URL` for worker task delivery.
+- S3-compatible storage: `S3_BUCKET_NAME`, `S3_ACCESS_KEY_ID`,
+ `S3_SECRET_ACCESS_KEY`, and `S3_TEMP_PATH`. For local development, the
+ example files point these at LocalStack.
+- `TMP_PATH`: local temporary directory. The directory must exist.
+
+## Required For Parsing And Retrieval
+
+Configure at least one OpenAI-compatible text LLM provider key:
+
+- `DS_KEY` for DeepSeek. `DS_URL` defaults to `https://api.deepseek.com/v1`.
+- `ALI_API_KEYS` for DashScope/Qwen models. `ALI_URL` has a code default.
+- `GPT_API_KEY` for OpenAI-compatible GPT models.
+- `GLM_API_KEY` for Zhipu GLM models. `GLM_URL` has a code default.
+
+`NORMOL_MODEL` defaults to `deepseek-chat`. `HIERARCHY_LLM_MODEL` defaults to
+empty and falls back to `NORMOL_MODEL`, so deployments do not need to set it
+unless they intentionally want a separate hierarchy-recognition model.
+
+## Feature-Specific Providers
+
+- PDF parsing currently routes non-atlas PDFs through MinerU, so PDF ingestion
+ requires `MINERU_API_KEYS`.
+- Image summaries, OCR, atlas classification, and image-aware retrieval require
+ a vision-capable provider for `IMAGE_MODEL` / `IMAGE_MODEL_MAX`. The model
+ names default to Qwen vision models, so configure `ALI_API_KEYS` or override
+ those model names to match your provider.
+- PPTX-to-PDF conversion can use iLovePDF. Configure `ILOVEAPI_PUBLIC_KEY` and
+ `ILOVEAPI_SECRET_KEY` only if you need that provider path.
+- Billing and outbound webhooks are optional. Configure Stripe, Moesif, or
+ QStash only when those integrations are enabled for your deployment.
+
+## Values With Code Defaults
+
+The following groups are intentionally optional in `.env` files:
+
+- Retrieval workflow knobs: `RETRIEVAL_AGENTIC_ENABLED`,
+ `RETRIEVAL_PLANNER_THINKING_BUDGET`, `RETRIEVAL_DECOMPOSITION_MAX_STEPS`,
+ `RETRIEVAL_WALLET_*`, and `RETRIEVAL_WORKFLOW_PARALLEL_MAX`.
+- Agentic retrieval internals: `RETRIEVAL_AGENTIC_MAX_REVISIONS`,
+ `RETRIEVAL_AGENTIC_MAX_NAV_DEPTH`, `RETRIEVAL_AGENTIC_TOKEN_BUDGET_TOTAL`,
+ `RETRIEVAL_AGENTIC_PLANNING_RATIO`,
+ `RETRIEVAL_AGENTIC_BOOTSTRAP_BUDGET`, and related trace or verbose flags.
+- Parser defaults: `MINERU_URL`, MinerU rate-limit settings,
+ `ILOVEAPI_BASE_URL`, `ILOVEAPI_TIMEOUT`, `SPLIT_CHAR`, and `ALL_DF_COLS`.
+- File handling defaults: `SUPPORTED_EXTENSIONS` and `MAX_FILE_SIZE`.
+
+Leave these unset for a normal fork or self-hosted deployment. Set them only
+when you need to deliberately change behavior.
diff --git a/packages/shared-python/shared/core/config/ai.py b/packages/shared-python/shared/core/config/ai.py
index 2164df3d2..b48567a2d 100644
--- a/packages/shared-python/shared/core/config/ai.py
+++ b/packages/shared-python/shared/core/config/ai.py
@@ -11,8 +11,10 @@ class AIConfig(BaseModel):
GLM_URL: str = Field(
default="https://open.bigmodel.cn/api/paas/v4", description="Zhipu GLM API URL"
)
- DS_KEY: str = Field(..., description="DeepSeek API key")
- DS_URL: str = Field(..., description="DeepSeek API URL")
+ DS_KEY: str = Field(default="", description="DeepSeek API key")
+ DS_URL: str = Field(
+ default="https://api.deepseek.com/v1", description="DeepSeek API URL"
+ )
GPT_API_KEY: str = Field(default="", description="OpenAI API key")
# Default behavior: text/table summaries use deepseek-chat. Hierarchy parsing
# can be overridden independently with HIERARCHY_LLM_MODEL. Existing