mirror of
https://github.com/HKUDS/RAG-Anything.git
synced 2026-07-08 15:05:26 +08:00
725 lines
27 KiB
Python
725 lines
27 KiB
Python
"""
|
|
Document processing functionality for RAGAnything
|
|
|
|
Contains methods for parsing documents and processing multimodal content
|
|
"""
|
|
|
|
import os
|
|
import time
|
|
import hashlib
|
|
import json
|
|
from typing import Dict, List, Any
|
|
from pathlib import Path
|
|
from raganything.parser import MineruParser, DoclingParser
|
|
from raganything.utils import (
|
|
separate_content,
|
|
insert_text_content,
|
|
get_processor_for_type,
|
|
)
|
|
|
|
|
|
class ProcessorMixin:
|
|
"""ProcessorMixin class containing document processing functionality for RAGAnything"""
|
|
|
|
def _generate_cache_key(
|
|
self, file_path: Path, parse_method: str = None, **kwargs
|
|
) -> str:
|
|
"""
|
|
Generate cache key based on file path and parsing configuration
|
|
|
|
Args:
|
|
file_path: Path to the file
|
|
parse_method: Parse method used
|
|
**kwargs: Additional parser parameters
|
|
|
|
Returns:
|
|
str: Cache key for the file and configuration
|
|
"""
|
|
|
|
# Get file modification time
|
|
mtime = file_path.stat().st_mtime
|
|
|
|
# Create configuration dict for cache key
|
|
config_dict = {
|
|
"file_path": str(file_path.absolute()),
|
|
"mtime": mtime,
|
|
"parser": self.config.parser,
|
|
"parse_method": parse_method or self.config.parse_method,
|
|
}
|
|
|
|
# Add relevant kwargs to config
|
|
relevant_kwargs = {
|
|
k: v
|
|
for k, v in kwargs.items()
|
|
if k
|
|
in [
|
|
"lang",
|
|
"device",
|
|
"start_page",
|
|
"end_page",
|
|
"formula",
|
|
"table",
|
|
"backend",
|
|
"source",
|
|
]
|
|
}
|
|
config_dict.update(relevant_kwargs)
|
|
|
|
# Generate hash from config
|
|
config_str = json.dumps(config_dict, sort_keys=True)
|
|
cache_key = hashlib.md5(config_str.encode()).hexdigest()
|
|
|
|
return cache_key
|
|
|
|
def _generate_content_based_doc_id(self, content_list: List[Dict[str, Any]]) -> str:
|
|
"""
|
|
Generate doc_id based on document content
|
|
|
|
Args:
|
|
content_list: Parsed content list
|
|
|
|
Returns:
|
|
str: Content-based document ID with doc- prefix
|
|
"""
|
|
from lightrag.utils import compute_mdhash_id
|
|
|
|
# Extract key content for ID generation
|
|
content_hash_data = []
|
|
|
|
for item in content_list:
|
|
if isinstance(item, dict):
|
|
# For text content, use the text
|
|
if item.get("type") == "text" and item.get("text"):
|
|
content_hash_data.append(item["text"].strip())
|
|
# For other content types, use key identifiers
|
|
elif item.get("type") == "image" and item.get("img_path"):
|
|
content_hash_data.append(f"image:{item['img_path']}")
|
|
elif item.get("type") == "table" and item.get("table_body"):
|
|
content_hash_data.append(f"table:{item['table_body']}")
|
|
elif item.get("type") == "equation" and item.get("text"):
|
|
content_hash_data.append(f"equation:{item['text']}")
|
|
else:
|
|
# For other types, use string representation
|
|
content_hash_data.append(str(item))
|
|
|
|
# Create a content signature
|
|
content_signature = "\n".join(content_hash_data)
|
|
|
|
# Generate doc_id from content
|
|
doc_id = compute_mdhash_id(content_signature, prefix="doc-")
|
|
|
|
return doc_id
|
|
|
|
async def _get_cached_result(
|
|
self, cache_key: str, file_path: Path, parse_method: str = None, **kwargs
|
|
) -> tuple[List[Dict[str, Any]], str] | None:
|
|
"""
|
|
Get cached parsing result if available and valid
|
|
|
|
Args:
|
|
cache_key: Cache key to look up
|
|
file_path: Path to the file for mtime check
|
|
parse_method: Parse method used
|
|
**kwargs: Additional parser parameters
|
|
|
|
Returns:
|
|
tuple[List[Dict[str, Any]], str] | None: (content_list, doc_id) or None if not found/invalid
|
|
"""
|
|
if not hasattr(self, "parse_cache") or self.parse_cache is None:
|
|
return None
|
|
|
|
try:
|
|
cached_data = await self.parse_cache.get_by_id(cache_key)
|
|
if not cached_data:
|
|
return None
|
|
|
|
# Check file modification time
|
|
current_mtime = file_path.stat().st_mtime
|
|
cached_mtime = cached_data.get("mtime", 0)
|
|
|
|
if current_mtime != cached_mtime:
|
|
self.logger.debug(f"Cache invalid - file modified: {cache_key}")
|
|
return None
|
|
|
|
# Check parsing configuration
|
|
cached_config = cached_data.get("parse_config", {})
|
|
current_config = {
|
|
"parser": self.config.parser,
|
|
"parse_method": parse_method or self.config.parse_method,
|
|
}
|
|
|
|
# Add relevant kwargs to current config
|
|
relevant_kwargs = {
|
|
k: v
|
|
for k, v in kwargs.items()
|
|
if k
|
|
in [
|
|
"lang",
|
|
"device",
|
|
"start_page",
|
|
"end_page",
|
|
"formula",
|
|
"table",
|
|
"backend",
|
|
"source",
|
|
]
|
|
}
|
|
current_config.update(relevant_kwargs)
|
|
|
|
if cached_config != current_config:
|
|
self.logger.debug(f"Cache invalid - config changed: {cache_key}")
|
|
return None
|
|
|
|
content_list = cached_data.get("content_list", [])
|
|
doc_id = cached_data.get("doc_id")
|
|
|
|
if content_list and doc_id:
|
|
self.logger.debug(
|
|
f"Found valid cached parsing result for key: {cache_key}"
|
|
)
|
|
return content_list, doc_id
|
|
else:
|
|
self.logger.debug(
|
|
f"Cache incomplete - missing content or doc_id: {cache_key}"
|
|
)
|
|
return None
|
|
|
|
except Exception as e:
|
|
self.logger.warning(f"Error accessing parse cache: {e}")
|
|
|
|
return None
|
|
|
|
async def _store_cached_result(
|
|
self,
|
|
cache_key: str,
|
|
content_list: List[Dict[str, Any]],
|
|
doc_id: str,
|
|
file_path: Path,
|
|
parse_method: str = None,
|
|
**kwargs,
|
|
) -> None:
|
|
"""
|
|
Store parsing result in cache
|
|
|
|
Args:
|
|
cache_key: Cache key to store under
|
|
content_list: Content list to cache
|
|
doc_id: Content-based document ID
|
|
file_path: Path to the file for mtime storage
|
|
parse_method: Parse method used
|
|
**kwargs: Additional parser parameters
|
|
"""
|
|
if not hasattr(self, "parse_cache") or self.parse_cache is None:
|
|
return
|
|
|
|
try:
|
|
# Get file modification time
|
|
file_mtime = file_path.stat().st_mtime
|
|
|
|
# Create parsing configuration
|
|
parse_config = {
|
|
"parser": self.config.parser,
|
|
"parse_method": parse_method or self.config.parse_method,
|
|
}
|
|
|
|
# Add relevant kwargs to config
|
|
relevant_kwargs = {
|
|
k: v
|
|
for k, v in kwargs.items()
|
|
if k
|
|
in [
|
|
"lang",
|
|
"device",
|
|
"start_page",
|
|
"end_page",
|
|
"formula",
|
|
"table",
|
|
"backend",
|
|
"source",
|
|
]
|
|
}
|
|
parse_config.update(relevant_kwargs)
|
|
|
|
cache_data = {
|
|
cache_key: {
|
|
"content_list": content_list,
|
|
"doc_id": doc_id,
|
|
"mtime": file_mtime,
|
|
"parse_config": parse_config,
|
|
"cached_at": time.time(),
|
|
"cache_version": "1.0",
|
|
}
|
|
}
|
|
await self.parse_cache.upsert(cache_data)
|
|
# Ensure data is persisted to disk
|
|
await self.parse_cache.index_done_callback()
|
|
self.logger.info(f"Stored parsing result in cache: {cache_key}")
|
|
except Exception as e:
|
|
self.logger.warning(f"Error storing to parse cache: {e}")
|
|
|
|
async def parse_document(
|
|
self,
|
|
file_path: str,
|
|
output_dir: str = None,
|
|
parse_method: str = None,
|
|
display_stats: bool = None,
|
|
**kwargs,
|
|
) -> tuple[List[Dict[str, Any]], str]:
|
|
"""
|
|
Parse document with caching support
|
|
|
|
Args:
|
|
file_path: Path to the file to parse
|
|
output_dir: Output directory (defaults to config.parser_output_dir)
|
|
parse_method: Parse method (defaults to config.parse_method)
|
|
display_stats: Whether to display content statistics (defaults to config.display_content_stats)
|
|
**kwargs: Additional parameters for parser (e.g., lang, device, start_page, end_page, formula, table, backend, source)
|
|
|
|
Returns:
|
|
tuple[List[Dict[str, Any]], str]: (content_list, doc_id)
|
|
"""
|
|
# Use config defaults if not provided
|
|
if output_dir is None:
|
|
output_dir = self.config.parser_output_dir
|
|
if parse_method is None:
|
|
parse_method = self.config.parse_method
|
|
if display_stats is None:
|
|
display_stats = self.config.display_content_stats
|
|
|
|
self.logger.info(f"Starting document parsing: {file_path}")
|
|
|
|
file_path = Path(file_path)
|
|
if not file_path.exists():
|
|
raise FileNotFoundError(f"File not found: {file_path}")
|
|
|
|
# Generate cache key based on file and configuration
|
|
cache_key = self._generate_cache_key(file_path, parse_method, **kwargs)
|
|
|
|
# Check cache first
|
|
cached_result = await self._get_cached_result(
|
|
cache_key, file_path, parse_method, **kwargs
|
|
)
|
|
if cached_result is not None:
|
|
content_list, doc_id = cached_result
|
|
self.logger.info(f"Using cached parsing result for: {file_path}")
|
|
if display_stats:
|
|
self.logger.info(
|
|
f"* Total blocks in cached content_list: {len(content_list)}"
|
|
)
|
|
return content_list, doc_id
|
|
|
|
# Choose appropriate parsing method based on file extension
|
|
ext = file_path.suffix.lower()
|
|
|
|
try:
|
|
doc_parser = (
|
|
DoclingParser() if self.config.parser == "docling" else MineruParser()
|
|
)
|
|
|
|
# Log parser and method information
|
|
self.logger.info(
|
|
f"Using {self.config.parser} parser with method: {parse_method}"
|
|
)
|
|
|
|
if ext in [".pdf"]:
|
|
self.logger.info("Detected PDF file, using parser for PDF...")
|
|
content_list = doc_parser.parse_pdf(
|
|
pdf_path=file_path,
|
|
output_dir=output_dir,
|
|
method=parse_method,
|
|
**kwargs,
|
|
)
|
|
elif ext in [
|
|
".jpg",
|
|
".jpeg",
|
|
".png",
|
|
".bmp",
|
|
".tiff",
|
|
".tif",
|
|
".gif",
|
|
".webp",
|
|
]:
|
|
self.logger.info("Detected image file, using parser for images...")
|
|
# Use the selected parser's image parsing capability
|
|
if hasattr(doc_parser, "parse_image"):
|
|
content_list = doc_parser.parse_image(
|
|
image_path=file_path, output_dir=output_dir, **kwargs
|
|
)
|
|
else:
|
|
# Fallback to MinerU for image parsing if current parser doesn't support it
|
|
self.logger.warning(
|
|
f"{self.config.parser} parser doesn't support image parsing, falling back to MinerU"
|
|
)
|
|
content_list = MineruParser().parse_image(
|
|
image_path=file_path, output_dir=output_dir, **kwargs
|
|
)
|
|
elif ext in [
|
|
".doc",
|
|
".docx",
|
|
".ppt",
|
|
".pptx",
|
|
".xls",
|
|
".xlsx",
|
|
".html",
|
|
".htm",
|
|
".xhtml",
|
|
]:
|
|
self.logger.info(
|
|
"Detected Office or HTML document, using parser for Office/HTML..."
|
|
)
|
|
content_list = doc_parser.parse_office_doc(
|
|
doc_path=file_path, output_dir=output_dir, **kwargs
|
|
)
|
|
else:
|
|
# For other or unknown formats, use generic parser
|
|
self.logger.info(
|
|
f"Using generic parser for {ext} file (method={parse_method})..."
|
|
)
|
|
content_list = doc_parser.parse_document(
|
|
file_path=file_path,
|
|
method=parse_method,
|
|
output_dir=output_dir,
|
|
**kwargs,
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.error(
|
|
f"Error during parsing with {self.config.parser} parser: {str(e)}"
|
|
)
|
|
self.logger.warning("Falling back to MinerU parser...")
|
|
# If specific parser fails, fall back to MinerU parser
|
|
content_list = MineruParser().parse_document(
|
|
file_path=file_path,
|
|
method=parse_method,
|
|
output_dir=output_dir,
|
|
**kwargs,
|
|
)
|
|
|
|
self.logger.info(
|
|
f"Parsing complete! Extracted {len(content_list)} content blocks"
|
|
)
|
|
|
|
# Generate doc_id based on content
|
|
doc_id = self._generate_content_based_doc_id(content_list)
|
|
|
|
# Store result in cache
|
|
await self._store_cached_result(
|
|
cache_key, content_list, doc_id, file_path, parse_method, **kwargs
|
|
)
|
|
|
|
# Display content statistics if requested
|
|
if display_stats:
|
|
self.logger.info("\nContent Information:")
|
|
self.logger.info(f"* Total blocks in content_list: {len(content_list)}")
|
|
|
|
# Count elements by type
|
|
block_types: Dict[str, int] = {}
|
|
for block in content_list:
|
|
if isinstance(block, dict):
|
|
block_type = block.get("type", "unknown")
|
|
if isinstance(block_type, str):
|
|
block_types[block_type] = block_types.get(block_type, 0) + 1
|
|
|
|
self.logger.info("* Content block types:")
|
|
for block_type, count in block_types.items():
|
|
self.logger.info(f" - {block_type}: {count}")
|
|
|
|
return content_list, doc_id
|
|
|
|
async def _process_multimodal_content(
|
|
self, multimodal_items: List[Dict[str, Any]], file_path: str, doc_id: str
|
|
):
|
|
"""
|
|
Process multimodal content (using specialized processors)
|
|
|
|
Args:
|
|
multimodal_items: List of multimodal items
|
|
file_path: File path (for reference)
|
|
doc_id: Document ID for proper chunk association
|
|
"""
|
|
if not multimodal_items:
|
|
self.logger.debug("No multimodal content to process")
|
|
return
|
|
|
|
# Check if multimodal content for this document is already processed
|
|
try:
|
|
existing_doc_status = await self.lightrag.doc_status.get_by_id(doc_id)
|
|
if existing_doc_status:
|
|
# Check if multimodal processing is already completed
|
|
multimodal_processed = existing_doc_status.get(
|
|
"multimodal_processed", False
|
|
)
|
|
existing_multimodal_chunks = existing_doc_status.get(
|
|
"multimodal_chunks_list", []
|
|
)
|
|
|
|
if multimodal_processed and len(existing_multimodal_chunks) >= len(
|
|
multimodal_items
|
|
):
|
|
self.logger.info(
|
|
f"Multimodal content already processed for document {doc_id} "
|
|
f"({len(existing_multimodal_chunks)} chunks found, {len(multimodal_items)} items to process)"
|
|
)
|
|
return
|
|
elif len(existing_multimodal_chunks) > 0:
|
|
self.logger.info(
|
|
f"Partial multimodal content found for document {doc_id} "
|
|
f"({len(existing_multimodal_chunks)} chunks exist, will reprocess all)"
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.debug(f"Error checking multimodal cache for {doc_id}: {e}")
|
|
# Continue with processing if cache check fails
|
|
|
|
self.logger.info("Starting multimodal content processing...")
|
|
|
|
file_name = os.path.basename(file_path)
|
|
|
|
# Collect all chunk results for batch processing (similar to text content processing)
|
|
all_chunk_results = []
|
|
multimodal_chunk_ids = []
|
|
|
|
# Get current text chunks count to set proper order indexes for multimodal chunks
|
|
|
|
existing_doc_status = await self.lightrag.doc_status.get_by_id(doc_id)
|
|
existing_chunks_count = (
|
|
existing_doc_status.get("chunks_count", 0) if existing_doc_status else 0
|
|
)
|
|
|
|
for i, item in enumerate(multimodal_items):
|
|
try:
|
|
content_type = item.get("type", "unknown")
|
|
self.logger.info(
|
|
f"Processing item {i+1}/{len(multimodal_items)}: {content_type} content"
|
|
)
|
|
|
|
# Select appropriate processor
|
|
processor = get_processor_for_type(self.modal_processors, content_type)
|
|
|
|
if processor:
|
|
# Prepare item info for context extraction
|
|
item_info = {
|
|
"page_idx": item.get("page_idx", 0),
|
|
"index": i,
|
|
"type": content_type,
|
|
}
|
|
|
|
# Process content and get chunk results instead of immediately merging
|
|
(
|
|
enhanced_caption,
|
|
entity_info,
|
|
chunk_results,
|
|
) = await processor.process_multimodal_content(
|
|
modal_content=item,
|
|
content_type=content_type,
|
|
file_path=file_name,
|
|
item_info=item_info, # Pass item info for context extraction
|
|
batch_mode=True,
|
|
doc_id=doc_id, # Pass doc_id for proper association
|
|
chunk_order_index=existing_chunks_count
|
|
+ i
|
|
+ 1, # Proper order index
|
|
)
|
|
|
|
# Collect chunk results for batch processing
|
|
all_chunk_results.extend(chunk_results)
|
|
|
|
# Extract chunk ID from the entity_info (actual chunk_id created by processor)
|
|
if entity_info and "chunk_id" in entity_info:
|
|
chunk_id = entity_info["chunk_id"]
|
|
multimodal_chunk_ids.append(chunk_id)
|
|
|
|
self.logger.info(
|
|
f"{content_type} processing complete: {entity_info.get('entity_name', 'Unknown')}"
|
|
)
|
|
else:
|
|
self.logger.warning(
|
|
f"No suitable processor found for {content_type} type content"
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error processing multimodal content: {str(e)}")
|
|
self.logger.debug("Exception details:", exc_info=True)
|
|
continue
|
|
|
|
# Update doc_status to include multimodal chunks
|
|
if multimodal_chunk_ids:
|
|
try:
|
|
# Get current document status
|
|
current_doc_status = await self.lightrag.doc_status.get_by_id(doc_id)
|
|
|
|
if current_doc_status:
|
|
existing_multimodal_chunks = current_doc_status.get(
|
|
"multimodal_chunks_list", []
|
|
)
|
|
|
|
# Combine existing chunks with new multimodal chunks
|
|
updated_multimodal_chunks_list = (
|
|
existing_multimodal_chunks + multimodal_chunk_ids
|
|
)
|
|
|
|
# Update document status with separated chunk lists
|
|
await self.lightrag.doc_status.upsert(
|
|
{
|
|
doc_id: {
|
|
**current_doc_status, # Keep existing fields
|
|
"multimodal_chunks_list": updated_multimodal_chunks_list, # Separated multimodal chunks
|
|
"multimodal_chunks_count": len(
|
|
updated_multimodal_chunks_list
|
|
),
|
|
"multimodal_processed": True, # Mark multimodal processing as complete
|
|
"updated_at": time.strftime("%Y-%m-%dT%H:%M:%S+00:00"),
|
|
}
|
|
}
|
|
)
|
|
|
|
# Ensure doc_status update is persisted to disk
|
|
await self.lightrag.doc_status.index_done_callback()
|
|
|
|
self.logger.info(
|
|
f"Updated doc_status with {len(multimodal_chunk_ids)} multimodal chunks"
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.warning(
|
|
f"Error updating doc_status with multimodal chunks: {e}"
|
|
)
|
|
|
|
# Batch merge all multimodal content results (similar to text content processing)
|
|
if all_chunk_results:
|
|
from lightrag.operate import merge_nodes_and_edges
|
|
from lightrag.kg.shared_storage import (
|
|
get_namespace_data,
|
|
get_pipeline_status_lock,
|
|
)
|
|
|
|
# Get pipeline status and lock from shared storage
|
|
pipeline_status = await get_namespace_data("pipeline_status")
|
|
pipeline_status_lock = get_pipeline_status_lock()
|
|
|
|
await merge_nodes_and_edges(
|
|
chunk_results=all_chunk_results,
|
|
knowledge_graph_inst=self.lightrag.chunk_entity_relation_graph,
|
|
entity_vdb=self.lightrag.entities_vdb,
|
|
relationships_vdb=self.lightrag.relationships_vdb,
|
|
global_config=self.lightrag.__dict__,
|
|
pipeline_status=pipeline_status,
|
|
pipeline_status_lock=pipeline_status_lock,
|
|
llm_response_cache=self.lightrag.llm_response_cache,
|
|
current_file_number=1,
|
|
total_files=1,
|
|
file_path=file_name,
|
|
)
|
|
|
|
await self.lightrag._insert_done()
|
|
|
|
self.logger.info("Multimodal content processing complete")
|
|
|
|
async def process_document_complete(
|
|
self,
|
|
file_path: str,
|
|
output_dir: str = None,
|
|
parse_method: str = None,
|
|
display_stats: bool = None,
|
|
split_by_character: str | None = None,
|
|
split_by_character_only: bool = False,
|
|
doc_id: str | None = None,
|
|
**kwargs,
|
|
):
|
|
"""
|
|
Complete document processing workflow
|
|
|
|
Args:
|
|
file_path: Path to the file to process
|
|
output_dir: output directory (defaults to config.parser_output_dir)
|
|
parse_method: Parse method (defaults to config.parse_method)
|
|
display_stats: Whether to display content statistics (defaults to config.display_content_stats)
|
|
split_by_character: Optional character to split the text by
|
|
split_by_character_only: If True, split only by the specified character
|
|
doc_id: Optional document ID, if not provided will be generated from content
|
|
**kwargs: Additional parameters for parser (e.g., lang, device, start_page, end_page, formula, table, backend, source)
|
|
"""
|
|
# Ensure LightRAG is initialized
|
|
await self._ensure_lightrag_initialized()
|
|
|
|
# Use config defaults if not provided
|
|
if output_dir is None:
|
|
output_dir = self.config.parser_output_dir
|
|
if parse_method is None:
|
|
parse_method = self.config.parse_method
|
|
if display_stats is None:
|
|
display_stats = self.config.display_content_stats
|
|
|
|
self.logger.info(f"Starting complete document processing: {file_path}")
|
|
|
|
# Step 1: Parse document
|
|
content_list, content_based_doc_id = await self.parse_document(
|
|
file_path, output_dir, parse_method, display_stats, **kwargs
|
|
)
|
|
|
|
# Use provided doc_id or fall back to content-based doc_id
|
|
if doc_id is None:
|
|
doc_id = content_based_doc_id
|
|
|
|
# Step 2: Separate text and multimodal content
|
|
text_content, multimodal_items = separate_content(content_list)
|
|
|
|
# Step 2.5: Set content source for context extraction in multimodal processing
|
|
if hasattr(self, "set_content_source_for_context") and multimodal_items:
|
|
self.logger.info(
|
|
"Setting content source for context-aware multimodal processing..."
|
|
)
|
|
self.set_content_source_for_context(
|
|
content_list, self.config.content_format
|
|
)
|
|
|
|
# Step 3: Insert pure text content with all parameters
|
|
if text_content.strip():
|
|
file_name = os.path.basename(file_path)
|
|
await insert_text_content(
|
|
self.lightrag,
|
|
text_content,
|
|
file_paths=file_name,
|
|
split_by_character=split_by_character,
|
|
split_by_character_only=split_by_character_only,
|
|
ids=doc_id,
|
|
)
|
|
|
|
# Step 4: Process multimodal content (using specialized processors)
|
|
if multimodal_items:
|
|
await self._process_multimodal_content(multimodal_items, file_path, doc_id)
|
|
else:
|
|
# If no multimodal content, mark as processed to avoid future checks
|
|
try:
|
|
existing_doc_status = await self.lightrag.doc_status.get_by_id(doc_id)
|
|
if existing_doc_status and not existing_doc_status.get(
|
|
"multimodal_processed", False
|
|
):
|
|
existing_multimodal_chunks = existing_doc_status.get(
|
|
"multimodal_chunks_list", []
|
|
)
|
|
|
|
await self.lightrag.doc_status.upsert(
|
|
{
|
|
doc_id: {
|
|
**existing_doc_status,
|
|
"multimodal_chunks_list": existing_multimodal_chunks,
|
|
"multimodal_chunks_count": len(
|
|
existing_multimodal_chunks
|
|
),
|
|
"multimodal_processed": True,
|
|
"updated_at": time.strftime("%Y-%m-%dT%H:%M:%S+00:00"),
|
|
}
|
|
}
|
|
)
|
|
await self.lightrag.doc_status.index_done_callback()
|
|
self.logger.debug(
|
|
f"Marked document {doc_id[:8]}... as having no multimodal content"
|
|
)
|
|
except Exception as e:
|
|
self.logger.debug(
|
|
f"Error updating doc_status for no multimodal content: {e}"
|
|
)
|
|
|
|
self.logger.info(f"Document {file_path} processing complete!")
|