Add page-memory VLM capacity gate

apps/worker/app/services/page_memory/memory_service.py

@@ -445,6 +445,7 @@ def _build_page_dataframe(
             vlm_model=vlm_model,
             page_assets_by_page=page_assets_by_page,
             node_summary_max_pages=page_memory_config.node_summary_max_pages,
+            node_assembly_concurrency=page_memory_config.node_assembly_concurrency,
         )
     logger.info(
         "[page_memory] C7 assembled {} node rows (verdict={})",
@@ -625,14 +626,15 @@ def _run_hierarchy_scope(
     fat_leaf_pages = compute_fat_leaf_pages(scope_skeletons, min_pages=fine_min)
     if fat_leaf_pages:
         title_pages = sorted(fat_leaf_pages)
-        title_rendered = render_document_pages(
-            pdf_path=pdf_path,
-            page_count=page_count,
-            output_dir=output_dir,
-            pages=title_pages,
-            page_features=page_features,
-            page_texts=page_texts,
-        )
+        with stage_timer("page_memory.title_render", page_count=len(title_pages)):
+            title_rendered = render_document_pages(
+                pdf_path=pdf_path,
+                page_count=page_count,
+                output_dir=output_dir,
+                pages=title_pages,
+                page_features=page_features,
+                page_texts=page_texts,
+            )
         title_tags = [
             PageTagResult(
                 page_index=page,
@@ -649,6 +651,7 @@ def _run_hierarchy_scope(
                 fat_leaf_pages=fat_leaf_pages,
                 budget=None,
                 vlm_model=vlm_model,
+                max_concurrent=page_memory_config.title_detection_concurrency,
             )
         _record_trace_stage(
             trace_recorder,

apps/worker/app/services/page_memory/node_assembler.py

@@ -25,10 +25,11 @@
 import shutil
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 
 from loguru import logger
 
+from app.services.document_parser.support.stage_profiler import stage_timer
 from app.services.document_parser.support.identifiers import gen_str_codes, get_str_time
 from app.services.document_parser.support.parser_rows import serialize_entities
 from app.services.page_memory.page_assets import (
@@ -416,63 +417,114 @@ def build_node_rows(
     vlm_model: str | None = None,
     page_assets_by_page: dict[int, list[PageAsset]] | None = None,
     node_summary_max_pages: int = _NODE_SUMMARY_MAX_PAGES_DEFAULT,
+    node_assembly_concurrency: int = 3,
 ) -> list[dict[str, Any]]:
     """Assemble one row per leaf section node (node-granularity chunks)."""
     available_pages = set(raw_text_by_page.keys())
     leaves = identify_leaf_nodes(skeletons)
     views, page_owner = assign_pages_to_leaves(leaves, available_pages=available_pages)
     page_to_leaves = pages_by_leaf_count(views)
+    resolved_concurrency = max(1, node_assembly_concurrency)
 
     # Resolve body text once per owned page (PyMuPDF, OCR fallback for scanned).
     resolved_text: dict[int, str] = {}
-    for view in views:
-        for page in view.owned_pages:
-            resolved_text[page] = resolve_page_text(
+    owned_pages = sorted({page for view in views for page in view.owned_pages})
+    if owned_pages:
+        import gevent
+        from gevent.pool import Pool as GeventPool
+
+        def _resolve_one(page: int) -> tuple[int, str]:
+            return page, resolve_page_text(
                 page=page,
                 raw_text=raw_text_by_page.get(page, ""),
                 image_path=image_path_by_page.get(page),
                 vlm_model=vlm_model,
             )
 
+        with stage_timer(
+            "page_memory.node_ocr",
+            page_count=len(owned_pages),
+            concurrency=resolved_concurrency,
+        ):
+            pool = GeventPool(size=min(resolved_concurrency, len(owned_pages)))
+            greenlets = [pool.spawn(_resolve_one, page) for page in owned_pages]
+            gevent.joinall(greenlets, raise_error=True)
+            resolved_pairs = [
+                cast(tuple[int, str], greenlet.value)
+                for greenlet in greenlets
+            ]
+            resolved_text = {page: text for page, text in resolved_pairs}
+
+    summaries: dict[int, tuple[str, list[str], list[dict[str, str]]]] = {}
+    if views:
+        import gevent
+        from gevent.pool import Pool as GeventPool
+
+        def _summarize_one(
+            index: int,
+        ) -> tuple[int, tuple[str, list[str], list[dict[str, str]]]]:
+            return index, compute_node_summary(
+                view=views[index],
+                page_to_leaves=page_to_leaves,
+                tag_by_page=tag_by_page,
+                image_path_by_page=image_path_by_page,
+                vlm_model=vlm_model,
+                node_summary_max_pages=node_summary_max_pages,
+            )
+
+        with stage_timer(
+            "page_memory.node_summary",
+            node_count=len(views),
+            concurrency=resolved_concurrency,
+        ):
+            pool = GeventPool(size=min(resolved_concurrency, len(views)))
+            greenlets = [
+                pool.spawn(_summarize_one, index)
+                for index in range(len(views))
+            ]
+            gevent.joinall(greenlets, raise_error=True)
+            summary_pairs = [
+                cast(
+                    tuple[int, tuple[str, list[str], list[dict[str, str]]]],
+                    greenlet.value,
+                )
+                for greenlet in greenlets
+            ]
+            summaries = {index: result for index, result in summary_pairs}
+
     rows: list[dict[str, Any]] = []
     rows_by_path: dict[str, dict[str, Any]] = {}
-    for view in views:
-        leaf = view.leaf
-        content = build_node_content(
-            view,
-            page_owner=page_owner,
-            page_text=resolved_text,
-        )
-        summary, keywords, entities = compute_node_summary(
-            view=view,
-            page_to_leaves=page_to_leaves,
-            tag_by_page=tag_by_page,
-            image_path_by_page=image_path_by_page,
-            vlm_model=vlm_model,
-            node_summary_max_pages=node_summary_max_pages,
-        )
-        know_id = f"node_{gen_str_codes(f'{filename}::{leaf.section_path}')}"
-        row = {
-            "content": content,
-            "path": leaf.section_path,
-            "type": "page",
-            "length": len(content),
-            "keywords": ";".join(keywords),
-            "summary": summary,
-            "know_id": know_id,
-            "tokens": "",
-            "connectto": "",
-            "addtime": get_str_time(),
-            "page_nums": ",".join(str(page) for page in view.pages),
-            "entities": serialize_entities(entities),
-            "asset_title": "",
-            "extra_metadata": _build_page_extra_metadata(
-                pages=view.pages,
-                image_path_by_page=image_path_by_page,
-            ),
-        }
-        rows.append(row)
-        rows_by_path[leaf.section_path] = row
+    with stage_timer("page_memory.node_rows", node_count=len(views)):
+        for index, view in enumerate(views):
+            leaf = view.leaf
+            content = build_node_content(
+                view,
+                page_owner=page_owner,
+                page_text=resolved_text,
+            )
+            summary, keywords, entities = summaries.get(index, ("", [], []))
+            know_id = f"node_{gen_str_codes(f'{filename}::{leaf.section_path}')}"
+            row = {
+                "content": content,
+                "path": leaf.section_path,
+                "type": "page",
+                "length": len(content),
+                "keywords": ";".join(keywords),
+                "summary": summary,
+                "know_id": know_id,
+                "tokens": "",
+                "connectto": "",
+                "addtime": get_str_time(),
+                "page_nums": ",".join(str(page) for page in view.pages),
+                "entities": serialize_entities(entities),
+                "asset_title": "",
+                "extra_metadata": _build_page_extra_metadata(
+                    pages=view.pages,
+                    image_path_by_page=image_path_by_page,
+                ),
+            }
+            rows.append(row)
+            rows_by_path[leaf.section_path] = row
 
     asset_rows: list[dict[str, Any]] = []
     if page_assets_by_page:

apps/worker/app/services/page_memory/page_tagger.py

@@ -26,6 +26,7 @@
 from app.services.page_memory.page_renderer import PageRenderResult
 from shared.services.ai.prompt_service import build_prompt
 from shared.services.ai.summary.engine import summarize
+from shared.core.exceptions.domain_exceptions import UnavailableException
 
 
 @dataclass
@@ -90,6 +91,8 @@ def tag_pages(
     resolved_max_concurrent = max_concurrent or int(
         getattr(settings, "SUMMARY_LLM_MAX_CONCURRENT", 4)
     )
+    if not pages:
+        return []
 
     def _tag_one(page: PageRenderResult) -> PageTagResult:
         plan = plan_map.get(page.page_index)
@@ -112,9 +115,9 @@ def _tag_one(page: PageRenderResult) -> PageTagResult:
 
     pool = GeventPool(size=min(resolved_max_concurrent, len(pages)))
     greenlets = [pool.spawn(_tag_one, page) for page in pages]
-    gevent.joinall(greenlets)
+    gevent.joinall(greenlets, raise_error=True)
 
-    results = [g.value for g in greenlets if g.value is not None]
+    results = [cast(PageTagResult, g.value) for g in greenlets]
     vlm_calls = sum(1 for r in results if r.strategy_used == "vlm_lite")
     logger.info(
         "[page_tagger] tagged {} pages ({} VLM calls, {} text_only, {} skipped, {} failed) concurrency={}",
@@ -182,6 +185,8 @@ def _tag_text_only(
             response_format={"type": "json_object"},
             usage_task="page_memory.text_tag",
         )
+    except UnavailableException:
+        raise
     except Exception as exc:
         logger.warning(
             "[page_tagger] text_only LLM failed for page {}: {}",
@@ -262,6 +267,7 @@ def tag_page_titles(
     fat_leaf_pages: set[int],
     budget: Any | None = None,
     vlm_model: str | None = None,
+    max_concurrent: int | None = None,
 ) -> list[PageTagResult]:
     """Run independent VLM title detection on fat-leaf pages.
 
@@ -278,6 +284,8 @@ def tag_page_titles(
         Deprecated, ignored. Kept for call-site compatibility.
     vlm_model:
         VLM model name; falls back to ``$IMAGE_MODEL``.
+    max_concurrent:
+        Maximum concurrent title-detection calls.
 
     Returns
     -------
@@ -294,29 +302,57 @@ def tag_page_titles(
 
     tag_map = {t.page_index: t for t in tag_results}
     page_map = {p.page_index: p for p in pages}
-    vlm_calls = 0
-    titles_found = 0
+    work_items = [
+        (page_idx, page, tag_map[page_idx])
+        for page_idx in sorted(fat_leaf_pages)
+        if (page := page_map.get(page_idx)) is not None
+        and page_idx in tag_map
+        and page.image_path
+        and os.path.exists(page.image_path)
+    ]
+    if not work_items:
+        return tag_results
 
-    for page_idx in sorted(fat_leaf_pages):
-        page = page_map.get(page_idx)
-        tag = tag_map.get(page_idx)
-        if page is None or tag is None:
-            continue
+    from shared.core.config import settings
+
+    resolved_max_concurrent = max_concurrent or int(
+        getattr(settings, "PAGE_MEMORY_TITLE_DETECTION_CONCURRENCY", 3)
+    )
 
-        # Skip pages without images (text_only / skip)
-        if not page.image_path or not os.path.exists(page.image_path):
-            continue
+    def _detect_one(
+        page_idx: int,
+        page: PageRenderResult,
+    ) -> tuple[int, list[dict[str, Any]]]:
+        return page_idx, _tag_vlm_titles(page, model=model)
 
-        observed = _tag_vlm_titles(page, model=model)
+    import gevent
+    from gevent.pool import Pool as GeventPool
+
+    pool = GeventPool(size=min(resolved_max_concurrent, len(work_items)))
+    greenlets = [
+        pool.spawn(_detect_one, page_idx, page)
+        for page_idx, page, _tag in work_items
+    ]
+    gevent.joinall(greenlets, raise_error=True)
+
+    title_pairs = [
+        cast(tuple[int, list[dict[str, Any]]], greenlet.value)
+        for greenlet in greenlets
+    ]
+    titles_by_page: dict[int, list[dict[str, Any]]] = dict(title_pairs)
+    for page_idx, _page, tag in work_items:
+        observed = titles_by_page.get(page_idx, [])
         tag.observed_titles = observed
-        vlm_calls += 1
-        titles_found += len(observed)
+
+    vlm_calls = len(work_items)
+    titles_found = sum(len(titles) for titles in titles_by_page.values())
 
     logger.info(
-        "[page_tagger] title detection: {} VLM calls on {} fat-leaf pages, {} titles found",
+        "[page_tagger] title detection: {} VLM calls on {} fat-leaf pages, {} titles found concurrency={}",
         vlm_calls,
         len(fat_leaf_pages),
         titles_found,
+        resolved_max_concurrent,
     )
     return tag_results
 
@@ -413,6 +449,8 @@ def _tag_vlm_titles(
                 page.page_index,
             )
             return []
+        except UnavailableException:
+            raise
         except Exception as exc:
             logger.warning(
                 "[page_tagger] title VLM failed for page {}: {}",