From 3a6b98f488d9515aa66cb629916f6fa36b6c5577 Mon Sep 17 00:00:00 2001 From: chengke <404835780@qq.com> Date: Tue, 21 Jul 2026 10:43:48 +0800 Subject: [PATCH 1/3] feat: enhance document navigation summaries with top_summary persistence - Introduced a new function to persist the document-level top_summary in doc_nav.json, ensuring it is saved after enrichment. - Updated `enrich_doc_nav_summaries` to support an opt-in flag for using LLM for top_summary generation, defaulting to true. - Refactored existing logic to prioritize loading enriched summaries from disk, improving efficiency and reducing unnecessary recomputation. - Adjusted related tests to validate the new functionality and ensure correct behavior of top_summary persistence. This update enhances the document navigation process by ensuring that summaries are consistently saved and retrieved, improving overall user experience. --- .../connect_builder/summary_builder.py | 93 ++++++++++----- .../success_finalization.py | 33 +++--- .../formats/markdown/deferred_summary.py | 54 +++++++-- .../formats/markdown/deferred_task.py | 1 - .../formats/markdown/table_asset.py | 1 - .../test_markdown_table_rename_contract.py | 109 ++++++++++++++++++ .../test_page_memory_navigation_contract.py | 36 ++++++ .../worker/tests/unit/test_summary_builder.py | 69 +++++++++++ .../shared/models/schemas/job.py | 8 ++ .../services/jobs/lifecycle/publication.py | 2 + .../shared/services/jobs/lifecycle/service.py | 2 + .../jobs/lifecycle/success_finalizer.py | 2 + .../services/retrieval/graph/service.py | 9 +- .../services/retrieval/publication_service.py | 10 +- .../services/storage/zip_chunk_schema.py | 3 - .../services/storage/zip_result_service.py | 32 +++++ 16 files changed, 404 insertions(+), 60 deletions(-) create mode 100644 apps/worker/tests/contract/test_markdown_table_rename_contract.py diff --git a/apps/worker/app/services/connect_builder/summary_builder.py b/apps/worker/app/services/connect_builder/summary_builder.py index f55dec47d..bd1d327da 100644 --- a/apps/worker/app/services/connect_builder/summary_builder.py +++ b/apps/worker/app/services/connect_builder/summary_builder.py @@ -3,6 +3,8 @@ Reads doc_nav.json and in-memory chunks, generates ``summary`` at every non-leaf via deterministic covers assembly or LLM aggregation (with self_only). +Persists document-level ``top_summary`` on doc_nav (default LLM); section-level +LLM remains opt-in via ``use_llm``. Usage (standalone): from app.services.connect_builder.summary_builder import enrich_doc_nav_summaries @@ -400,11 +402,29 @@ def _build_nav_top_summary( ) +def _persist_doc_nav_top_summary( + *, + file_dir: str, + doc_nav: Dict[str, Any], + top_summary: str, +) -> str: + """Write document-level top_summary once onto doc_nav and save.""" + cleaned = str(top_summary or "").strip() + if cleaned: + doc_nav["top_summary"] = cleaned + elif "top_summary" in doc_nav: + doc_nav.pop("top_summary", None) + _save_doc_nav(file_dir, doc_nav) + return cleaned + + def enrich_doc_nav_summaries( document_workspace_dir: str, source_file: Optional[str] = None, force: bool = False, - use_llm: bool = True, + use_llm: bool = False, + *, + top_summary_use_llm: bool = True, chunks: Optional[List[Dict[str, Any]]] = None, ) -> Dict[str, str]: """Enrich doc_nav.json with bottom-up recursive summaries. @@ -413,14 +433,16 @@ def enrich_doc_nav_summaries( document_workspace_dir: Absolute path to the temporary document workspace. source_file: If given, only process this file. Otherwise process all. force: If True, regenerate even if summaries already exist. - use_llm: If True, use LLM when contribution length sum exceeds threshold. + use_llm: Section-level LLM summaries (default off). + top_summary_use_llm: Document-level top summary LLM (default on). chunks: In-memory parse chunks for exact-path self_only extraction. Returns: Dict mapping file_name → top-level summary string. """ results: Dict[str, str] = {} - mode_label = "LLM" if use_llm else "title-concat" + section_mode = "LLM" if use_llm else "title-concat" + top_mode = "LLM" if top_summary_use_llm else "title-concat" if source_file: targets = [source_file] @@ -445,21 +467,39 @@ def enrich_doc_nav_summaries( source_file_name=source_file_name, ) - if not force and _doc_nav_has_enriched_summaries(doc_nav): + existing_top = str(doc_nav.get("top_summary") or "").strip() + if ( + not force + and _doc_nav_has_enriched_summaries(doc_nav) + and existing_top + ): logger.debug( f"Summaries already exist in {DOC_NAV_FILENAME} for {file_name}, skipping" ) - results[file_name] = _build_nav_top_summary( + results[file_name] = existing_top + continue + + if not force and _doc_nav_has_enriched_summaries(doc_nav): + logger.info( + f"📝 Building missing {DOC_NAV_FILENAME} top_summary for {file_name} " + f"(top_mode={top_mode})" + ) + top_summary = _build_nav_top_summary( doc_nav, - use_llm=use_llm, + use_llm=top_summary_use_llm, self_only_lookup=self_only_lookup, source_file_name=source_file_name, ) + results[file_name] = _persist_doc_nav_top_summary( + file_dir=file_dir, + doc_nav=doc_nav, + top_summary=top_summary, + ) continue logger.info( f"📝 Enriching {DOC_NAV_FILENAME} summaries for {file_name} " - f"(mode={mode_label})" + f"(section_mode={section_mode}, top_mode={top_mode})" ) for section in doc_nav.get("sections", []): @@ -470,31 +510,36 @@ def enrich_doc_nav_summaries( source_file_name=source_file_name, ) - _save_doc_nav(file_dir, doc_nav) - logger.info(f"✅ doc_nav summaries saved for {file_name}") - top_summary = _build_nav_top_summary( doc_nav, - use_llm=use_llm, + use_llm=top_summary_use_llm, self_only_lookup=self_only_lookup, source_file_name=source_file_name, ) - results[file_name] = top_summary + results[file_name] = _persist_doc_nav_top_summary( + file_dir=file_dir, + doc_nav=doc_nav, + top_summary=top_summary, + ) + logger.info(f"✅ doc_nav summaries saved for {file_name}") return results def load_nav_top_summary(file_dir: str, file_name: str = "") -> str: - """Load doc_nav.json and extract the navigation top summary.""" + """Load persisted doc_nav top_summary, with deterministic fallback.""" doc_nav = _load_doc_nav(file_dir) - if doc_nav is not None: - source_file_name = file_name or str(doc_nav.get("file_name") or "") - return _build_nav_top_summary( - doc_nav, - use_llm=False, - source_file_name=source_file_name, - ) - return "" + if doc_nav is None: + return "" + existing = str(doc_nav.get("top_summary") or "").strip() + if existing: + return existing + source_file_name = file_name or str(doc_nav.get("file_name") or "") + return _build_nav_top_summary( + doc_nav, + use_llm=False, + source_file_name=source_file_name, + ) def build_section_summary_lookup(file_dir: str) -> Dict[str, str]: @@ -530,11 +575,7 @@ def _walk(node: Dict[str, Any]) -> None: _walk(section) if "Root" not in lookup: - top_summary = _build_nav_top_summary( - doc_nav, - use_llm=False, - source_file_name=source_file_name, - ) + top_summary = load_nav_top_summary(file_dir, source_file_name) if top_summary: lookup["Root"] = top_summary diff --git a/apps/worker/app/services/document_ingestion/success_finalization.py b/apps/worker/app/services/document_ingestion/success_finalization.py index 29ecd4be3..9acc0e3ab 100644 --- a/apps/worker/app/services/document_ingestion/success_finalization.py +++ b/apps/worker/app/services/document_ingestion/success_finalization.py @@ -48,7 +48,6 @@ def finalize_parse_success( job_context=job_context, source_file_name=source_file_name, ) - _attach_document_top_summary(result_package.chunks, document_top_summary) _refresh_processing_stages(job_context) lifecycle_service.update_progress( @@ -92,6 +91,7 @@ def finalize_parse_success( stored_count=stored_count, delivery_mode="url", section_summaries=section_summaries, + document_top_summary=document_top_summary, ) if finalization_response.get("status") != "success": logger.error( @@ -137,6 +137,7 @@ def _enrich_document_navigation( ) -> tuple[str, dict[str, str]]: document_top_summary = "" section_summaries: dict[str, str] = {} + enrich_results: dict[str, str] = {} add_dir = artifact.add_dir parsed_contents_df = artifact.dataframe if add_dir and source_file_name: @@ -153,16 +154,27 @@ def _enrich_document_navigation( "summary_use_llm", False, ) - enrich_doc_nav_summaries( + top_summary_use_llm = JobMetadataHelper.get_parsing_param( + job_context.job_metadata, + "top_summary_use_llm", + True, + ) + enrich_results = enrich_doc_nav_summaries( document_root_for_enrich, source_file=source_file_name, use_llm=summary_use_llm, + top_summary_use_llm=top_summary_use_llm, chunks=chunks, ) section_summaries = build_section_summary_lookup(str(add_dir)) except Exception as exc: logger.warning(f"doc_nav enrichment failed (non-fatal): {exc}") - document_top_summary = load_nav_top_summary(str(add_dir), source_file_name) + enrich_results = {} + document_top_summary = str( + enrich_results.get(source_file_name) or "" + ).strip() + if not document_top_summary: + document_top_summary = load_nav_top_summary(str(add_dir), source_file_name) return document_top_summary, section_summaries @@ -181,21 +193,6 @@ def _refresh_processing_stages(job_context: ParseJobContext) -> None: job_context.job_metadata["stages"] = stages -def _attach_document_top_summary( - chunks: list[dict[str, Any]], - document_top_summary: str, -) -> None: - if not document_top_summary: - return - - for chunk in chunks: - metadata = chunk.get("metadata") - if not isinstance(metadata, dict): - metadata = {} - chunk["metadata"] = metadata - metadata["document_top_summary"] = document_top_summary - - def _record_processing_completion( *, job_id: str, diff --git a/apps/worker/app/services/document_parser/formats/markdown/deferred_summary.py b/apps/worker/app/services/document_parser/formats/markdown/deferred_summary.py index d946db4f1..e5b5bf90a 100644 --- a/apps/worker/app/services/document_parser/formats/markdown/deferred_summary.py +++ b/apps/worker/app/services/document_parser/formats/markdown/deferred_summary.py @@ -83,20 +83,46 @@ def apply_markdown_deferred_summaries( def replace_chunk_ref_in_rows( rows: list[list[str | int]], old_path: str, new_path: str ) -> None: + """Rewrite asset paths after deferred rename. + + Text/image rows store bracketed refs (``[tables/...]`` / ``[images/...]``). + Table rows store the bare relative path as ``content``. Both forms must move + with the renamed file; historically only the bracketed form was updated, + leaving table ``content`` stuck on the pre-rename path. + """ + if not old_path or old_path == new_path: + return + old_ref = build_chunk_ref(old_path) new_ref = build_chunk_ref(new_path) - if not old_ref or old_ref == new_ref: - return for row in rows: if len(row) > 0 and isinstance(row[0], str): - row[0] = row[0].replace(old_ref, new_ref) + updated = row[0] + if old_ref and new_ref and old_ref != new_ref: + updated = updated.replace(old_ref, new_ref) + # Table asset rows use the bare path as content (no brackets). + if updated == old_path: + updated = new_path + elif old_path in updated: + updated = updated.replace(old_path, new_path) + row[0] = updated if len(row) > 1 and row[1] == old_path: row[1] = new_path if len(row) > 2 and isinstance(row[2], str): - row[2] = row[2].replace(old_ref, new_ref) + updated_type = row[2] + if old_ref and new_ref and old_ref != new_ref: + updated_type = updated_type.replace(old_ref, new_ref) + if old_path in updated_type: + updated_type = updated_type.replace(old_path, new_path) + row[2] = updated_type if len(row) > 8 and isinstance(row[8], str): - row[8] = row[8].replace(old_ref, new_ref) + updated_connect = row[8] + if old_ref and new_ref and old_ref != new_ref: + updated_connect = updated_connect.replace(old_ref, new_ref) + if old_path in updated_connect: + updated_connect = updated_connect.replace(old_path, new_path) + row[8] = updated_connect def _run_deferred_summary_tasks( @@ -285,10 +311,20 @@ def _apply_table_summary_result( table_dir = original_task.table_dir old_table_name = original_task.table_name - table_count = original_task.table_count + # Keep the original table-N index (same pattern as image rename). Do not + # re-derive from a separate counter — the legacy ``table_count - 1`` path + # produced off-by-one filenames vs text-chunk refs. + table_num_match = re.match(r"table-(\d+)", str(old_table_name)) + table_num = ( + table_num_match.group(1) + if table_num_match + else str(old_table_name).split("-")[1] + if "-" in str(old_table_name) + else "0" + ) safe_title = sanitize_table_name_from_header(str(title)) new_table_name = path_handle( - f"table-{table_count} {safe_title}", mode="clean_single" + f"table-{table_num} {safe_title}", mode="clean_single" ) old_path = os.path.join(table_dir, f"{old_table_name}.html") new_path = os.path.join(table_dir, f"{new_table_name}.html") @@ -296,6 +332,8 @@ def _apply_table_summary_result( return os.rename(old_path, new_path) + old_relative_path = str(row[1]) if len(row) > 1 else f"tables/{old_table_name}.html" new_relative_path = f"tables/{new_table_name}.html" - replace_chunk_ref_in_rows(rows, str(row[1]), new_relative_path) + replace_chunk_ref_in_rows(rows, old_relative_path, new_relative_path) + row[0] = new_relative_path row[1] = new_relative_path diff --git a/apps/worker/app/services/document_parser/formats/markdown/deferred_task.py b/apps/worker/app/services/document_parser/formats/markdown/deferred_task.py index f2cb82409..f26df1683 100644 --- a/apps/worker/app/services/document_parser/formats/markdown/deferred_task.py +++ b/apps/worker/app/services/document_parser/formats/markdown/deferred_task.py @@ -19,7 +19,6 @@ class TableDeferredSummaryTask: table_html: str table_dir: str table_name: str - table_count: int @dataclass(frozen=True) diff --git a/apps/worker/app/services/document_parser/formats/markdown/table_asset.py b/apps/worker/app/services/document_parser/formats/markdown/table_asset.py index 89558cb1e..dc7808f38 100644 --- a/apps/worker/app/services/document_parser/formats/markdown/table_asset.py +++ b/apps/worker/app/services/document_parser/formats/markdown/table_asset.py @@ -69,7 +69,6 @@ def build_markdown_table_asset( table_html=request.table_html, table_dir=request.table_dir, table_name=table_name, - table_count=request.table_count - 1, ) return MarkdownTableAsset( diff --git a/apps/worker/tests/contract/test_markdown_table_rename_contract.py b/apps/worker/tests/contract/test_markdown_table_rename_contract.py new file mode 100644 index 000000000..6360181b9 --- /dev/null +++ b/apps/worker/tests/contract/test_markdown_table_rename_contract.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +import os +from pathlib import Path + +os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test") +os.environ.setdefault("TMP_PATH", "/tmp/knowhere-test") +os.environ.setdefault("S3_BUCKET_NAME", "test-uploads") +os.environ.setdefault("S3_ACCESS_KEY_ID", "test") +os.environ.setdefault("S3_SECRET_ACCESS_KEY", "test") +os.environ.setdefault("S3_TEMP_PATH", "/tmp") + +from app.services.document_parser.formats.markdown.deferred_summary import ( # noqa: E402 + replace_chunk_ref_in_rows, + _apply_table_summary_result, +) +from app.services.document_parser.formats.markdown.deferred_task import ( # noqa: E402 + TableDeferredSummaryTask, +) +from app.services.document_parser.formats.markdown.table_asset import ( # noqa: E402 + MarkdownTableAssetRequest, + build_markdown_table_asset, +) +from shared.services.ai.summary.model import AssetSummary, Entity # noqa: E402 +from shared.utils.chunk_refs import build_chunk_ref # noqa: E402 + + +def test_replace_chunk_ref_updates_bare_table_content_and_bracketed_text() -> None: + old_path = "tables/table-12 流程名称 招标文件.html" + new_path = "tables/table-12 招标文件及议标项目评审流程图.html" + rows: list[list[str | int]] = [ + [old_path, old_path, "table", 10, "", "table-12", "id", "", ""], + [ + f"see {build_chunk_ref(old_path)}", + "doc/section", + "ptxt", + 20, + "", + "", + "text-id", + "", + "", + ], + ] + + replace_chunk_ref_in_rows(rows, old_path, new_path) + + assert rows[0][0] == new_path + assert rows[0][1] == new_path + assert rows[1][0] == f"see {build_chunk_ref(new_path)}" + + +def test_table_deferred_task_has_no_legacy_count_field(tmp_path: Path) -> None: + asset = build_markdown_table_asset( + MarkdownTableAssetRequest( + table_html="
流程名称招标文件
", + table_dir=str(tmp_path), + table_count=12, + timestamp="2026-07-21 00:00:00", + summary_table=True, + row_index=0, + ) + ) + + assert asset.deferred_task is not None + assert isinstance(asset.deferred_task, TableDeferredSummaryTask) + assert not hasattr(asset.deferred_task, "table_count") + assert asset.deferred_task.table_name.startswith("table-12 ") + + +def test_apply_table_summary_rename_keeps_index_and_syncs_content( + tmp_path: Path, +) -> None: + old_stem = "table-12 流程名称 招标文件及议标项目评审流程图 流程编号" + old_relative = f"tables/{old_stem}.html" + old_file = tmp_path / f"{old_stem}.html" + old_file.write_text("
x
", encoding="utf-8") + + text_content = f"\n{build_chunk_ref(old_relative)}\n" + rows: list[list[str | int]] = [ + [old_relative, old_relative, "table", len(old_relative), "", "table-12", "t", "", ""], + [text_content, "doc/3、招标文件评审流程运行图", "ptxt", len(text_content), "", "", "x", "", ""], + ] + task = TableDeferredSummaryTask( + row_index=0, + table_html="
x
", + table_dir=str(tmp_path), + table_name=old_stem, + ) + + _apply_table_summary_result( + rows, + task, + 0, + AssetSummary( + title="招标文件及议标项目评审流程图", + summary="三类风险招标文件评审流程", + entities=[Entity(text="市场开发部", type="org")], + kind="table", + ), + ) + + new_relative = str(rows[0][1]) + assert new_relative.startswith("tables/table-12 ") + assert "招标文件及议标项目评审流程图" in new_relative + assert rows[0][0] == new_relative + assert build_chunk_ref(new_relative) in str(rows[1][0]) + assert not old_file.exists() + assert (tmp_path / Path(new_relative).name).exists() diff --git a/apps/worker/tests/contract/test_page_memory_navigation_contract.py b/apps/worker/tests/contract/test_page_memory_navigation_contract.py index 23e60c289..47913989c 100644 --- a/apps/worker/tests/contract/test_page_memory_navigation_contract.py +++ b/apps/worker/tests/contract/test_page_memory_navigation_contract.py @@ -44,6 +44,7 @@ def test_page_doc_nav_uses_path_based_leaf_summaries_and_page_counts() -> None: def test_zip_result_service_builds_navigation_from_chunk_paths() -> None: doc_nav, hierarchy = ZipResultService()._build_navigation_outputs( # noqa: SLF001 + add_dir="", formatted_chunks=_page_chunks(), source_file_name="demo.pdf", ) @@ -57,6 +58,41 @@ def test_zip_result_service_builds_navigation_from_chunk_paths() -> None: } +def test_zip_result_service_prefers_enriched_on_disk_doc_nav(tmp_path) -> None: + enriched = { + "version": "1.0", + "file_name": "demo.pdf", + "top_summary": "Document overview from enrich", + "stats": {}, + "sections": [ + { + "title": "Kept From Disk", + "path": "demo.pdf/Kept From Disk", + "summary": "enriched section", + "chunk_count": 1, + "children": [], + } + ], + "resources": {"images": [], "tables": []}, + } + (tmp_path / "doc_nav.json").write_text( + __import__("json").dumps(enriched, ensure_ascii=False), + encoding="utf-8", + ) + + doc_nav, hierarchy = ZipResultService()._build_navigation_outputs( # noqa: SLF001 + add_dir=str(tmp_path), + formatted_chunks=_page_chunks(), + source_file_name="demo.pdf", + ) + + assert doc_nav is not None + assert doc_nav["top_summary"] == "Document overview from enrich" + assert doc_nav["sections"][0]["title"] == "Kept From Disk" + assert hierarchy == {"Kept From Disk": {}} + + + def _page_chunks() -> list[dict[str, object]]: return [ { diff --git a/apps/worker/tests/unit/test_summary_builder.py b/apps/worker/tests/unit/test_summary_builder.py index c0685ef2a..c6364c431 100644 --- a/apps/worker/tests/unit/test_summary_builder.py +++ b/apps/worker/tests/unit/test_summary_builder.py @@ -231,3 +231,72 @@ def chat_completion(self, **kwargs: Any) -> str: assert "[ChildB] ChildB" in user # legacy flat blob prompt removed assert "You will receive summaries of sub-sections" not in user + + +class TestDocNavTopSummaryPersistence: + def test_enrich_persists_top_summary_and_defaults_top_llm( + self, tmp_path, monkeypatch: pytest.MonkeyPatch + ) -> None: + import json + + from app.services.connect_builder.summary_builder import ( + enrich_doc_nav_summaries, + load_nav_top_summary, + ) + + captured: Dict[str, Any] = {} + + def _fake_llm(**kwargs: Any) -> str: + captured["is_top"] = kwargs.get("node_name") == "Document Overview" + captured["max_tokens"] = kwargs.get("max_tokens") + return "LLM document overview" + + monkeypatch.setattr( + "app.services.connect_builder.summary_builder._llm_summarize", + _fake_llm, + ) + + file_dir = tmp_path / "report.pdf" + file_dir.mkdir() + long_leaf = "L" * (SUMMARY_MAX_LEN + 5) + doc_nav = { + "version": "1.0", + "file_name": "report.pdf", + "stats": {}, + "sections": [ + { + "title": "Chapter 1", + "path": "report.pdf/Chapter 1", + "summary": long_leaf, + "chunk_count": 1, + "children": [], + }, + { + "title": "Chapter 2", + "path": "report.pdf/Chapter 2", + "summary": long_leaf, + "chunk_count": 1, + "children": [], + }, + ], + "resources": {"images": [], "tables": []}, + } + (file_dir / "doc_nav.json").write_text( + json.dumps(doc_nav, ensure_ascii=False), + encoding="utf-8", + ) + + results = enrich_doc_nav_summaries( + str(tmp_path), + source_file="report.pdf", + use_llm=False, + top_summary_use_llm=True, + ) + assert results["report.pdf"] == "LLM document overview" + assert captured["is_top"] is True + + saved = json.loads((file_dir / "doc_nav.json").read_text(encoding="utf-8")) + assert saved["top_summary"] == "LLM document overview" + assert load_nav_top_summary(str(file_dir), "report.pdf") == ( + "LLM document overview" + ) diff --git a/packages/shared-python/shared/models/schemas/job.py b/packages/shared-python/shared/models/schemas/job.py index 2a948f792..1ea775fde 100644 --- a/packages/shared-python/shared/models/schemas/job.py +++ b/packages/shared-python/shared/models/schemas/job.py @@ -38,6 +38,14 @@ class ParsingParams(BaseModel): "Increases parse time and API token cost." ), ) + top_summary_use_llm: bool = Field( + True, + description=( + "Use LLM for the document-level top_summary written to doc_nav. " + "Defaults to True because agentic document selection consumes this " + "field. Section-level summaries remain controlled by summary_use_llm." + ), + ) class JobCreateBase(BaseModel): diff --git a/packages/shared-python/shared/services/jobs/lifecycle/publication.py b/packages/shared-python/shared/services/jobs/lifecycle/publication.py index f29ed6acb..a8a3e7473 100644 --- a/packages/shared-python/shared/services/jobs/lifecycle/publication.py +++ b/packages/shared-python/shared/services/jobs/lifecycle/publication.py @@ -51,6 +51,7 @@ def publish_result( job_result_id: str, chunks: list[dict[str, Any]], section_summaries: dict[str, str] | None, + document_top_summary: str | None = None, ) -> JobPublicationOutcome: previous_document_scope = self._retrieval_publication.get_existing_document_scope( db, @@ -69,6 +70,7 @@ def publish_result( db, job_id=job_id, job_result_id=job_result_id, + top_summary=document_top_summary, ) cache_invalidation = self._build_cache_invalidation( diff --git a/packages/shared-python/shared/services/jobs/lifecycle/service.py b/packages/shared-python/shared/services/jobs/lifecycle/service.py index 3683fca30..c7953a005 100644 --- a/packages/shared-python/shared/services/jobs/lifecycle/service.py +++ b/packages/shared-python/shared/services/jobs/lifecycle/service.py @@ -52,6 +52,7 @@ def finalize_job_success( stored_count: int = 0, delivery_mode: str = "url", section_summaries: Optional[Dict[str, str]] = None, + document_top_summary: Optional[str] = None, ) -> Dict[str, Any]: """Finalize a successful job in a single atomic transaction. @@ -78,6 +79,7 @@ def finalize_job_success( stored_count=stored_count, delivery_mode=delivery_mode, section_summaries=section_summaries, + document_top_summary=document_top_summary, ), should_commit=lambda finalization: finalization.response.should_commit(), build_response=lambda finalization: finalization.response.to_dict(), diff --git a/packages/shared-python/shared/services/jobs/lifecycle/success_finalizer.py b/packages/shared-python/shared/services/jobs/lifecycle/success_finalizer.py index d05b41436..2edfc2b66 100644 --- a/packages/shared-python/shared/services/jobs/lifecycle/success_finalizer.py +++ b/packages/shared-python/shared/services/jobs/lifecycle/success_finalizer.py @@ -79,6 +79,7 @@ def finalize( stored_count: int, delivery_mode: str, section_summaries: dict[str, str] | None, + document_top_summary: str | None = None, ) -> JobSuccessFinalization: job_result = self._result_writer.upsert_job_result( db, @@ -95,6 +96,7 @@ def finalize( job_result_id=job_result.id, chunks=chunks, section_summaries=section_summaries, + document_top_summary=document_top_summary, ) transition_outcome = self._state_machine.mark_completed_outcome( diff --git a/packages/shared-python/shared/services/retrieval/graph/service.py b/packages/shared-python/shared/services/retrieval/graph/service.py index 35d05fec3..f5b9c51b6 100644 --- a/packages/shared-python/shared/services/retrieval/graph/service.py +++ b/packages/shared-python/shared/services/retrieval/graph/service.py @@ -75,6 +75,7 @@ def publish_document_graph( namespace: str, document_id: str, job_result_id: str, + top_summary: str | None = None, ) -> None: document = db.execute( select(Document).where(Document.document_id == document_id) @@ -103,7 +104,11 @@ def publish_document_graph( types_breakdown[chunk_type or 'text'] += 1 chunks_count = len(chunk_meta_rows) - top_summary = extract_document_top_summary(chunk_metadata_list) + resolved_top_summary = str(top_summary or "").strip() + if not resolved_top_summary: + # Backward compatible fallback for older chunks that still carry + # per-chunk document_top_summary copies. + resolved_top_summary = extract_document_top_summary(chunk_metadata_list) # ── Clean up old graph data for this document ── self.remove_document_graph( @@ -137,7 +142,7 @@ def publish_document_graph( 'top_entities': top_entities, 'chunks_count': chunks_count, 'types': dict(types_breakdown), - 'top_summary': top_summary, + 'top_summary': resolved_top_summary, }, ) ) diff --git a/packages/shared-python/shared/services/retrieval/publication_service.py b/packages/shared-python/shared/services/retrieval/publication_service.py index 6e70aa711..8911c3b7c 100644 --- a/packages/shared-python/shared/services/retrieval/publication_service.py +++ b/packages/shared-python/shared/services/retrieval/publication_service.py @@ -237,12 +237,18 @@ def publish_document_graph( *, job_id: str, job_result_id: str, + top_summary: str | None = None, ) -> None: job = db.execute(select(Job).where(Job.job_id == job_id)).scalar_one_or_none() if not job: raise RuntimeError(f"Job not found for graph publication: {job_id}") - self._publish_document_graph_for_job(db, job=job, job_result_id=job_result_id) + self._publish_document_graph_for_job( + db, + job=job, + job_result_id=job_result_id, + top_summary=top_summary, + ) def _publish_document_graph_for_job( self, @@ -250,6 +256,7 @@ def _publish_document_graph_for_job( *, job: Job, job_result_id: str, + top_summary: str | None = None, ) -> None: metadata = job.job_metadata or {} @@ -271,6 +278,7 @@ def _publish_document_graph_for_job( namespace=namespace, document_id=document_id, job_result_id=job_result_id, + top_summary=top_summary, ) def remove_document_graph( diff --git a/packages/shared-python/shared/services/storage/zip_chunk_schema.py b/packages/shared-python/shared/services/storage/zip_chunk_schema.py index 2cb36b432..9f43d3bd2 100644 --- a/packages/shared-python/shared/services/storage/zip_chunk_schema.py +++ b/packages/shared-python/shared/services/storage/zip_chunk_schema.py @@ -140,9 +140,6 @@ def _base_chunk_metadata( "summary": existing_metadata.get("summary") or chunk.get("summary", ""), "page_nums": existing_metadata.get("page_nums", []), } - document_top_summary = str(existing_metadata.get("document_top_summary") or "").strip() - if document_top_summary: - metadata["document_top_summary"] = document_top_summary return metadata diff --git a/packages/shared-python/shared/services/storage/zip_result_service.py b/packages/shared-python/shared/services/storage/zip_result_service.py index fe1f0c6dc..8d6b229a2 100644 --- a/packages/shared-python/shared/services/storage/zip_result_service.py +++ b/packages/shared-python/shared/services/storage/zip_result_service.py @@ -6,6 +6,8 @@ from __future__ import annotations +import json +import os from typing import Any from loguru import logger @@ -69,6 +71,7 @@ def generate_zip_package( statistics = self._schema.calculate_statistics(formatted_chunks) doc_nav, hierarchy = self._build_navigation_outputs( + add_dir=add_dir, formatted_chunks=formatted_chunks, source_file_name=source_file_name, ) @@ -120,13 +123,42 @@ def generate_zip_package( def _build_navigation_outputs( self, *, + add_dir: str, formatted_chunks: list[dict[str, Any]], source_file_name: str, ) -> tuple[dict[str, Any] | None, dict[str, Any]]: + """Prefer the already-enriched on-disk doc_nav; rebuild only as fallback.""" try: + existing = self._load_existing_doc_nav(add_dir) + if existing is not None: + hierarchy = self._schema.build_hierarchy_dict( + existing.get("sections", []) + ) + logger.info("Using enriched on-disk doc_nav.json for ZIP package") + return existing, hierarchy + doc_nav = self._schema.build_doc_nav(formatted_chunks, source_file_name) hierarchy = self._schema.build_hierarchy_dict(doc_nav.get("sections", [])) return doc_nav, hierarchy except Exception as exc: logger.warning(f"generate doc_nav.json fail {exc}") return None, {} + + @staticmethod + def _load_existing_doc_nav(add_dir: str) -> dict[str, Any] | None: + if not add_dir: + return None + path = os.path.join(add_dir, "doc_nav.json") + if not os.path.isfile(path): + return None + try: + with open(path, encoding="utf-8") as handle: + payload = json.load(handle) + except Exception as exc: + logger.warning(f"Failed to read existing doc_nav.json: {exc}") + return None + if not isinstance(payload, dict): + return None + if not isinstance(payload.get("sections"), list): + return None + return payload From c065bbfe592404299bf7de123735bc3ed4f7fce2 Mon Sep 17 00:00:00 2001 From: chengke <404835780@qq.com> Date: Tue, 21 Jul 2026 12:15:35 +0800 Subject: [PATCH 2/3] feat: enhance document parsing and image handling with new features - Updated the summary builder to improve document-level top summary generation, ensuring child contributions are deterministically summarized without LLM unless specified. - Enhanced inline asset handling to support table-embedded images, allowing for dynamic type assignment based on image references. - Introduced image size filtering across various document formats to skip undersized images, improving overall processing efficiency. - Refactored markdown image asset handling to include new properties for better management of image renaming and deferred tasks. - Improved table asset handling to include image references, ensuring that images embedded in tables are correctly processed and displayed. These updates collectively enhance the document parsing capabilities, streamline image handling, and improve the overall user experience in document navigation and summary generation. --- .../connect_builder/summary_builder.py | 44 +++- .../assets/image_size_filter.py | 46 ++++ .../document_parser/assets/inline_asset.py | 8 +- .../formats/docx/block_stream.py | 9 +- .../document_parser/formats/docx/parser.py | 3 - .../document_parser/formats/image/parser.py | 14 +- .../formats/markdown/deferred_summary.py | 9 +- .../formats/markdown/deferred_task.py | 1 + .../formats/markdown/image_asset.py | 33 ++- .../formats/markdown/parser.py | 23 +- .../formats/markdown/table_asset.py | 2 + .../formats/markdown/table_embedded_images.py | 173 +++++++++++++ .../orchestration/postprocess.py | 48 +++- .../test_image_size_filter_contract.py | 35 +++ .../test_table_embedded_images_contract.py | 234 ++++++++++++++++++ .../worker/tests/unit/test_summary_builder.py | 4 + .../chunks/dataframe_chunk_converter.py | 2 +- .../retrieval/agentic/evidence/renderer.py | 68 +++++ .../services/retrieval/hydration/connected.py | 2 +- .../retrieval/hydration/result_assembly.py | 87 +++++-- .../services/storage/zip_chunk_schema.py | 37 +-- 21 files changed, 805 insertions(+), 77 deletions(-) create mode 100644 apps/worker/app/services/document_parser/assets/image_size_filter.py create mode 100644 apps/worker/app/services/document_parser/formats/markdown/table_embedded_images.py create mode 100644 apps/worker/tests/contract/test_image_size_filter_contract.py create mode 100644 apps/worker/tests/contract/test_table_embedded_images_contract.py diff --git a/apps/worker/app/services/connect_builder/summary_builder.py b/apps/worker/app/services/connect_builder/summary_builder.py index bd1d327da..c02a57c4d 100644 --- a/apps/worker/app/services/connect_builder/summary_builder.py +++ b/apps/worker/app/services/connect_builder/summary_builder.py @@ -383,23 +383,45 @@ def _build_nav_top_summary( self_only_lookup: Optional[Dict[str, str]] = None, source_file_name: str = "", ) -> str: - """Build navigation-facing top summary from enriched doc_nav.json.""" - sections = doc_nav.get("sections", []) + """Build document-level top summary from already-enriched section nodes. + + Children are never re-summarized with LLM here — section LLM is controlled + only by ``enrich_doc_nav_summaries(use_llm=...)``. This path may optionally + LLM-aggregate the document overview from child contributions. + """ + sections = list(doc_nav.get("sections") or []) if not sections: return "" file_name = source_file_name or str(doc_nav.get("file_name") or "") - virtual_doc_node = { - "title": "Document Overview", - "children": sections, - } - return _recursive_summarize_nav( - virtual_doc_node, - use_llm=use_llm, + # Fill any missing child summaries deterministically without enabling LLM. + for section in sections: + if isinstance(section, dict): + _recursive_summarize_nav( + section, + use_llm=False, + self_only_lookup=self_only_lookup, + source_file_name=file_name, + ) + + child_titles = _child_title_list(sections, is_top_level=True) + child_rows = _child_contribution_rows(sections, is_top_level=True) + contrib_len = sum(len(contrib) for _, contrib in child_rows) + deterministic = _deterministic_section_summary( is_top_level=True, - self_only_lookup=self_only_lookup, - source_file_name=file_name, + self_only="", + child_titles=child_titles, + ) + if not use_llm or contrib_len <= SUMMARY_MAX_LEN: + return deterministic + + llm_result = _llm_summarize( + node_name="Document Overview", + self_only="", + child_rows=child_rows, + max_tokens=NAVIGATION_TOP_SUMMARY_MAX_TOKENS, ) + return llm_result or deterministic def _persist_doc_nav_top_summary( diff --git a/apps/worker/app/services/document_parser/assets/image_size_filter.py b/apps/worker/app/services/document_parser/assets/image_size_filter.py new file mode 100644 index 000000000..3c2c59dc7 --- /dev/null +++ b/apps/worker/app/services/document_parser/assets/image_size_filter.py @@ -0,0 +1,46 @@ +"""Shared minimum image size gate used across parsers. + +Discard before rename / VLM summary so undersized icons never enter the +asset pipeline. +""" + +from __future__ import annotations + +from pathlib import Path + +from loguru import logger + +from shared.core.constants.processing import ProcessingConstants + + +def is_below_img_min_size(byte_size: int) -> bool: + """Return True when ``byte_size`` is under ``IMG_MIN_SIZE`` (10KB).""" + return byte_size < ProcessingConstants.IMG_MIN_SIZE + + +def discard_undersized_image_file( + path: Path | str, + *, + label: str = "image", +) -> bool: + """Delete ``path`` when it exists and is undersized. + + Returns True when the file was discarded (caller should skip rename/LLM). + Returns False when the file is large enough to keep, or does not exist. + """ + image_path = Path(path) + if not image_path.exists(): + return False + + file_size = image_path.stat().st_size + if not is_below_img_min_size(file_size): + return False + + logger.debug( + f"Skipping {label} (too small: {file_size / 1024:.1f} KB): {image_path}" + ) + try: + image_path.unlink() + except OSError as exc: + logger.debug(f"Failed to remove undersized {label} {image_path}: {exc}") + return True diff --git a/apps/worker/app/services/document_parser/assets/inline_asset.py b/apps/worker/app/services/document_parser/assets/inline_asset.py index c39917d61..27348a19b 100644 --- a/apps/worker/app/services/document_parser/assets/inline_asset.py +++ b/apps/worker/app/services/document_parser/assets/inline_asset.py @@ -38,12 +38,18 @@ def build_table_asset_row( addtime: str, entities: str = "", asset_title: str = "", + image_refs: list[str] | None = None, ) -> ParsedRow: row_content = relative_path + # Multiline type channel carries table→image embeds + # (same pattern as PTXT\n[tables/...] for text rows). + type_value = "table" + if image_refs: + type_value = "\n".join(["table", *image_refs]) return ParsedRow( content=row_content, path=relative_path, - type="table", + type=type_value, keywords=keywords, summary=summary, know_id=know_id, diff --git a/apps/worker/app/services/document_parser/formats/docx/block_stream.py b/apps/worker/app/services/document_parser/formats/docx/block_stream.py index dafe28aba..3d8ddacf0 100644 --- a/apps/worker/app/services/document_parser/formats/docx/block_stream.py +++ b/apps/worker/app/services/document_parser/formats/docx/block_stream.py @@ -5,6 +5,7 @@ import zipfile from app.services.document_parser.formats.docx.toc import detect_doc_tocs, detect_sdt_toc +from app.services.document_parser.assets.image_size_filter import is_below_img_min_size from docx import Document from docx.oxml.table import CT_Tbl from docx.oxml.text.paragraph import CT_P @@ -134,6 +135,8 @@ def iter_block_items(doc_data): continue seen_rids.add(rid) data = docx.read("word/" + target) + if is_below_img_min_size(len(data)): + continue yield ( ele_num, None, @@ -234,9 +237,7 @@ def iter_block_items(doc_data): continue cell_seen_rids.add(rid) data = docx.read("word/" + target) - if ( - len(data) < 10 * 1024 - ): # Skip small images (<10KB, likely icons) + if is_below_img_min_size(len(data)): continue imgs_in_cell.append( { @@ -271,7 +272,7 @@ def iter_block_items(doc_data): f"Failed to convert VML cell image to PNG: {e}" ) continue - if len(png_data) < 10 * 1024: + if is_below_img_min_size(len(png_data)): continue orig_name = target.split("/")[-1] png_name = os.path.splitext(orig_name)[0] + ".png" diff --git a/apps/worker/app/services/document_parser/formats/docx/parser.py b/apps/worker/app/services/document_parser/formats/docx/parser.py index e195319f1..f978ab77d 100755 --- a/apps/worker/app/services/document_parser/formats/docx/parser.py +++ b/apps/worker/app/services/document_parser/formats/docx/parser.py @@ -584,9 +584,6 @@ def parse_docx( headings_stack[-1]["content"].append(text) elif label == "IMAGE": - if meta and meta.get("size", 0) < 10 * 1024: - continue - headings_stack = asset_accumulator.append_image( meta, headings_stack, diff --git a/apps/worker/app/services/document_parser/formats/image/parser.py b/apps/worker/app/services/document_parser/formats/image/parser.py index dd7fdd09d..52ff4df6d 100755 --- a/apps/worker/app/services/document_parser/formats/image/parser.py +++ b/apps/worker/app/services/document_parser/formats/image/parser.py @@ -87,7 +87,7 @@ def local_image_to_data_url(path, cut=True, min_size=None, max_size=None): if cut: file_size = path.stat().st_size # Bytes. - if file_size < min_size: # Smaller than 10 KB. + if file_size < min_size: logger.debug(f"Skipping {path} (too small: {file_size / 1024:.1f} KB)") return None if file_size >= max_size: # Larger than 5 MB. @@ -232,15 +232,13 @@ def parse_image( img_obj = Image.open(io.BytesIO(img_bytes)) img_obj.save(img_path) - # Early exit: skip images smaller than 10KB + # Early exit: skip images smaller than IMG_MIN_SIZE before VLM work. + from app.services.document_parser.assets.image_size_filter import ( + discard_undersized_image_file, + ) from shared.core.constants import ProcessingConstants - saved_size = os.path.getsize(img_path) - if saved_size < ProcessingConstants.IMG_MIN_SIZE: - logger.debug( - f"Skipping image {filename} (too small: {saved_size / 1024:.1f} KB)" - ) - os.remove(img_path) + if discard_undersized_image_file(img_path, label=f"image {filename}"): return pd.DataFrame(columns=list(PARSER_ROW_COLUMNS)) # Extract image content diff --git a/apps/worker/app/services/document_parser/formats/markdown/deferred_summary.py b/apps/worker/app/services/document_parser/formats/markdown/deferred_summary.py index e5b5bf90a..06ff1ad06 100644 --- a/apps/worker/app/services/document_parser/formats/markdown/deferred_summary.py +++ b/apps/worker/app/services/document_parser/formats/markdown/deferred_summary.py @@ -268,6 +268,11 @@ def _apply_image_summary_result( row = rows[row_index] _apply_asset_result_preserving_index(row, result) + # Table-embedded images keep a stable filename so in + # tables/*.html stays valid after summary generation. + if not original_task.rename_file: + return + img_title = result.title if not img_title: return @@ -311,9 +316,7 @@ def _apply_table_summary_result( table_dir = original_task.table_dir old_table_name = original_task.table_name - # Keep the original table-N index (same pattern as image rename). Do not - # re-derive from a separate counter — the legacy ``table_count - 1`` path - # produced off-by-one filenames vs text-chunk refs. + # Keep the original table-N index (same pattern as image rename). table_num_match = re.match(r"table-(\d+)", str(old_table_name)) table_num = ( table_num_match.group(1) diff --git a/apps/worker/app/services/document_parser/formats/markdown/deferred_task.py b/apps/worker/app/services/document_parser/formats/markdown/deferred_task.py index f26df1683..c0467504e 100644 --- a/apps/worker/app/services/document_parser/formats/markdown/deferred_task.py +++ b/apps/worker/app/services/document_parser/formats/markdown/deferred_task.py @@ -11,6 +11,7 @@ class ImageDeferredSummaryTask: image_dir: str image_name: str image_suffix: str + rename_file: bool = True @dataclass(frozen=True) diff --git a/apps/worker/app/services/document_parser/formats/markdown/image_asset.py b/apps/worker/app/services/document_parser/formats/markdown/image_asset.py index ed5f51abf..52305689d 100644 --- a/apps/worker/app/services/document_parser/formats/markdown/image_asset.py +++ b/apps/worker/app/services/document_parser/formats/markdown/image_asset.py @@ -7,6 +7,9 @@ from app.services.document_parser.support.identifiers import gen_str_codes from app.services.document_parser.formats.image.parser import perceptual_hash +from app.services.document_parser.assets.image_size_filter import ( + discard_undersized_image_file, +) from app.services.document_parser.assets.inline_asset import build_image_asset_row from app.services.document_parser.formats.markdown.deferred_task import ( ImageDeferredSummaryTask, @@ -27,6 +30,8 @@ class MarkdownImageAsset: cache_entry: dict[str, str] | None deferred_task: MarkdownDeferredSummaryTask | None should_advance_image_count: bool + relative_path: str | None + discarded_undersized: bool = False @dataclass(frozen=True) @@ -42,6 +47,7 @@ class MarkdownImageAssetRequest: seen_images: dict[str, dict[str, str]] summary_image: bool row_index: int + rename_on_summary: bool = True def build_markdown_image_asset( @@ -56,6 +62,13 @@ def build_markdown_image_asset( logger.warning(f"Image file not found, skipping rename: {request.image_path}") return _empty_asset(should_advance_image_count=True) + # Gate before rename / perceptual hash / deferred VLM summary. + if discard_undersized_image_file(source_path, label="markdown image"): + return _empty_asset( + should_advance_image_count=False, + discarded_undersized=True, + ) + with open(source_path, "rb") as image_file: image_binary_hash = perceptual_hash(image_file.read()) @@ -105,6 +118,7 @@ def build_markdown_image_asset( image_dir=request.image_dir, image_name=request.image_name, image_suffix=image_suffix, + rename_file=request.rename_on_summary, ) return MarkdownImageAsset( @@ -114,12 +128,15 @@ def build_markdown_image_asset( cache_entry=cache_entry, deferred_task=deferred_task, should_advance_image_count=True, + relative_path=relative_image_path, ) def build_markdown_image_name(*, image_count: int, last_context: str) -> str: - image_name_context = path_handle(last_context[:10], mode="clean_single") - return f"image-{str(image_count)}-{image_name_context}" + image_name_context = path_handle(last_context.strip(), mode="clean_single") + if image_name_context: + return f"image-{image_count}-{image_name_context}" + return f"image-{image_count}" def resolve_workspace_image_path( @@ -164,9 +181,10 @@ def _build_duplicate_image_asset( cache_entry: dict[str, str], timestamp: str, ) -> MarkdownImageAsset: + relative_path = cache_entry["relative_img_path"] row_values = _build_image_row_values( content=cache_entry["img_content"], - relative_path=cache_entry["relative_img_path"], + relative_path=relative_path, summary=cache_entry["img_summary_field"], know_id=cache_entry["temp_uid"], timestamp=timestamp, @@ -183,6 +201,7 @@ def _build_duplicate_image_asset( cache_entry=None, deferred_task=None, should_advance_image_count=False, + relative_path=relative_path, ) @@ -211,7 +230,11 @@ def _build_image_row_values( return cast(ParserRowValues, image_row.to_list()) -def _empty_asset(*, should_advance_image_count: bool) -> MarkdownImageAsset: +def _empty_asset( + *, + should_advance_image_count: bool, + discarded_undersized: bool = False, +) -> MarkdownImageAsset: return MarkdownImageAsset( content_item=None, row_values=None, @@ -219,4 +242,6 @@ def _empty_asset(*, should_advance_image_count: bool) -> MarkdownImageAsset: cache_entry=None, deferred_task=None, should_advance_image_count=should_advance_image_count, + relative_path=None, + discarded_undersized=discarded_undersized, ) diff --git a/apps/worker/app/services/document_parser/formats/markdown/parser.py b/apps/worker/app/services/document_parser/formats/markdown/parser.py index 4f552cbb5..32b5094b1 100755 --- a/apps/worker/app/services/document_parser/formats/markdown/parser.py +++ b/apps/worker/app/services/document_parser/formats/markdown/parser.py @@ -19,6 +19,9 @@ MarkdownTableAssetRequest, build_markdown_table_asset, ) +from app.services.document_parser.formats.markdown.table_embedded_images import ( + extract_table_embedded_images, +) from app.services.document_parser.support.parser_rows import ParsedRow from app.services.document_parser.support.path_helpers import find_matches_parsing from app.services.document_parser.formats.html.parser import ( @@ -417,8 +420,6 @@ def parse_md( if image_asset.should_advance_image_count: parser_state.image_count += 1 - # TODO for large and dense tables, such as "Epstein flight logs", - # integrate tabula-py as an independent extraction path to solve VLM hallucinations and misplacement # b. handle lines containing tables tb_bool, form, _ = identify_tables(line) if tb_bool: @@ -447,14 +448,30 @@ def parse_md( else: continue # Unknown form, skip + embedded = extract_table_embedded_images( + table_html=tb_str, + parser_state=parser_state, + output_dir=output_dir, + image_dir=img_dir, + summary_image=bool(base_llm_paras["summary_image"]), + ) + for image_asset in embedded.image_assets: + if image_asset.row_values is not None: + parser_state.append_row(image_asset.row_values) + if image_asset.deferred_task is not None: + parser_state.schedule_deferred_task(image_asset.deferred_task) + for image_ref in embedded.image_refs: + parser_state.append_content_item(f"\n{image_ref}\n") + table_asset = build_markdown_table_asset( MarkdownTableAssetRequest( - table_html=tb_str, + table_html=embedded.rewritten_html, table_dir=tb_dir, table_count=parser_state.table_count, timestamp=parser_state.timestamp, summary_table=bool(base_llm_paras["summary_table"]), row_index=len(parser_state.rows), + image_refs=embedded.image_refs, ) ) parser_state.append_content_item(table_asset.content_item) diff --git a/apps/worker/app/services/document_parser/formats/markdown/table_asset.py b/apps/worker/app/services/document_parser/formats/markdown/table_asset.py index dc7808f38..355dc7ecf 100644 --- a/apps/worker/app/services/document_parser/formats/markdown/table_asset.py +++ b/apps/worker/app/services/document_parser/formats/markdown/table_asset.py @@ -34,6 +34,7 @@ class MarkdownTableAssetRequest: timestamp: str summary_table: bool row_index: int + image_refs: list[str] | None = None def build_markdown_table_asset( @@ -60,6 +61,7 @@ def build_markdown_table_asset( keywords="", know_id=gen_str_codes((request.table_html + str(request.table_count))), addtime=request.timestamp, + image_refs=request.image_refs or [], ) deferred_task = None diff --git a/apps/worker/app/services/document_parser/formats/markdown/table_embedded_images.py b/apps/worker/app/services/document_parser/formats/markdown/table_embedded_images.py new file mode 100644 index 000000000..ab82543d5 --- /dev/null +++ b/apps/worker/app/services/document_parser/formats/markdown/table_embedded_images.py @@ -0,0 +1,173 @@ +"""Extract and rewrite tags embedded inside MinerU HTML tables.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +from bs4 import BeautifulSoup, Tag +from loguru import logger + +from app.services.document_parser.formats.markdown.image_asset import ( + MarkdownImageAsset, + MarkdownImageAssetRequest, + build_markdown_image_asset, + build_markdown_image_name, +) +from app.services.document_parser.formats.markdown.parse_state import MarkdownParseState +from shared.utils.chunk_refs import build_chunk_ref + +_IMG_SRC_RE = re.compile( + r"""]*\bsrc\s*=\s*(?P["'])(?P[^"']+)(?P=quote)""", + re.IGNORECASE, +) +_IMG_TAG_WITH_SRC_RE = re.compile( + r"""]*\bsrc\s*=\s*(?P["'])(?P[^"']+)(?P=quote)[^>]*/?\s*>""", + re.IGNORECASE, +) + + +@dataclass(frozen=True) +class TableEmbeddedImagesResult: + rewritten_html: str + """HTML with rewritten to stable images/image-N-* paths.""" + + image_assets: list[MarkdownImageAsset] + """Newly created image assets that should be registered as rows.""" + + image_refs: list[str] + """Chunk refs ([images/...]) for text content_items and table type channel.""" + + +def extract_table_embedded_images( + *, + table_html: str, + parser_state: MarkdownParseState, + output_dir: str, + image_dir: str, + summary_image: bool, +) -> TableEmbeddedImagesResult: + """Pull assets out of a table, rename them, and rewrite HTML srcs. + + Duplicate perceptual hashes and repeated srcs reuse the first stable path + without creating an extra image row. Missing files leave the original src. + """ + srcs = _unique_img_srcs(table_html) + if not srcs: + return TableEmbeddedImagesResult( + rewritten_html=table_html, + image_assets=[], + image_refs=[], + ) + + naming_context = _first_cell_text(table_html) + rewritten_html = table_html + image_assets: list[MarkdownImageAsset] = [] + image_refs: list[str] = [] + src_to_relative: dict[str, str] = {} + + for src in srcs: + if src in src_to_relative: + continue + + image_name = build_markdown_image_name( + image_count=parser_state.image_count, + last_context=naming_context, + ) + image_asset = build_markdown_image_asset( + MarkdownImageAssetRequest( + output_dir=output_dir, + image_dir=image_dir, + image_path=src, + image_name=image_name, + image_count=parser_state.image_count, + last_context=naming_context, + image_summary=naming_context or None, + timestamp=parser_state.timestamp, + seen_images=parser_state.seen_images, + summary_image=summary_image, + row_index=len(parser_state.rows) + len(image_assets), + rename_on_summary=False, + ) + ) + if image_asset.discarded_undersized: + rewritten_html = _strip_img_tags(rewritten_html, src) + continue + if image_asset.relative_path is None: + logger.warning( + f"Table-embedded image not found, leaving original src: {src}" + ) + continue + + src_to_relative[src] = image_asset.relative_path + image_refs.append(build_chunk_ref(image_asset.relative_path)) + + if image_asset.should_advance_image_count: + image_assets.append(image_asset) + # Advance immediately so the next distinct src gets a new image-N. + parser_state.image_count += 1 + if ( + image_asset.cache_key is not None + and image_asset.cache_entry is not None + ): + parser_state.seen_images[image_asset.cache_key] = ( + image_asset.cache_entry + ) + + for old_src, new_relative in src_to_relative.items(): + rewritten_html = _rewrite_img_src(rewritten_html, old_src, new_relative) + + return TableEmbeddedImagesResult( + rewritten_html=rewritten_html, + image_assets=image_assets, + image_refs=image_refs, + ) + + +def _unique_img_srcs(table_html: str) -> list[str]: + seen: set[str] = set() + ordered: list[str] = [] + for match in _IMG_SRC_RE.finditer(table_html): + src = match.group("src").strip() + if not src or src in seen: + continue + seen.add(src) + ordered.append(src) + return ordered + + +def _first_cell_text(table_html: str) -> str: + soup = BeautifulSoup(table_html, "html.parser") + table = soup.find("table") + if not isinstance(table, Tag): + return "" + for cell in table.find_all(["td", "th"]): + if not isinstance(cell, Tag): + continue + text = cell.get_text(strip=True) + if text: + return text + return "" + + +def _rewrite_img_src(html: str, old_src: str, new_src: str) -> str: + pattern = re.compile( + r"""(]*\bsrc\s*=\s*)(["'])""" + + re.escape(old_src) + + r"""\2""", + re.IGNORECASE, + ) + + def _replace(match: re.Match[str]) -> str: + return f"{match.group(1)}{match.group(2)}{new_src}{match.group(2)}" + + return pattern.sub(_replace, html) + + +def _strip_img_tags(html: str, src: str) -> str: + def _drop(match: re.Match[str]) -> str: + if match.group("src") == src: + return "" + return match.group(0) + + return _IMG_TAG_WITH_SRC_RE.sub(_drop, html) diff --git a/apps/worker/app/services/document_parser/orchestration/postprocess.py b/apps/worker/app/services/document_parser/orchestration/postprocess.py index 0fbb9f5c8..28f30a966 100644 --- a/apps/worker/app/services/document_parser/orchestration/postprocess.py +++ b/apps/worker/app/services/document_parser/orchestration/postprocess.py @@ -11,6 +11,15 @@ from app.services.document_parser.support.stage_profiler import stage_timer from loguru import logger +_IMG_SRC_BASENAME_RE = re.compile( + r"""]*\bsrc\s*=\s*["']([^"']+)["']""", + re.IGNORECASE, +) +_HASH_IMAGE_RE = re.compile( + r"^[a-f0-9]{64}\.(?:jpg|jpeg|png|gif|webp)$", + re.IGNORECASE, +) + def apply_parse_postprocess( output_dir: str, @@ -39,19 +48,23 @@ def apply_parse_postprocess( def cleanup_unreferenced_images(output_dir: str) -> int: - """Remove UUID-named images that are not referenced by final parsed output.""" + """Remove hash-named MinerU images that are not referenced by table HTML. + + Images extracted as ``image-N-*`` are never matched by the hash pattern and + are always kept. Hash-named files still referenced from ``tables/*.html`` + (e.g. failed extraction) are also preserved. + """ image_dir = os.path.join(output_dir, "images") if not os.path.isdir(image_dir): return 0 - uuid_pattern = re.compile( - r"^[a-f0-9]{64}\.(?:jpg|jpeg|png|gif|webp)$", - re.IGNORECASE, - ) + protected_basenames = _collect_table_img_basenames(output_dir) removed_count = 0 for filename in os.listdir(image_dir): - if not uuid_pattern.match(filename): + if not _HASH_IMAGE_RE.match(filename): + continue + if filename in protected_basenames: continue file_path = os.path.join(image_dir, filename) @@ -68,3 +81,26 @@ def cleanup_unreferenced_images(output_dir: str) -> int: ) return removed_count + + +def _collect_table_img_basenames(output_dir: str) -> set[str]: + tables_dir = os.path.join(output_dir, "tables") + if not os.path.isdir(tables_dir): + return set() + + basenames: set[str] = set() + for filename in os.listdir(tables_dir): + if not filename.endswith(".html"): + continue + table_path = os.path.join(tables_dir, filename) + try: + with open(table_path, encoding="utf-8") as table_file: + html = table_file.read() + except OSError as exc: + logger.warning(f"Failed to read table HTML {table_path}: {exc}") + continue + for match in _IMG_SRC_BASENAME_RE.finditer(html): + src = match.group(1).strip() + if src: + basenames.add(os.path.basename(src)) + return basenames diff --git a/apps/worker/tests/contract/test_image_size_filter_contract.py b/apps/worker/tests/contract/test_image_size_filter_contract.py new file mode 100644 index 000000000..1b04bdb22 --- /dev/null +++ b/apps/worker/tests/contract/test_image_size_filter_contract.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +import os +from pathlib import Path + +os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test") +os.environ.setdefault("TMP_PATH", "/tmp/knowhere-test") +os.environ.setdefault("S3_BUCKET_NAME", "test-uploads") +os.environ.setdefault("S3_ACCESS_KEY_ID", "test") +os.environ.setdefault("S3_SECRET_ACCESS_KEY", "test") +os.environ.setdefault("S3_TEMP_PATH", "/tmp") + +from app.services.document_parser.assets.image_size_filter import ( # noqa: E402 + discard_undersized_image_file, + is_below_img_min_size, +) +from shared.core.constants.processing import ProcessingConstants # noqa: E402 + + +def test_is_below_img_min_size_boundary() -> None: + assert is_below_img_min_size(ProcessingConstants.IMG_MIN_SIZE - 1) + assert not is_below_img_min_size(ProcessingConstants.IMG_MIN_SIZE) + assert not is_below_img_min_size(ProcessingConstants.IMG_MIN_SIZE + 1) + + +def test_discard_undersized_image_file(tmp_path: Path) -> None: + small = tmp_path / "small.jpg" + small.write_bytes(b"x" * (ProcessingConstants.IMG_MIN_SIZE - 1)) + assert discard_undersized_image_file(small, label="unit") is True + assert not small.exists() + + large = tmp_path / "large.jpg" + large.write_bytes(b"y" * ProcessingConstants.IMG_MIN_SIZE) + assert discard_undersized_image_file(large, label="unit") is False + assert large.exists() diff --git a/apps/worker/tests/contract/test_table_embedded_images_contract.py b/apps/worker/tests/contract/test_table_embedded_images_contract.py new file mode 100644 index 000000000..0da70fad8 --- /dev/null +++ b/apps/worker/tests/contract/test_table_embedded_images_contract.py @@ -0,0 +1,234 @@ +from __future__ import annotations + +import os +from pathlib import Path + +os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test") +os.environ.setdefault("TMP_PATH", "/tmp/knowhere-test") +os.environ.setdefault("S3_BUCKET_NAME", "test-uploads") +os.environ.setdefault("S3_ACCESS_KEY_ID", "test") +os.environ.setdefault("S3_SECRET_ACCESS_KEY", "test") +os.environ.setdefault("S3_TEMP_PATH", "/tmp") + +from PIL import Image # noqa: E402 + +from app.services.document_parser.formats.markdown.parse_state import ( # noqa: E402 + MarkdownParseState, +) +from app.services.document_parser.formats.markdown.parser import ( # noqa: E402 + update_df_list, +) +from app.services.document_parser.formats.markdown.table_asset import ( # noqa: E402 + MarkdownTableAssetRequest, + build_markdown_table_asset, +) +from app.services.document_parser.formats.markdown.table_embedded_images import ( # noqa: E402 + extract_table_embedded_images, +) +from app.services.document_parser.orchestration.postprocess import ( # noqa: E402 + cleanup_unreferenced_images, +) +from shared.core.constants.processing import ProcessingConstants # noqa: E402 +from shared.services.chunks.dataframe_chunk_converter import ( # noqa: E402 + dataframe_to_chunks, +) +from shared.services.retrieval.agentic.evidence.renderer import ( # noqa: E402 + render_table_chunk_lines, +) + + +def _write_jpeg( + path: Path, + *, + size: tuple[int, int] = (640, 640), + min_bytes: int | None = None, +) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + # Solid colors compress too well; patterned noise keeps file size realistic. + width, height = size + pixels = bytearray(width * height * 3) + for index in range(len(pixels)): + pixels[index] = (index * 37) % 256 + Image.frombytes("RGB", size, bytes(pixels)).save(path, format="JPEG", quality=95) + if min_bytes is not None: + current_size = path.stat().st_size + if current_size < min_bytes: + with path.open("ab") as handle: + handle.write(b"\x00" * (min_bytes - current_size)) + + +def _make_parser_state() -> MarkdownParseState: + return MarkdownParseState( + relative_root="doc.pdf", + split_char="/", + llm_parameters={ + "summary_image": False, + "summary_table": False, + "summary_txt": False, + "stopwords": set(), + }, + timestamp="2026-01-01 00:00:00", + row_updater=update_df_list, + ) + + +def test_table_embedded_images_are_extracted_rewritten_and_linked( + tmp_path: Path, +) -> None: + output_dir = tmp_path / "doc" + images_dir = output_dir / "images" + tables_dir = output_dir / "tables" + images_dir.mkdir(parents=True) + tables_dir.mkdir(parents=True) + + hash_name = "a" * 64 + ".jpg" + image_path = images_dir / hash_name + _write_jpeg(image_path, min_bytes=ProcessingConstants.IMG_MIN_SIZE) + assert image_path.stat().st_size >= ProcessingConstants.IMG_MIN_SIZE + + table_html = ( + "" + f'' + "
流程名称评审流程图
说明审批节点
" + ) + + parser_state = _make_parser_state() + + embedded = extract_table_embedded_images( + table_html=table_html, + parser_state=parser_state, + output_dir=str(output_dir), + image_dir=str(images_dir), + summary_image=False, + ) + + assert len(embedded.image_assets) == 1 + assert len(embedded.image_refs) == 1 + assert hash_name not in embedded.rewritten_html + assert " None: + output_dir = tmp_path / "doc" + images_dir = output_dir / "images" + images_dir.mkdir(parents=True) + + hash_name = "c" * 64 + ".jpg" + image_path = images_dir / hash_name + _write_jpeg(image_path, size=(16, 16)) + assert image_path.stat().st_size < ProcessingConstants.IMG_MIN_SIZE + + table_html = ( + "" + f'
Logo
' + ) + + embedded = extract_table_embedded_images( + table_html=table_html, + parser_state=_make_parser_state(), + output_dir=str(output_dir), + image_dir=str(images_dir), + summary_image=False, + ) + + assert embedded.image_assets == [] + assert embedded.image_refs == [] + assert " str: captured["is_top"] = kwargs.get("node_name") == "Document Overview" captured["max_tokens"] = kwargs.get("max_tokens") + captured["calls"] = int(captured.get("calls") or 0) + 1 return "LLM document overview" monkeypatch.setattr( @@ -293,10 +294,13 @@ def _fake_llm(**kwargs: Any) -> str: top_summary_use_llm=True, ) assert results["report.pdf"] == "LLM document overview" + assert captured["calls"] == 1 assert captured["is_top"] is True saved = json.loads((file_dir / "doc_nav.json").read_text(encoding="utf-8")) assert saved["top_summary"] == "LLM document overview" + # Section leaves keep original summaries; top LLM must not rewrite them. + assert saved["sections"][0]["summary"] == long_leaf assert load_nav_top_summary(str(file_dir), "report.pdf") == ( "LLM document overview" ) diff --git a/packages/shared-python/shared/services/chunks/dataframe_chunk_converter.py b/packages/shared-python/shared/services/chunks/dataframe_chunk_converter.py index 513f8b78b..e5ad3931f 100644 --- a/packages/shared-python/shared/services/chunks/dataframe_chunk_converter.py +++ b/packages/shared-python/shared/services/chunks/dataframe_chunk_converter.py @@ -372,7 +372,7 @@ def dataframe_to_chunks(df: _ParserDataFrame | None) -> list[Dict[str, JsonValue for chunk in chunks: metadata = chunk["metadata"] relationship_refs = metadata.pop("_relationship_refs", []) - if chunk["type"] not in {"text", "page"}: + if chunk["type"] not in {"text", "page", "table"}: continue embed_connections = convert_refs_to_embed_connections( relationship_refs, resource_target_map diff --git a/packages/shared-python/shared/services/retrieval/agentic/evidence/renderer.py b/packages/shared-python/shared/services/retrieval/agentic/evidence/renderer.py index db9338027..3b8bd9c73 100644 --- a/packages/shared-python/shared/services/retrieval/agentic/evidence/renderer.py +++ b/packages/shared-python/shared/services/retrieval/agentic/evidence/renderer.py @@ -183,6 +183,9 @@ def render_leaf_chunks( table_lines = render_table_chunk_lines( target, display_ref=display_ref, + chunk_by_id=chunk_by_id, + asset_lookup=asset_lookup, + rendered_ids=rendered_ids, ) content = content.replace(ref_str, "\n" + "\n".join(table_lines) + "\n") elif target_type == "image": @@ -227,6 +230,9 @@ def render_leaf_chunks( for line in render_table_chunk_lines( chunk, display_ref=display_ref, + chunk_by_id=chunk_by_id, + asset_lookup=asset_lookup, + rendered_ids=rendered_ids, ): if line.strip(): parts.append(f"{indent}┈ {line}") @@ -262,6 +268,9 @@ def render_table_chunk_lines( chunk: dict[str, Any], *, display_ref: str, + chunk_by_id: dict[str, dict[str, Any]] | None = None, + asset_lookup: dict[str, AssetLookupValue] | None = None, + rendered_ids: set[str] | None = None, ) -> list[str]: header = f"[Table: {display_ref}]" if display_ref else "[Table]" lines = [header] @@ -292,6 +301,65 @@ def render_table_chunk_lines( if caption: lines.append("Caption:") lines.append(str(caption).strip()) + + lines.extend( + _render_table_embedded_image_lines( + chunk, + chunk_by_id=chunk_by_id or {}, + asset_lookup=asset_lookup, + rendered_ids=rendered_ids, + ) + ) + return lines + + +def _render_table_embedded_image_lines( + table_chunk: dict[str, Any], + *, + chunk_by_id: dict[str, dict[str, Any]], + asset_lookup: dict[str, AssetLookupValue] | None, + rendered_ids: set[str] | None, +) -> list[str]: + metadata = table_chunk.get("chunk_metadata") or table_chunk.get("metadata") or {} + if not isinstance(metadata, dict): + return [] + + lines: list[str] = [] + for connection in metadata.get("connect_to") or []: + if not isinstance(connection, dict): + continue + if str(connection.get("relation") or "").strip() != "embeds": + continue + target_id = str(connection.get("target") or "").strip() + if not target_id: + continue + target = chunk_by_id.get(target_id) + if not target: + continue + target_type = ( + target.get("chunk_type") or target.get("type") or "" + ).strip().lower() + if target_type != "image": + continue + if rendered_ids is not None: + rendered_ids.add(target_id) + + file_path = target.get("file_path") or "" + asset_url = _lookup_asset_url(asset_lookup, target_id) + display_ref = asset_url or file_path + image_description = str(target.get("content") or "").strip() + ref_str = str(connection.get("ref") or "").strip() + if ref_str and ref_str in image_description: + image_description = image_description.replace(ref_str, "").strip() + + if display_ref: + lines.append(f"[Image: {display_ref}]") + elif image_description: + lines.append("[Image description]") + if image_description: + lines.extend( + line for line in image_description.split("\n") if line.strip() + ) return lines diff --git a/packages/shared-python/shared/services/retrieval/hydration/connected.py b/packages/shared-python/shared/services/retrieval/hydration/connected.py index 2267b91c8..fcbc647bf 100644 --- a/packages/shared-python/shared/services/retrieval/hydration/connected.py +++ b/packages/shared-python/shared/services/retrieval/hydration/connected.py @@ -31,7 +31,7 @@ async def hydrate_connected_target_rows( } target_ids_by_revision: dict[tuple[str, str], set[str]] = {} for row in rows: - if normalize_chunk_type(row.get('chunk_type')) not in ('text', 'page'): + if normalize_chunk_type(row.get('chunk_type')) not in ('text', 'page', 'table'): continue document_id = str(row.get('document_id') or '').strip() job_result_id = str(row.get('job_result_id') or '').strip() diff --git a/packages/shared-python/shared/services/retrieval/hydration/result_assembly.py b/packages/shared-python/shared/services/retrieval/hydration/result_assembly.py index 8c9a1b93e..96534bce3 100644 --- a/packages/shared-python/shared/services/retrieval/hydration/result_assembly.py +++ b/packages/shared-python/shared/services/retrieval/hydration/result_assembly.py @@ -60,28 +60,14 @@ async def assemble_retrieval_results( assembled_row['content'] = _page_summary(row) assembled_row['content_source'] = 'summary' elif chunk_type == 'table': - assembled_row['content'] = _table_summary_content(row) + assembled_row['content'] = _compose_table_content(row, rows_by_chunk_id) assembled_row['content_source'] = 'summary' elif chunk_type == 'text': - connected_targets: list[tuple[int, str]] = [] - for target_id in iter_connected_target_ids(row): - target_row = rows_by_chunk_id.get(target_id) - if not target_row: - continue - if normalize_chunk_type(target_row.get('chunk_type')) != 'table': - continue - target_content = _table_summary_content(target_row) - if target_content: - sort_key = int(target_row.get('sort_order', 0) or 0) - connected_targets.append((sort_key, target_content)) - connected_targets.sort(key=lambda item: item[0]) - related_parts = [content for _, content in connected_targets] - - # TODO: Dedicated Large Table Agent - # For the Notebook/Agent environment, consider introducing a dedicated - # "Large Table Agent" that can fetch and query oversized tables via URL. + related_parts = _connected_media_parts(row, rows_by_chunk_id) if base_content and related_parts: assembled_row['content'] = '\n\n'.join([base_content, *related_parts]) + elif related_parts: + assembled_row['content'] = '\n\n'.join(related_parts) else: assembled_row['content'] = base_content assembled_row['content_source'] = 'content' @@ -100,6 +86,71 @@ def _page_summary(row: dict[str, Any]) -> str: return str(metadata.get('summary') or '').strip() +def _compose_table_content( + row: dict[str, Any], + rows_by_chunk_id: dict[str, dict[str, Any]], +) -> str: + parts = [_table_summary_content(row)] + parts.extend(_connected_image_parts(row, rows_by_chunk_id)) + return '\n\n'.join(part for part in parts if part) + + +def _connected_media_parts( + row: dict[str, Any], + rows_by_chunk_id: dict[str, dict[str, Any]], +) -> list[str]: + connected_targets: list[tuple[int, str]] = [] + for target_id in iter_connected_target_ids(row): + target_row = rows_by_chunk_id.get(target_id) + if not target_row: + continue + target_type = normalize_chunk_type(target_row.get('chunk_type')) + if target_type == 'table': + target_content = _compose_table_content(target_row, rows_by_chunk_id) + elif target_type == 'image': + target_content = _image_display_content(target_row) + else: + continue + if target_content: + sort_key = int(target_row.get('sort_order', 0) or 0) + connected_targets.append((sort_key, target_content)) + connected_targets.sort(key=lambda item: item[0]) + return [content for _, content in connected_targets] + + +def _connected_image_parts( + row: dict[str, Any], + rows_by_chunk_id: dict[str, dict[str, Any]], +) -> list[str]: + parts: list[str] = [] + for target_id in iter_connected_target_ids(row): + target_row = rows_by_chunk_id.get(target_id) + if not target_row: + continue + if normalize_chunk_type(target_row.get('chunk_type')) != 'image': + continue + content = _image_display_content(target_row) + if content: + parts.append(content) + return parts + + +def _image_display_content(row: dict[str, Any]) -> str: + display_ref = ( + str(row.get('asset_url') or '').strip() + or str(row.get('file_path') or '').strip() + ) + description = str(row.get('content') or '').strip() + lines: list[str] = [] + if display_ref: + lines.append(f'[Image: {display_ref}]') + elif description: + lines.append('[Image description]') + if description: + lines.extend(line for line in description.split('\n') if line.strip()) + return '\n'.join(lines) + + def _table_summary_content(row: dict[str, Any]) -> str: metadata = row.get('chunk_metadata') or row.get('metadata') or {} if not isinstance(metadata, dict): diff --git a/packages/shared-python/shared/services/storage/zip_chunk_schema.py b/packages/shared-python/shared/services/storage/zip_chunk_schema.py index 9f43d3bd2..7dc0be09e 100644 --- a/packages/shared-python/shared/services/storage/zip_chunk_schema.py +++ b/packages/shared-python/shared/services/storage/zip_chunk_schema.py @@ -78,13 +78,19 @@ def format_chunks( if chunk_type == "text": metadata.update( - _format_text_metadata( - chunk=chunk, - chunk_type_str=chunk_type_str, - content=str(content), - existing_metadata=existing_metadata, - resource_target_map=resource_target_map, - ) + { + "tokens": existing_metadata.get("tokens") + or chunk.get("tokens", 0), + "keywords": existing_metadata.get("keywords") + or chunk.get("keywords", []), + "connect_to": _build_embed_connect_to( + chunk=chunk, + chunk_type_str=chunk_type_str, + content=str(content), + existing_metadata=existing_metadata, + resource_target_map=resource_target_map, + ), + } ) elif chunk_type == "image": if image_info: @@ -105,6 +111,13 @@ def format_chunks( "keywords", [] ) metadata["tokens"] = [] + metadata["connect_to"] = _build_embed_connect_to( + chunk=chunk, + chunk_type_str=chunk_type_str, + content=str(content), + existing_metadata=existing_metadata, + resource_target_map=resource_target_map, + ) elif chunk_type == "page": metadata["keywords"] = existing_metadata.get("keywords") or [] metadata["connect_to"] = existing_metadata.get("connect_to") or [] @@ -143,14 +156,14 @@ def _base_chunk_metadata( return metadata -def _format_text_metadata( +def _build_embed_connect_to( *, chunk: dict[str, Any], chunk_type_str: Any, content: str, existing_metadata: dict[str, Any], resource_target_map: dict[str, str], -) -> dict[str, Any]: +) -> list[Any]: relationship_refs = parse_relationship_refs( chunk.get("type_raw") or chunk_type_str, content, @@ -165,11 +178,7 @@ def _format_text_metadata( or chunk.get("connectto"), resource_target_map, ) - return { - "tokens": existing_metadata.get("tokens") or chunk.get("tokens", 0), - "keywords": existing_metadata.get("keywords") or chunk.get("keywords", []), - "connect_to": merge_connections(embed_connections, related_connections), - } + return merge_connections(embed_connections, related_connections) def _resolve_table_file_path( From 30742c0cd888afe83adaaeb76e6317a7e1a2df32 Mon Sep 17 00:00:00 2001 From: chengke <404835780@qq.com> Date: Tue, 21 Jul 2026 12:17:22 +0800 Subject: [PATCH 3/3] fix: accept model kwarg in summary_builder prompt unit test Co-authored-by: Cursor --- apps/worker/tests/unit/test_summary_builder.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/worker/tests/unit/test_summary_builder.py b/apps/worker/tests/unit/test_summary_builder.py index eb7b566df..a0eaf2c73 100644 --- a/apps/worker/tests/unit/test_summary_builder.py +++ b/apps/worker/tests/unit/test_summary_builder.py @@ -202,7 +202,7 @@ def test_file_summary_prompt_contains_scope_blocks( ) -> None: captured: Dict[str, Any] = {} - def _fake_client() -> Any: + def _fake_client(**_kwargs: Any) -> Any: class _C: def chat_completion(self, **kwargs: Any) -> str: captured["messages"] = kwargs.get("messages")