fix/wuchengke: optimize document parser image cleanup and agentic retrieval logic

Author: EricNGOntos

apps/worker/app/services/document_parser/doc_parser.py

@@ -608,6 +608,21 @@ def iter_block_items(doc_data):
                 # <v:imagedata> piece-by-piece loses textual overlay and positioning. 
                 # Future plan: Use LibreOffice headless conversion to render the entire document 
                 # and map the perfectly rendered images back to the layout via text anchors.
+                #
+                # Temporary: detect VML-only paragraphs and inject a placeholder so the
+                # paragraph isn't silently swallowed, leaving its parent section empty.
+                if not text and not seen_rids:
+                    # No text and no DrawingML images — check for VML content
+                    vml_groups = elem.xpath(".//v:group", namespaces=ns)
+                    vml_images_check = elem.xpath(".//v:imagedata", namespaces=ns)
+                    if vml_groups or vml_images_check:
+                        vml_placeholder = "[VML graphic \u2014 extraction not yet supported]"
+                        yield ele_num, vml_placeholder, "PTXT", None
+                        ele_num += 1
+                        logger.debug(
+                            f"Injected VML placeholder for paragraph with "
+                            f"{len(vml_groups)} v:group, {len(vml_images_check)} v:imagedata"
+                        )
                 """
                 # images (VML: <v:imagedata>) — convert to PNG
                 from PIL import Image as PILImage
@@ -942,6 +957,13 @@ def convert_doc2dics(
     for _, row in leaf_dics.iterrows():
         key = row["path_identifier"]
 
+        # Skip leaf nodes with no actual content (empty heading-only sections)
+        content_lst = row["content_lst"]
+        joined = "\n".join(content_lst).strip()
+        if not joined:
+            logger.debug(f"Skipping empty leaf node: {key}")
+            continue
+
         # Build tentative path to check for duplicates
         tentative_path = doc_name + split_char + key
 
@@ -954,7 +976,7 @@ def convert_doc2dics(
             path_counter[tentative_path] = 1
 
         path_keys.append((doc_name + split_char + key))
-        bottom_content = "\n".join(row["content_lst"])
+        bottom_content = joined
         bottom_tokens = tokenize2stw_remove(
             [bottom_content], base_llm_paras["stopwords"]
         )

apps/worker/app/services/document_parser/md_parser.py

@@ -318,7 +318,10 @@ def parse_md(
     os.makedirs(tb_dir, exist_ok=True)
     img_dir = os.path.join(output_dir, "images")
     if os.path.isdir(img_dir):
-        shutil.rmtree(img_dir)
+        # Only remove parse_md's own output (image-N-*) from previous runs
+        for fname in os.listdir(img_dir):
+            if re.match(r"^image-\d+", fname):
+                os.remove(os.path.join(img_dir, fname))
     os.makedirs(img_dir, exist_ok=True)
 
     # initialize vars

packages/shared-python/shared/services/retrieval/__init__.py

@@ -1,4 +1,4 @@
-from .app_service import list_lexical_chunks, merge_channels_rrf, run_retrieval_query
+from .app_service import merge_channels_rrf, run_retrieval_query
 from .cache_service import (
     bump_retrieval_namespace_cache_version,
     get_cached_retrieval_query_result,
@@ -13,7 +13,6 @@
 __all__ = [
     "create_retrieval_llm_fn",
     "run_retrieval_query",
-    "list_lexical_chunks",
     "merge_channels_rrf",
     "DocumentGraphService",
     "GraphQueryService",

packages/shared-python/shared/services/retrieval/agent_navigate.py

@@ -56,17 +56,19 @@
 User query: {query}
 
 Select the most relevant section paths (at most {max_select}).
+If NO section path is relevant to the query, you MUST return an empty array []. Do not force-select irrelevant sections.
 Prefer specific sub-items over broad parents when both are listed and the sub-item is sufficient.
 
-For each selected path, choose a hydrate_mode:
+For each selected path, assign a confidence score (0.0 to 1.0) where 1.0 means exactly answers the query and 0.5 means tangentially related.
+Also choose a hydrate_mode:
 - "chunks"       (default) return all text/image/table chunks
 - "outline"      return only section title + summary, no chunk content
 - "assets_only"  return only image and table chunks
 - "image_only"   return only image chunks
 - "table_only"   return only table chunks
 
 Return ONLY a JSON array:
-[{{"path": "section/path", "confidence": 0.9, "hydrate_mode": "chunks"}}, ...]
+[{{"path": "section/path", "confidence": <float>, "hydrate_mode": "chunks"}}, ...]
 Do not include any explanation.
 """
 

packages/shared-python/shared/services/retrieval/agentic/orchestrator.py

@@ -130,7 +130,27 @@ async def run(
             'channel_weights': channel_weights,
         }
 
-        # ── Agent loop ──
+        # ── Mandatory pre-step: bottom discovery ─────────────────────────────
+        # bottom_discovery is always the first action; running it via the LLM
+        # policy wastes ~1-2s on a trivial LLM call.  We execute it directly
+        # and let the LLM loop start from step 2 (kg_document_select etc.).
+        logger.info('  agentic: running mandatory bottom_discovery pre-step')
+        discovery_result = await self._execute_tool(
+            db, ActionType.BOTTOM_DISCOVERY, state, config, **tool_kwargs,
+        )
+        state.apply(ActionType.BOTTOM_DISCOVERY, discovery_result)
+        if trace_enabled:
+            trace.record_step(
+                ActionType.BOTTOM_DISCOVERY, discovery_result,
+                decision_reason='mandatory_pre_step',
+            )
+        state.step_count += 1
+        logger.info(
+            f'  agentic step {state.step_count} (pre-step): action=bottom_discovery '
+            f'status={discovery_result.status} latency={discovery_result.latency_ms}ms'
+        )
+
+        # ── Agent loop (LLM decisions start from here) ────────────────────────
         stop_reason = 'max_steps'
         while state.step_count < config.max_steps:
             if state.elapsed_ms >= config.latency_budget_ms:
@@ -143,13 +163,9 @@ async def run(
                 break
 
             if policy is None:
-                # No LLM: run discovery once then stop
-                if not state.discovery_done:
-                    action_type = ActionType.BOTTOM_DISCOVERY
-                    decision_reason = 'no_llm_fn: discovery only'
-                else:
-                    stop_reason = 'no_llm_fn'
-                    break
+                # No LLM: discovery already ran — stop
+                stop_reason = 'no_llm_fn'
+                break
             else:
                 action_type, decision_reason = await policy.decide(state, config)
 
@@ -244,6 +260,26 @@ async def run(
                 state.kg_done = False
                 state.discovery_done = False
 
+                # Mandatory bottom_discovery pre-step for revision round
+                logger.info(
+                    f'  agentic: running mandatory bottom_discovery pre-step '
+                    f'(revision {state.revision_count})'
+                )
+                rev_discovery = await self._execute_tool(
+                    db, ActionType.BOTTOM_DISCOVERY, state, config, **tool_kwargs,
+                )
+                state.apply(ActionType.BOTTOM_DISCOVERY, rev_discovery)
+                if trace_enabled:
+                    trace.record_step(
+                        ActionType.BOTTOM_DISCOVERY, rev_discovery,
+                        decision_reason=f'mandatory_pre_step (revision {state.revision_count})',
+                    )
+                state.step_count += 1
+                logger.info(
+                    f'  agentic step {state.step_count} (rev {state.revision_count} pre-step): '
+                    f'action=bottom_discovery status={rev_discovery.status}'
+                )
+
                 # Re-enter agent loop
                 while state.step_count < config.max_steps:
                     if state.elapsed_ms >= config.latency_budget_ms:

packages/shared-python/shared/services/retrieval/agentic/policy.py

@@ -23,15 +23,11 @@
 
 # ── Available actions presented to the LLM ───────────────────────────────────
 
+# NOTE: BOTTOM_DISCOVERY is intentionally excluded from this list.
+# It is now a mandatory pre-step executed automatically by the orchestrator
+# before the LLM decision loop begins.  The LLM should never need to decide
+# whether to run it — doing so wastes one LLM call per run.
 _AVAILABLE_ACTIONS: list[dict[str, Any]] = [
-    {
-        'action': ActionType.BOTTOM_DISCOVERY.value,
-        'description': (
-            'Run BM25 3-channel bottom-layer discovery (path / content / term). '
-            'Always call this first to get candidate chunks and top document hints.'
-        ),
-        'when': 'discovery_done is false',
-    },
     {
         'action': ActionType.KG_DOCUMENT_SELECT.value,
         'description': (
@@ -94,12 +90,12 @@
 {actions_block}
 
 RULES:
-1. Always run bottom_discovery first (if discovery_done is false).
-2. After discovery, run kg_document_select (if kg_done is false).
-3. After kg select, run document_path_select for each pending document.
-4. Call done when all pending documents are processed OR you have >= {min_evidence} evidence paths.
-5. Only use grep_document_discover if kg_document_select found 0 documents.
-6. Only use graph_expand_docs if you need more related docs after reviewing results.
+1. Run kg_document_select when discovery_done is true and kg_done is false.
+2. After kg select, run document_path_select for each pending document.
+3. Call done when all pending documents are processed OR you have >= {min_evidence} evidence paths.
+4. Only use grep_document_discover if kg_document_select found 0 documents.
+5. Only use graph_expand_docs if you need more related docs after reviewing results.
+Note: bottom_discovery is already executed automatically before this loop — do NOT attempt to call it.
 
 Return ONLY a JSON object, no markdown, no explanation:
 {{"action": "<action_name>", "reason": "<one sentence why>"}}
@@ -145,9 +141,26 @@ def __init__(self, llm_fn: LLMFn, *, query: str = '') -> None:
     def build_prompt(self, state: AgentState, config: AgentRunConfig) -> str:
         """Build the decision prompt.  Public for test inspection."""
         state_data = state.state_summary()
+        
+        has_pending_docs = state.pending_doc_index < len(state.selected_docs)
+        state_data['has_pending_docs'] = has_pending_docs
         state_json = json.dumps(state_data, ensure_ascii=False, indent=2)
 
-        # Count pending docs
+        allowed_actions = []
+        for action in _AVAILABLE_ACTIONS:
+            name = action['action']
+            if name == ActionType.KG_DOCUMENT_SELECT.value and (not state.discovery_done or state.kg_done):
+                continue
+            if name == ActionType.DOCUMENT_PATH_SELECT.value and (not state.kg_done or not has_pending_docs):
+                continue
+            if name == ActionType.GREP_DOCUMENT_DISCOVER.value and (not state.kg_done or len(state.selected_docs) > 0):
+                continue
+            allowed_actions.append(action)
+
+        actions_block = '\n'.join(
+            f"  {i+1}. \"{a['action']}\": {a['description']} [{a['when']}]"
+            for i, a in enumerate(allowed_actions)
+        )
 
         return _POLICY_PROMPT_TEMPLATE.format(
             query=self._query,
@@ -156,7 +169,7 @@ def build_prompt(self, state: AgentState, config: AgentRunConfig) -> str:
             budget_ms=config.latency_budget_ms,
             step=state.step_count,
             max_steps=config.max_steps,
-            actions_block=_ACTIONS_BLOCK,
+            actions_block=actions_block,
             min_evidence=config.min_evidence_paths,
         )
 

packages/shared-python/shared/services/retrieval/app_service.py

@@ -251,89 +251,7 @@ async def assemble_retrieval_results(
     return assembled
 
 
-async def list_lexical_chunks(
-    db: AsyncSession,
-    *,
-    user_id: str,
-    namespace: str,
-    query: str,
-    top_k: int,
-    exclude_document_ids: list[str],
-    exclude_sections: list[dict[str, str]],
-) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
-    """Independent lexical retrieval: path + content channels via ILIKE."""
-    recall_k = top_k * _INTERNAL_RECALL_K_MULTIPLIER
-    excluded_docs = set(exclude_document_ids)
-
-    base_stmt = (
-        select(Document, DocumentChunk, DocumentSection, JobResult)
-        .join(DocumentChunk, (DocumentChunk.document_id == Document.document_id) & (DocumentChunk.job_result_id == Document.current_job_result_id))
-        .outerjoin(DocumentSection, DocumentSection.section_id == DocumentChunk.section_id)
-        .join(JobResult, JobResult.id == DocumentChunk.job_result_id)
-        .where(Document.user_id == user_id)
-        .where(Document.namespace == namespace)
-        .where(Document.status == 'active')
-    )
-    if excluded_docs:
-        base_stmt = base_stmt.where(Document.document_id.notin_(list(excluded_docs)))
-
-    like = f'%{query}%'
-    content_stmt = base_stmt.where(DocumentChunk.content_lexical_text.ilike(like)).order_by(DocumentChunk.sort_order).limit(recall_k)
-    path_stmt = base_stmt.where(DocumentChunk.path_lexical_text.ilike(like)).order_by(DocumentChunk.sort_order).limit(recall_k)
-
-    # AsyncSession is stateful and should not be shared across concurrent tasks.
-    content_result = await db.execute(content_stmt)
-    path_result = await db.execute(path_stmt)
-
-    def _to_rows(result, channel_score: float) -> list[dict[str, Any]]:
-        rows: list[dict[str, Any]] = []
-        for document, chunk, section, job_result in result.all():
-            section_path = section.section_path if section else None
-            if is_excluded_section(document_id=document.document_id, section_path=section_path, exclude_sections=exclude_sections):
-                continue
-            rows.append({
-                'document_id': document.document_id,
-                'chunk_id': chunk.chunk_id,
-                'section_id': chunk.section_id,
-                'section_path': section_path,
-                'source_file_name': document.source_file_name,
-                'chunk_type': chunk.chunk_type,
-                'content': chunk.content,
-                'score': channel_score,
-                'file_path': chunk.file_path,
-                'chunk_metadata': chunk.chunk_metadata or {},
-                'job_result_id': chunk.job_result_id,
-                'job_id': job_result.job_id if job_result else None,
-            })
-        return rows
-
-    content_rows = _to_rows(content_result, _CHANNEL_WEIGHT_CONTENT)
-    path_rows = _to_rows(path_result, _CHANNEL_WEIGHT_PATH)
-    return content_rows, path_rows
-
 
-def _grep_search_rows(rows: list[dict[str, Any]], query: str) -> list[dict[str, Any]]:
-    """Term/grep channel: exact substring matching with scoring from knowhere-kb."""
-    import re
-    query_lower = query.lower().strip()
-    if not query_lower:
-        return []
-
-    units = re.findall(r'[一-鿿]+|[a-zA-Z0-9]+', query_lower)
-    units = [u for u in units if len(u) > 1]
-
-    scored: list[tuple[float, dict[str, Any]]] = []
-    for row in rows:
-        haystack = (str(row.get('content') or '') + ' ' + str(row.get('section_path') or '')).lower()
-        if query_lower in haystack:
-            scored.append((100.0, row))
-        elif units:
-            hit_count = sum(1 for u in units if u in haystack)
-            if hit_count > 0:
-                scored.append((float(hit_count), row))
-
-    scored.sort(key=lambda x: x[0], reverse=True)
-    return [dict(row, score=score) for score, row in scored]
 
 
 def _merge_same_section_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
@@ -690,13 +608,14 @@ def _rank_candidates_by_path(
         else:
             primary_rows.append(row)
 
-    _sort_key = lambda row: (
-        float(row.get('agent_score', 0.0) or 0.0),
-        float(row.get('discovery_score', 0.0) or 0.0),
-        int(row.get('dual_hit_flag', 0) or 0),
-        float(row.get('importance_norm_score', 0.0) or 0.0),
-        -int(row.get('_candidate_order', 0) or 0),
-    )
+    def _sort_key(row):
+        return (
+            float(row.get('agent_score', 0.0) or 0.0),
+            float(row.get('discovery_score', 0.0) or 0.0),
+            int(row.get('dual_hit_flag', 0) or 0),
+            float(row.get('importance_norm_score', 0.0) or 0.0),
+            -int(row.get('_candidate_order', 0) or 0),
+        )
 
     primary_rows.sort(key=_sort_key, reverse=True)
     ranked_rows = primary_rows[:top_k]