diff --git a/common/chunkers/__init__.py b/common/chunkers/__init__.py index d508b42..d08ab60 100644 --- a/common/chunkers/__init__.py +++ b/common/chunkers/__init__.py @@ -1,6 +1,8 @@ from .base_chunker import BaseChunker from .character_chunker import CharacterChunker +from .html_chunker import HTMLChunker from .markdown_chunker import MarkdownChunker from .regex_chunker import RegexChunker from .semantic_chunker import SemanticChunker -from .recursive_chunker import RecursiveChunker \ No newline at end of file +from .recursive_chunker import RecursiveChunker +from .single_chunker import SingleChunker \ No newline at end of file diff --git a/common/chunkers/html_chunker.py b/common/chunkers/html_chunker.py new file mode 100644 index 0000000..ba84666 --- /dev/null +++ b/common/chunkers/html_chunker.py @@ -0,0 +1,84 @@ +# Copyright (c) 2025 TigerGraph, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from typing import Optional, List, Tuple +import re +from common.chunkers.base_chunker import BaseChunker +from langchain_text_splitters import HTMLSectionSplitter + + +class HTMLChunker(BaseChunker): + """ + HTML chunker that splits HTML content into chunks based on header tags. + + - Automatically detects which headers (h1-h6) are present in the HTML + - Uses only the headers that exist in the document for optimal chunking + - If custom headers are provided, uses those instead of auto-detection + """ + + def __init__( + self, + headers: Optional[List[Tuple[str, str]]] = None # e.g. [("h1", "Header 1"), ("h2", "Header 2")] + ): + self.headers = headers + + def _detect_headers(self, html_content: str) -> List[Tuple[str, str]]: + """ + Automatically detect which header tags (h1-h6) are present in the HTML. + Returns a list of header tuples for headers that exist in the document. + """ + # All possible headers in hierarchical order + all_headers = [ + ("h1", "Header 1"), + ("h2", "Header 2"), + ("h3", "Header 3"), + ("h4", "Header 4"), + ("h5", "Header 5"), + ("h6", "Header 6") + ] + + # Detect which headers are actually present in the HTML + detected_headers = [] + for tag, name in all_headers: + # Use regex to find header tags (case insensitive) + pattern = f'<{tag}[\\s>]' + if re.search(pattern, html_content, re.IGNORECASE): + detected_headers.append((tag, name)) + + # If no headers detected, use h1-h3 as fallback + if not detected_headers: + detected_headers = [ + ("h1", "Header 1"), + ("h2", "Header 2"), + ("h3", "Header 3") + ] + + return detected_headers + + def chunk(self, input_string: str) -> List[str]: + # Use custom headers if provided, otherwise auto-detect from HTML + if self.headers: + headers_to_use = self.headers + else: + headers_to_use = self._detect_headers(input_string) + + # Use HTMLSectionSplitter with detected/provided headers + splitter = HTMLSectionSplitter(headers_to_split_on=headers_to_use) + docs = splitter.split_text(input_string) + + # Extract text content from Document objects + return [doc.page_content for doc in docs] + + def __call__(self, input_string: str) -> List[str]: + return self.chunk(input_string) diff --git a/common/chunkers/single_chunker.py b/common/chunkers/single_chunker.py new file mode 100644 index 0000000..cda2db8 --- /dev/null +++ b/common/chunkers/single_chunker.py @@ -0,0 +1,30 @@ +""" +Single Chunker - Always returns the entire content as ONE chunk. +Used for images to preserve [IMAGE_REF:] markers and prevent splitting. +""" +from common.chunkers.base_chunker import BaseChunker + + +class SingleChunker(BaseChunker): + """ + Chunker that NEVER splits content - always returns ONE chunk. + + This is critical for image descriptions to: + 1. Keep [IMAGE_REF:] markers intact + 2. Prevent losing image references when displayed in UI + 3. Maintain semantic integrity of image descriptions + """ + + def chunk(self, text: str) -> list[str]: + """ + Return the entire text as a single chunk, regardless of length. + + Args: + text: The text to "chunk" (actually just return as-is) + + Returns: + List with single element containing all text + """ + # Always return ONE chunk with entire content + return [text] if text and text.strip() else [] + diff --git a/common/config.py b/common/config.py index 1f3e72c..010c319 100644 --- a/common/config.py +++ b/common/config.py @@ -91,6 +91,41 @@ raise Exception("embedding_service is not found in llm_config") embedding_dimension = embedding_config.get("dimensions", 1536) +# Get multimodal_service config (optional, for vision/image tasks) +multimodal_config = llm_config.get("multimodal_service") + +# Merge shared authentication configuration from llm_config level into service configs +# Services can still override by defining their own authentication_configuration +shared_auth = llm_config.get("authentication_configuration", {}) +if shared_auth: + # Merge into embedding_config (service-specific auth takes precedence) + if "authentication_configuration" not in embedding_config: + embedding_config["authentication_configuration"] = shared_auth.copy() + else: + # Merge shared auth with service-specific auth (service-specific takes precedence) + merged_embedding_auth = shared_auth.copy() + merged_embedding_auth.update(embedding_config["authentication_configuration"]) + embedding_config["authentication_configuration"] = merged_embedding_auth + + # Merge into completion_config (service-specific auth takes precedence) + if "authentication_configuration" not in completion_config: + completion_config["authentication_configuration"] = shared_auth.copy() + else: + # Merge shared auth with service-specific auth (service-specific takes precedence) + merged_completion_auth = shared_auth.copy() + merged_completion_auth.update(completion_config["authentication_configuration"]) + completion_config["authentication_configuration"] = merged_completion_auth + + # Merge into multimodal_config if it exists (service-specific auth takes precedence) + if multimodal_config: + if "authentication_configuration" not in multimodal_config: + multimodal_config["authentication_configuration"] = shared_auth.copy() + else: + # Merge shared auth with service-specific auth (service-specific takes precedence) + merged_multimodal_auth = shared_auth.copy() + merged_multimodal_auth.update(multimodal_config["authentication_configuration"]) + multimodal_config["authentication_configuration"] = merged_multimodal_auth + if graphrag_config is None: graphrag_config = {"reuse_embedding": True} if "chunker" not in graphrag_config: @@ -149,6 +184,38 @@ def get_llm_service(llm_config) -> LLM_Model: else: raise Exception("LLM Completion Service Not Supported") +def get_multimodal_service() -> LLM_Model: + """ + Get the multimodal/vision LLM service for image description tasks. + Uses multimodal_service if configured, otherwise falls back to completion_service. + Currently supports: OpenAI, Azure, GenAI, VertexAI + """ + # Use multimodal_service if available, otherwise fallback to completion_service + service_config = multimodal_config if multimodal_config else completion_config + + # Make a copy to avoid modifying the original config + config_copy = service_config.copy() + + # Add default prompt_path if not present (required by LLM service classes but not used for multimodal) + if "prompt_path" not in config_copy: + config_copy["prompt_path"] = "./common/prompts/openai_gpt4/" + + service_type = config_copy["llm_service"].lower() + + if service_type == "openai": + return OpenAI(config_copy) + elif service_type == "azure": + return AzureOpenAI(config_copy) + elif service_type == "genai": + return GoogleGenAI(config_copy) + elif service_type == "vertexai": + return GoogleVertexAI(config_copy) + else: + raise Exception( + f"Multimodal service '{service_type}' not supported. " + "Only OpenAI, Azure, GenAI, and VertexAI are currently supported for vision tasks." + ) + if os.getenv("INIT_EMBED_STORE", "true") == "true": conn = TigerGraphConnection( host=db_config.get("hostname", "http://tigergraph"), diff --git a/common/gsql/supportai/SupportAI_InitialLoadJSON_WithImages.gsql b/common/gsql/supportai/SupportAI_InitialLoadJSON_WithImages.gsql new file mode 100644 index 0000000..7ee5bb2 --- /dev/null +++ b/common/gsql/supportai/SupportAI_InitialLoadJSON_WithImages.gsql @@ -0,0 +1,41 @@ +CREATE LOADING JOB load_documents_content_json_with_images_@uuid@ { + DEFINE FILENAME DocumentContent; + + // Load Documents (only entries with content field, not image storage entries) + LOAD DocumentContent TO VERTEX Document VALUES(gsql_lower($"doc_id"), gsql_current_time_epoch(0), _, _) + WHERE $"content" != "" + USING JSON_FILE="true"; + + // Load Content (for documents with content field) + // Both markdown and standalone image documents create Content vertices + LOAD DocumentContent TO VERTEX Content VALUES(gsql_lower($"doc_id"), $"doc_type", $"content", gsql_current_time_epoch(0)) + WHERE $"content" != "" + USING JSON_FILE="true"; + + // Edge: Document -> Content (for documents with content) + LOAD DocumentContent TO EDGE HAS_CONTENT VALUES( + gsql_lower($"doc_id") Document, + gsql_lower($"doc_id") Content + ) WHERE $"content" != "" + USING JSON_FILE="true"; + + // Load Images (ONLY for image storage entries which have image_data field) + LOAD DocumentContent TO VERTEX Image VALUES( + gsql_lower($"doc_id"), + $"image_data", + $"image_format", + gsql_lower($"parent_doc"), + $"page_number", + gsql_current_time_epoch(0) + ) WHERE $"image_data" != "" + USING JSON_FILE="true"; + + // Edge: Document -> Image (for image documents) + LOAD DocumentContent TO EDGE HAS_IMAGE VALUES( + gsql_lower($"parent_doc") Document, + gsql_lower($"doc_id") Image, + $"position" + ) WHERE $"doc_type" == "image" AND $"parent_doc" != "" + USING JSON_FILE="true"; +} + diff --git a/common/gsql/supportai/SupportAI_Schema_Images.gsql b/common/gsql/supportai/SupportAI_Schema_Images.gsql new file mode 100644 index 0000000..a05ffd6 --- /dev/null +++ b/common/gsql/supportai/SupportAI_Schema_Images.gsql @@ -0,0 +1,35 @@ +/* + * Copyright (c) 2025 TigerGraph, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. +*/ + +CREATE SCHEMA_CHANGE JOB add_image_support_schema { + ADD VERTEX Image( + PRIMARY_ID id STRING, + image_data STRING, // base64 encoded image data + image_format STRING, // jpg, png, gif, etc. + source_document STRING, // parent document ID + page_number INT, // for PDFs, which page + epoch_added UINT + ) WITH STATS="OUTDEGREE_BY_EDGETYPE", PRIMARY_ID_AS_ATTRIBUTE="true"; + + // Document can have multiple images (standalone or embedded) + ADD DIRECTED EDGE HAS_IMAGE(FROM Document, TO Image, position_index INT) + WITH REVERSE_EDGE="reverse_HAS_IMAGE"; + + // DocumentChunk can reference images that appear in its content + ADD DIRECTED EDGE REFERENCES_IMAGE(FROM DocumentChunk, TO Image) + WITH REVERSE_EDGE="reverse_REFERENCES_IMAGE"; +} + diff --git a/common/prompts/google_gemini/chatbot_response.txt b/common/prompts/google_gemini/chatbot_response.txt index 5865304..cb23c53 100644 --- a/common/prompts/google_gemini/chatbot_response.txt +++ b/common/prompts/google_gemini/chatbot_response.txt @@ -20,10 +20,11 @@ Example format for responses: - **Escalation**: "This issue may require human assistance. Please reach out via [Support Portal](https://www.tigergraph.com/support/) or email us at [support@tigergraph.com](mailto:support@tigergraph.com)." Your mission is to ensure a seamless, satisfying customer experience while upholding TigerGraph's values and commitment to enterprise excellence. +Make sure to extract and include the image references in [IMAGE_REF:image_id] format in your generated_answer if they are present in the context. Images are critical visual information that must be included in your response. Do NOT modify or omit these image references. Give the context in JSON format, combine and rephrase it to answer the question. Use only the provided information in question and context without adding any reasoning or additional logic. -Make sure all information in the question and context are covered in the generated answer. +Make sure all information in the question and context are covered in the generated answer, including any image markdown references. Generate the answer in JSON format, make sure to escape necessary characters in order to return a valid JSON response only. Make sure all the fields required by the format instructions are included, set a field to empty if you don't have that information. diff --git a/common/prompts/openai_gpt4/chatbot_response.txt b/common/prompts/openai_gpt4/chatbot_response.txt index c2b8333..b9cab1a 100644 --- a/common/prompts/openai_gpt4/chatbot_response.txt +++ b/common/prompts/openai_gpt4/chatbot_response.txt @@ -15,13 +15,11 @@ Example format for responses: - **Escalation**: "This issue may require human assistance. Please reach out via [Support Portal](https://www.tigergraph.com/support/) or email us at [support@tigergraph.com](mailto:support@tigergraph.com)." Format your answer using Markdown. Organize the content into paragraphs, bulleted or numbered lists, and include links to images where relevant. +Make sure to extract and include the image references in [IMAGE_REF:image_id] format in your generated_answer if they are present in the context. Images are critical visual information that must be included in your response. Do NOT modify or omit these image references. Give the context in JSON format, combine and rephrase it to answer the question. Use only the provided information in context without adding any reasoning or additional logic. -Make sure all information in the context are covered in the generated answer. -Make sure to extract and include the image links in markdown syntax in the generated answer when their summaries are referenced, and preserve the link URLs in their original format. -Use compact markdown syntax to geneate the answer, including title, bulleted or numbered list, images and tables if any, and place images or tables below the related text section. -Ensure that each row of every table, including the header row, starts on a new line. +Make sure all information in the context are covered in the generated answer, including any image markdown references. Generate the answer in JSON format, make sure to escape necessary characters in order to return a valid JSON response only. Make sure all the fields required by the format instructions are included, set a field to empty if you don't have that information. diff --git a/common/py_schemas/schemas.py b/common/py_schemas/schemas.py index 624c729..c5aee80 100644 --- a/common/py_schemas/schemas.py +++ b/common/py_schemas/schemas.py @@ -119,7 +119,7 @@ class CreateVectorIndexConfig(BaseModel): class CreateIngestConfig(BaseModel): data_source: str data_source_config: Dict - loader_config: Dict = {"doc_id_field": str, "content_field": str} + loader_config: Optional[Dict] = None # Made optional - will auto-generate defaults file_format: str = "json" diff --git a/common/requirements.txt b/common/requirements.txt index 77be241..42c7c48 100644 --- a/common/requirements.txt +++ b/common/requirements.txt @@ -108,8 +108,9 @@ ordered-set==4.1.0 orjson==3.10.18 packaging==24.2 pandas==2.2.3 -pathtools==0.1.2 +#pathtools==0.1.2 pillow==11.2.1 +PyMuPDF==1.26.4 platformdirs==4.3.8 pluggy==1.6.0 prometheus_client==0.22.1 @@ -127,6 +128,8 @@ pygit2==1.18.0 pyparsing==3.2.3 pypdf==5.6.1 pytest==8.4.1 +python-docx==1.1.2 +pytesseract==0.3.10 python-dateutil==2.9.0.post0 python-dotenv==1.1.1 python-iso639==2025.2.18 diff --git a/common/utils/image_data_extractor.py b/common/utils/image_data_extractor.py new file mode 100644 index 0000000..a6e025a --- /dev/null +++ b/common/utils/image_data_extractor.py @@ -0,0 +1,164 @@ +import base64 +import io +import logging +import os +import uuid +import hashlib +from pathlib import Path +from langchain_core.messages import HumanMessage, SystemMessage + +from common.config import get_multimodal_service + +logger = logging.getLogger(__name__) + + + +def describe_image_with_llm(image_input): + """ + Send image (pixmap or PIL image) to LLM vision model and return description. + Uses multimodal_service from config if available, otherwise falls back to completion_service. + Currently supports: OpenAI, Azure OpenAI, Google GenAI, and Google VertexAI + """ + try: + client = get_multimodal_service() + if not client: + return "[Image: Failed to create multimodal LLM client]" + + buffer = io.BytesIO() + # Convert to RGB if needed for better compatibility + if image_input.mode != 'RGB': + image_input = image_input.convert('RGB') + image_input.save(buffer, format="JPEG", quality=95) + b64_img = base64.b64encode(buffer.getvalue()).decode("utf-8") + + # Build messages (system + human) + messages = [ + SystemMessage( + content="You are a helpful assistant that describes images concisely for document analysis." + ), + HumanMessage( + content=[ + { + "type": "text", + "text": ( + "Please describe what you see in this image and " + "if the image has scanned text then extract all the text. " + "Focus on any text, diagrams, charts, or other visual elements." + "If this is a logo, icon, or branding element, start your response with 'LOGO:' or 'ICON:'." + ), + }, + { + "type": "image_url", + "image_url": {"url": f"data:image/jpeg;base64,{b64_img}"}, + }, + ] + ), + ] + + # Get response from LangChain LLM client + # Access the underlying LangChain client + langchain_client = client.llm + response = langchain_client.invoke(messages) + + return response.content if hasattr(response, 'content') else str(response) + + except Exception as e: + logger.error(f"Failed to describe image with LLM: {str(e)}") + return "[Image: Error processing image description]" + + +def save_image_and_get_markdown(image_input, context_info="", graphname=None): + """ + Save image locally to static/images/ folder and return markdown reference with description. + + LEGACY/OLD APPROACH: Used for backward compatibility with JSONL-based loading. + Images are saved as files and served via /ui/images/ endpoint with img:// protocol. + + For NEW direct loading approach, images are stored in Image vertex as base64 + and served via /ui/image_vertex/ endpoint with image:// protocol. + + Args: + image_input: PIL Image object + context_info: Optional context (e.g., "page 3 of invoice.pdf") + graphname: Graph name to organize images by graph (optional) + + Returns: + dict with: + - 'markdown': Markdown string with img:// reference + - 'image_id': Unique identifier for the saved image + - 'image_path': Path where image was saved to static/images/ + """ + try: + # FIRST: Get description from LLM to check if it's a logo + description = describe_image_with_llm(image_input) + + # Check if the image is a logo, icon, or decorative element BEFORE saving + # These should be filtered out as they're not content-relevant + description_lower = description.lower() + logo_indicators = ['logo', 'icon', 'branding', 'watermark', 'trademark', 'company logo', 'brand logo'] + + if any(indicator in description_lower for indicator in logo_indicators): + logger.info(f"Detected logo/icon in image, skipping: {description[:100]}") + return None + + # If not a logo, proceed with saving the image + # Generate unique image ID using hash of image content + buffer = io.BytesIO() + if image_input.mode != 'RGB': + image_input = image_input.convert('RGB') + image_input.save(buffer, format="JPEG", quality=95) + image_bytes = buffer.getvalue() + + # Create hash-based ID (deterministic for same image) + image_hash = hashlib.sha256(image_bytes).hexdigest()[:16] + image_id = f"{image_hash}.jpg" + + # Save image to local storage directory organized by graphname + project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + + # If graphname is provided, organize images by graph + if graphname: + images_dir = os.path.join(project_root, "static", "images", graphname) + # Include graphname in the image reference for URL construction + image_reference = f"{graphname}/{image_id}" + else: + images_dir = os.path.join(project_root, "static", "images") + image_reference = image_id + + os.makedirs(images_dir, exist_ok=True) + + image_path = os.path.join(images_dir, image_id) + + # Save image file (skip if already exists with same hash) + if not os.path.exists(image_path): + with open(image_path, 'wb') as f: + f.write(image_bytes) + logger.info(f"Saved content image to: {image_path}") + else: + logger.debug(f"Image already exists: {image_path}") + + # Generate markdown with custom img:// protocol (will be replaced later) + # Format: ![description](img://graphname/image_id) or ![description](img://image_id) + markdown = f"![{description}](img://{image_reference})" + + logger.info(f"Created image reference: {image_reference} with description") + + return { + 'markdown': markdown, + 'image_id': image_reference, + 'image_path': image_path, + 'description': description + } + + except Exception as e: + logger.error(f"Failed to save image and generate markdown: {str(e)}") + # Fallback to text description only + fallback_desc = f"[Image: {context_info} - processing failed]" + return { + 'markdown': fallback_desc, + 'image_id': None, + 'image_path': None, + 'description': fallback_desc + } + + diff --git a/common/utils/text_extractors.py b/common/utils/text_extractors.py new file mode 100644 index 0000000..e2bc856 --- /dev/null +++ b/common/utils/text_extractors.py @@ -0,0 +1,454 @@ +""" +Text extraction utilities for various file formats. +This module handles the extraction of text content from different document types. +""" +import os +import json +import logging +import uuid +import base64 +import io +from pathlib import Path +import shutil +import asyncio +from concurrent.futures import ThreadPoolExecutor + +logger = logging.getLogger(__name__) + + +class TextExtractor: + """Class for handling text extraction from various file formats and cleanup.""" + + def __init__(self): + """Initialize the TextExtractor.""" + self.supported_extensions = { + '.txt': 'text/plain', + '.md': 'text/markdown', + '.pdf': 'application/pdf', + '.docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document', + '.doc': 'application/msword', + '.html': 'text/html', + '.htm': 'text/html', + '.json': 'application/json', + '.csv': 'text/csv', + '.xlsx': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', + '.xls': 'application/vnd.ms-excel', + '.xml': 'application/xml', + '.jpeg': 'image/jpeg', + '.jpg': 'image/jpeg' + } + + async def _process_file_async(self, file_path, folder_path_obj, graphname): + """ + Async helper to process a single file. + Runs in thread pool to avoid blocking on I/O operations. + """ + try: + loop = asyncio.get_event_loop() + + doc_entries = await loop.run_in_executor( + None, + extract_text_from_file_with_images_as_docs, + file_path, + graphname + ) + + return { + 'success': True, + 'file_path': str(file_path), + 'documents': doc_entries, + 'num_documents': len(doc_entries) + } + + except FileNotFoundError: + return {'success': False, 'file_path': str(file_path), 'error': 'File not found'} + except PermissionError: + return {'success': False, 'file_path': str(file_path), 'error': 'Permission denied'} + except Exception as e: + logger.warning(f"Failed to process file {file_path}: {e}") + return {'success': False, 'file_path': str(file_path), 'error': str(e)} + + async def _process_folder_async(self, folder_path, graphname=None, max_concurrent=10): + """ + Async version of process_folder for parallel file processing. + This prevents conflicts when multiple users process folders simultaneously. + """ + logger.info(f"Processing local folder ASYNC: {folder_path} for graph: {graphname} (max_concurrent={max_concurrent})") + + folder_path_obj = Path(folder_path) + + if not folder_path_obj.exists(): + raise Exception(f"Folder path does not exist: {folder_path}") + + if not folder_path_obj.is_dir(): + raise Exception(f"Path is not a directory: {folder_path}") + + def safe_walk(path): + try: + for item in path.iterdir(): + if item.name.startswith(('.', '~', '$')) or 'BROMIUM' in item.name.upper(): + continue + if item.is_file(): + yield item + elif item.is_dir(): + yield from safe_walk(item) + except (PermissionError, OSError) as e: + logger.warning(f"Cannot access directory {path}: {e}") + + files_to_process = [] + for file_path in safe_walk(folder_path_obj): + if file_path.is_file(): + if file_path.name.startswith(('.', '~', '$')) or 'BROMIUM' in file_path.name.upper(): + continue + file_ext = file_path.suffix.lower() + if file_ext in self.supported_extensions: + files_to_process.append(file_path) + + logger.info(f"Found {len(files_to_process)} files to process") + + semaphore = asyncio.Semaphore(max_concurrent) + + async def process_with_semaphore(file_path): + async with semaphore: + return await self._process_file_async(file_path, folder_path_obj, graphname) + + tasks = [process_with_semaphore(fp) for fp in files_to_process] + results = await asyncio.gather(*tasks, return_exceptions=True) + + all_documents = [] + processed_files_info = [] + + for result in results: + if isinstance(result, Exception): + logger.error(f"File processing failed with exception: {result}") + continue + + if result.get('success'): + all_documents.extend(result.get('documents', [])) + processed_files_info.append({ + 'file_path': result['file_path'], + 'num_documents': result.get('num_documents', len(result.get('documents', []))), + 'status': 'success' + }) + else: + processed_files_info.append({ + 'file_path': result['file_path'], + 'status': 'failed', + 'error': result.get('error', 'Unknown error') + }) + + logger.info(f"Processed {len(processed_files_info)} files, extracted {len(all_documents)} total documents") + + return { + 'statusCode': 200, + 'message': f'Processed {len(processed_files_info)} files, {len(all_documents)} documents', + 'documents': all_documents, + 'files': processed_files_info, + 'num_documents': len(all_documents) + } + + def process_folder(self, folder_path, graphname=None): + """ + Process local folder with multiple file formats and extract text content. + Uses async processing internally for parallel file handling. + """ + logger.info(f"Processing local folder: {folder_path} for graph: {graphname}") + return asyncio.run(self._process_folder_async(folder_path, graphname)) + + +def extract_text_from_file_with_images_as_docs(file_path, graphname=None): + """ + Extract text and images from a file, treating images as separate document entries. + """ + file_path = Path(file_path) + extension = file_path.suffix.lower() + base_doc_id = str(file_path.stem) + + logger.debug(f"Extracting with images as docs: {file_path} (type: {extension})") + + if extension == '.pdf': + return _extract_pdf_with_images_as_docs(file_path, base_doc_id, graphname) + elif extension in ['.jpeg', '.jpg', '.png', '.gif']: + return _extract_standalone_image_as_doc(file_path, base_doc_id, graphname) + else: + content = extract_text_from_file(file_path, graphname) + doc_type = get_doc_type_from_extension(extension) + return [{ + "doc_id": base_doc_id, + "doc_type": doc_type, + "content": content, + "position": 0 + }] + + +def _extract_pdf_with_images_as_docs(file_path, base_doc_id, graphname=None): + """ + Extract PDF as ONE markdown document with inline image references. + """ + try: + import fitz # PyMuPDF + from PIL import Image as PILImage + + doc = fitz.open(file_path) + markdown_parts = [] + image_entries = [] + image_counter = 0 + + for page_num, page in enumerate(doc, start=1): + if page_num > 1: + markdown_parts.append("\n\n") + markdown_parts.append(f"--- Page {page_num} ---\n\n") + + blocks = page.get_text("blocks", sort=True) + text_blocks_with_pos = [] + + for block in blocks: + block_type = block[6] if len(block) > 6 else 0 + if block_type == 0: + text = block[4].strip() + if text: + y_pos = block[1] + text_blocks_with_pos.append({'type': 'text', 'content': text, 'y_pos': y_pos}) + + image_list = page.get_images(full=True) + images_with_pos = [] + + if image_list: + for img_index, img_info in enumerate(image_list): + try: + xref = img_info[0] + base_image = doc.extract_image(xref) + image_bytes = base_image["image"] + image_ext = base_image["ext"] + + img_rects = page.get_image_rects(xref) + y_pos = img_rects[0].y0 if img_rects else 999999 + + pil_image = PILImage.open(io.BytesIO(image_bytes)) + if pil_image.width < 100 or pil_image.height < 100: + continue + + from common.utils.image_data_extractor import describe_image_with_llm + description = describe_image_with_llm(pil_image) + description_lower = description.lower() + logo_indicators = [ + 'logo:', 'icon:', 'logo', 'icon', 'branding', + 'watermark', 'trademark', 'stylized letter', + 'stylized text', 'word "', "word '" + ] + if any(indicator in description_lower for indicator in logo_indicators): + continue + + buffer = io.BytesIO() + if pil_image.mode != 'RGB': + pil_image = pil_image.convert('RGB') + pil_image.save(buffer, format="JPEG", quality=95) + image_base64 = base64.b64encode(buffer.getvalue()).decode('utf-8') + + image_counter += 1 + image_doc_id = f"{base_doc_id}_image_{image_counter}" + + images_with_pos.append({ + 'type': 'image', + 'image_doc_id': image_doc_id, + 'description': description, + 'y_pos': y_pos, + 'image_data': image_base64, + 'image_format': image_ext, + 'width': pil_image.width, + 'height': pil_image.height + }) + except Exception as img_error: + logger.warning(f"Failed to extract image on page {page_num}: {img_error}") + + all_elements = text_blocks_with_pos + images_with_pos + all_elements.sort(key=lambda x: x['y_pos']) + + for element in all_elements: + if element['type'] == 'text': + markdown_parts.append(element['content']) + markdown_parts.append("\n\n") + else: + markdown_parts.append("### Image Description\n\n") + markdown_parts.append(element['description']) + markdown_parts.append(f"\n\n[IMAGE_REF:{element['image_doc_id']}]\n\n") + + image_entries.append({ + "doc_id": element['image_doc_id'], + "doc_type": "image", + "image_description": element['description'], + "image_data": element['image_data'], + "image_format": element['image_format'], + "parent_doc": base_doc_id, + "page_number": page_num, + "width": element['width'], + "height": element['height'], + "position": int(element['image_doc_id'].split('_')[-1]) + }) + + doc.close() + + markdown_content = "".join(markdown_parts) if markdown_parts else "[No content extracted from PDF]" + result = [{ + "doc_id": base_doc_id, + "doc_type": "markdown", + "content": markdown_content, + "position": 0 + }] + result.extend(image_entries) + return result + + except ImportError: + logger.error("PyMuPDF not available") + return [{ + "doc_id": base_doc_id, + "doc_type": "markdown", + "content": "[PDF extraction requires PyMuPDF]", + "position": 0 + }] + except Exception as e: + logger.error(f"Error extracting PDF: {e}") + raise + + +def _extract_standalone_image_as_doc(file_path, base_doc_id, graphname=None): + """ + Extract standalone image file as ONE markdown document with inline image reference. + """ + try: + from PIL import Image as PILImage + from common.utils.image_data_extractor import describe_image_with_llm + + pil_image = PILImage.open(file_path) + if pil_image.width < 100 or pil_image.height < 100: + return [{ + "doc_id": base_doc_id, + "doc_type": "markdown", + "content": f"[Skipped small image: {file_path.name}]", + "position": 0 + }] + + description = describe_image_with_llm(pil_image) + description_lower = description.lower() + logo_indicators = ['logo:', 'icon:', 'logo', 'icon', 'branding', + 'watermark', 'trademark', 'stylized letter', + 'stylized text', 'word "', "word '"] + if any(indicator in description_lower for indicator in logo_indicators): + return [{ + "doc_id": base_doc_id, + "doc_type": "markdown", + "content": f"[Skipped logo/icon: {file_path.name}]", + "position": 0 + }] + + buffer = io.BytesIO() + if pil_image.mode != 'RGB': + pil_image = pil_image.convert('RGB') + pil_image.save(buffer, format="JPEG", quality=95) + image_base64 = base64.b64encode(buffer.getvalue()).decode('utf-8') + + image_id = f"{base_doc_id}_image_1" + content = f"{description}\n\n[IMAGE_REF:{image_id}]" + + return [ + { + "doc_id": base_doc_id, + "doc_type": "image", + "content": content, + "position": 0 + }, + { + "doc_id": image_id, + "doc_type": "image", + "image_description": description, + "image_data": image_base64, + "image_format": "jpg", + "parent_doc": base_doc_id, + "page_number": 0, + "width": pil_image.width, + "height": pil_image.height, + "position": 1 + } + ] + + except Exception as e: + logger.error(f"Error extracting image: {e}") + return [{ + "doc_id": base_doc_id, + "doc_type": "markdown", + "content": f"[Image extraction failed: {str(e)}]", + "position": 0 + }] + + +def extract_text_from_file(file_path, graphname=None): + """ + Extract text content from a file based on its extension. + """ + file_path = Path(file_path) + extension = file_path.suffix.lower() + + logger.debug(f"Extracting text from {file_path} (type: {extension}) for graph: {graphname}") + + try: + if extension in ['.txt', '.md']: + with open(file_path, 'r', encoding='utf-8') as f: + return f.read().strip() + elif extension in ['.html', '.htm', '.csv']: + with open(file_path, 'r', encoding='utf-8') as f: + return f.read().strip() + elif extension == '.json': + with open(file_path, 'r', encoding='utf-8') as f: + data = json.load(f) + return json.dumps(data, indent=2, ensure_ascii=False) + elif extension == '.docx': + import docx + doc = docx.Document(file_path) + return "\n".join(p.text for p in doc.paragraphs if p.text.strip()) + elif extension == '.xml': + import xml.etree.ElementTree as ET + tree = ET.parse(file_path) + root = tree.getroot() + + def extract_text_from_element(element): + text = element.text or "" + for child in element: + text += " " + extract_text_from_element(child) + if element.tail: + text += " " + element.tail + return text.strip() + + content = extract_text_from_element(root) + import re + return re.sub(r'\s+', ' ', content).strip() + else: + return f"[Unsupported file type: {extension}]" + + except Exception as e: + logger.error(f"Error extracting text from {file_path}: {e}") + raise Exception(f"Text extraction failed: {e}") + + +def get_doc_type_from_extension(extension): + """Map file extension to a chunker-compatible document type.""" + if not extension.startswith('.'): + extension = '.' + extension + extension = extension.lower() + + if extension in ['.html', '.htm']: + return 'html' + elif extension in ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff', '.webp']: + return 'image' + else: + return 'markdown' + + +def get_supported_extensions(): + """Get list of supported file extensions.""" + return {'.txt', '.md', '.html', '.htm', '.csv', '.json', '.pdf', '.docx', '.xml', '.jpeg', '.jpg', '.png', '.gif'} + + +def is_supported_file(file_path): + """Check if a file is supported for text extraction.""" + extension = Path(file_path).suffix.lower() + return extension in get_supported_extensions() diff --git a/configs/server_config.json b/configs/server_config.json index 41764e4..d607a10 100644 --- a/configs/server_config.json +++ b/configs/server_config.json @@ -14,23 +14,27 @@ "chat_history_api": "http://chat-history:8002" }, "llm_config": { + "authentication_configuration": { + "OPENAI_API_KEY": "YOUR_OPENAI_API_KEY_HERE" + }, "embedding_service": { "model_name": "text-embedding-3-small", - "embedding_model_service": "openai", - "authentication_configuration": { - "OPENAI_API_KEY": "YOUR_OPENAI_API_KEY_HERE" - } + "embedding_model_service": "openai" }, "completion_service": { "llm_service": "openai", "llm_model": "gpt-4.1-mini", - "authentication_configuration": { - "OPENAI_API_KEY": "YOUR_OPENAI_API_KEY_HERE" - }, "model_kwargs": { "temperature": 0 }, "prompt_path": "./common/prompts/openai_gpt4/" + }, + "multimodal_service": { + "llm_service": "openai", + "llm_model": "gpt-4o-mini", + "model_kwargs": { + "temperature": 0 + } } }, "chat_config": { diff --git a/docker-compose.yml b/docker-compose.yml index 8be754b..29d4cad 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -18,6 +18,7 @@ services: USE_CYPHER: "true" volumes: - ./configs/:/code/configs + - YOUR_DATA_PATH_HERE:/data graphrag-ecc: image: tigergraph/graphrag-ecc:latest diff --git a/docs/notebooks/GraphRAGDemo.ipynb b/docs/notebooks/GraphRAGDemo.ipynb index 96e444c..78587fc 100644 --- a/docs/notebooks/GraphRAGDemo.ipynb +++ b/docs/notebooks/GraphRAGDemo.ipynb @@ -126,6 +126,21 @@ "print(res)" ] }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# for server multi files \n", + "res=conn.ai.createDocumentIngest(\n", + " data_source=\"server\",\n", + " data_source_config={\"folder_path\": \"/data\"}, # Docker volume mount path\n", + " loader_config={},\n", + " file_format=\"multi\" # Automatically uses direct loading (like Bedrock)\n", + " )\n" + ] + }, { "cell_type": "markdown", "metadata": {}, @@ -142,6 +157,20 @@ "conn.ai.runDocumentIngest(res[\"load_job_id\"], res[\"data_source_id\"], res[\"data_path\"])" ] }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# for server multi \n", + "conn.ai.runDocumentIngest(\n", + " res[\"load_job_id\"],\n", + " res[\"data_source_id\"],\n", + " res[\"data_path\"]\n", + " )\n" + ] + }, { "cell_type": "markdown", "metadata": {}, diff --git a/ecc/app/common b/ecc/app/common deleted file mode 120000 index 248927d..0000000 --- a/ecc/app/common +++ /dev/null @@ -1 +0,0 @@ -../../common/ \ No newline at end of file diff --git a/ecc/app/common b/ecc/app/common new file mode 100644 index 0000000..248927d --- /dev/null +++ b/ecc/app/common @@ -0,0 +1 @@ +../../common/ \ No newline at end of file diff --git a/ecc/app/configs b/ecc/app/configs deleted file mode 120000 index 5992d10..0000000 --- a/ecc/app/configs +++ /dev/null @@ -1 +0,0 @@ -../../configs \ No newline at end of file diff --git a/ecc/app/configs b/ecc/app/configs new file mode 100644 index 0000000..5992d10 --- /dev/null +++ b/ecc/app/configs @@ -0,0 +1 @@ +../../configs \ No newline at end of file diff --git a/ecc/app/ecc_util.py b/ecc/app/ecc_util.py index fe50de8..8d19c48 100644 --- a/ecc/app/ecc_util.py +++ b/ecc/app/ecc_util.py @@ -1,4 +1,4 @@ -from common.chunkers import character_chunker, regex_chunker, semantic_chunker, markdown_chunker, recursive_chunker +from common.chunkers import character_chunker, regex_chunker, semantic_chunker, markdown_chunker, recursive_chunker, html_chunker, single_chunker from common.config import graphrag_config, embedding_service, llm_config from common.llm_services import ( AWS_SageMaker_Endpoint, @@ -36,11 +36,19 @@ def get_chunker(chunker_type: str = ""): chunk_size=chunker_config.get("chunk_size", 0), chunk_overlap=chunker_config.get("overlap_size", 0), ) + elif chunker_type == "html": + chunker = html_chunker.HTMLChunker( + headers=chunker_config.get("headers", None) + ) elif chunker_type == "recursive": chunker = recursive_chunker.RecursiveChunker( chunk_size=chunker_config.get("chunk_size", 1024), overlap_size=chunker_config.get("overlap_size", 0), ) + elif chunker_type == "single" or chunker_type == "image": + # Single chunker: NEVER splits, always returns 1 chunk + # Used for images to preserve [IMAGE_REF:] markers + chunker = single_chunker.SingleChunker() else: raise ValueError(f"Invalid chunker type: {chunker_type}") diff --git a/ecc/app/graphrag/workers.py b/ecc/app/graphrag/workers.py index e439304..1317322 100644 --- a/ecc/app/graphrag/workers.py +++ b/ecc/app/graphrag/workers.py @@ -91,15 +91,19 @@ async def chunk_doc( chunker_type = doc["attributes"]["ctype"].lower().strip() else: chunker_type = "" - chunker = ecc_util.get_chunker(chunker_type) - # decode the text return from tigergraph as it was encoded when written into jsonl file for uploading - chunks = chunker.chunk(doc["attributes"]["text"].encode('utf-8').decode('unicode_escape')) + v_id = util.process_id(doc["v_id"]) if v_id != doc["v_id"]: logger.info(f"""Cloning doc/content {doc["v_id"]} -> {v_id}""") await upsert_chan.put((upsert_doc, (conn, v_id, chunker_type, doc["attributes"]["text"]))) + + # Use get_chunker for all types (including images) + # For images, get_chunker returns SingleChunker which preserves [IMAGE_REF:] markers + chunker = ecc_util.get_chunker(chunker_type) + # decode the text return from tigergraph as it was encoded when written into jsonl file for uploading + chunks = chunker.chunk(doc["attributes"]["text"].encode('utf-8').decode('unicode_escape')) - logger.info(f"Chunking {v_id}") + logger.info(f"Chunking {v_id} into {len(chunks)} chunk(s)") for i, chunk in enumerate(chunks): chunk_id = f"{v_id}_chunk_{i}" logger.info(f"Processing chunk {chunk_id}") diff --git a/ecc/app/supportai/workers.py b/ecc/app/supportai/workers.py index ab1c89f..300bee4 100644 --- a/ecc/app/supportai/workers.py +++ b/ecc/app/supportai/workers.py @@ -80,10 +80,15 @@ async def chunk_doc( chunker_type = doc["attributes"]["ctype"].lower().strip() else: chunker_type = "" + + v_id = util.process_id(doc["v_id"]) + + # Use markdown chunker for all documents + # Image descriptions wrapped in headers will naturally become single chunks chunker = ecc_util.get_chunker(chunker_type) chunks = chunker.chunk(doc["attributes"]["text"]) - v_id = util.process_id(doc["v_id"]) - logger.info(f"Chunking {v_id}") + + logger.info(f"Chunking {v_id} into {len(chunks)} chunk(s)") for i, chunk in enumerate(chunks): chunk_id = f"{v_id}_chunk_{i}" # send chunks to be upserted (func, args) diff --git a/graphrag-ui/src/components/CustomChatMessage.tsx b/graphrag-ui/src/components/CustomChatMessage.tsx index 9fa02a0..36d2ebd 100755 --- a/graphrag-ui/src/components/CustomChatMessage.tsx +++ b/graphrag-ui/src/components/CustomChatMessage.tsx @@ -1,4 +1,4 @@ -import { FC, useState } from "react"; +import { FC, useState, useEffect } from "react"; import Markdown from 'react-markdown' import { Dialog, @@ -50,6 +50,69 @@ const getReasoning = (msg) => { return msg.query_sources.reasoning } +// Custom Image component that fetches images with authentication headers +const AuthenticatedImage: FC<{ src: string; alt: string }> = ({ src, alt }) => { + const [imageSrc, setImageSrc] = useState(""); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(false); + + useEffect(() => { + const fetchImage = async () => { + try { + // Get credentials from localStorage (same pattern as Interact.tsx and SideMenu.tsx) + const creds = localStorage.getItem("creds"); + if (!creds) { + setError(true); + setLoading(false); + return; + } + + // Fetch image with authentication header + const response = await fetch(src, { + headers: { + Authorization: `Basic ${creds}`, + }, + }); + + if (!response.ok) { + throw new Error(`Failed to load image: ${response.status}`); + } + + // Convert to blob and create object URL + const blob = await response.blob(); + const objectUrl = URL.createObjectURL(blob); + setImageSrc(objectUrl); + setLoading(false); + } catch (err) { + console.error("Error loading image:", err); + setError(true); + setLoading(false); + } + }; + + if (src) { + fetchImage(); + } + + // Cleanup object URL on unmount + return () => { + if (imageSrc) { + URL.revokeObjectURL(imageSrc); + } + }; + }, [src]); + + if (loading) { + return Loading image...; + } + + if (error || !imageSrc) { + return Failed to load image; + } + + return {alt}; +}; + export const CustomChatMessage: FC = ({ message, }) => { @@ -82,11 +145,24 @@ export const CustomChatMessage: FC = ({ return true; }; + // Custom markdown components to handle images with authentication + const markdownComponents = { + img: ({ src, alt }: { src?: string; alt?: string }) => { + if (!src) return null; + // Check if it's an internal API image that needs authentication + if (src.startsWith('/ui/image_vertex/')) { + return ; + } + // For external images, use regular img tag + return {alt}; + }, + }; + return ( <> {typeof message === "string" ? (
- {message} + {message}
) : message.key === null ? ( message @@ -96,7 +172,7 @@ export const CustomChatMessage: FC = ({ {message.response_type === "progress" ? (

{message.content}

) : ( - {message.content} + {message.content} )} tuple[dict, dict]: @@ -49,6 +49,20 @@ def init_supportai(conn: TigerGraphConnection, graphname: str) -> tuple[dict, di graphname, schema ) ) + + # Add Image vertex schema (for storing images from documents) + if "- VERTEX Image" in current_schema: + schema_res += " Image schema already exists, skipped" + else: + file_path = "common/gsql/supportai/SupportAI_Schema_Images.gsql" + with open(file_path, "r") as f: + image_schema = f.read() + schema_res += " " + schema_res += conn.gsql( + """USE GRAPH {}\n{}\nRUN SCHEMA_CHANGE JOB add_image_support_schema""".format( + graphname, image_schema + ) + ) if "- embedding(Dimension=" in current_schema: schema_res+=" Embeddding schema already exists, skipped" @@ -281,13 +295,21 @@ def create_ingest( conn: TigerGraphConnection, ): # Check for invalid combination of multi format and non-s3 data source - if ingest_config.file_format.lower() == "multi" and ingest_config.data_source.lower() != "s3": + if ingest_config.file_format.lower() == "multi" and ingest_config.data_source.lower() not in ["s3", "server"]: raise Exception( - "AWS Bedrock BDA preprocessing with 'multi' file format is only supported for S3 data sources.") - - if ingest_config.file_format.lower() == "json" or ingest_config.file_format.lower() == "multi": + "Multi-format file processing is only supported for S3 and server data sources.") + + # Choose loading job template based on source/format + if ( + ingest_config.file_format.lower() == "multi" + and ingest_config.data_source.lower() == "server" + ): + # Server multi can include images; use the WithImages loading job + file_path = "common/gsql/supportai/SupportAI_InitialLoadJSON_WithImages.gsql" + elif ingest_config.file_format.lower() == "json" or ingest_config.file_format.lower() == "multi": file_path = "common/gsql/supportai/SupportAI_InitialLoadJSON.gsql" + if ingest_config.file_format.lower() in ["json", "multi"]: with open(file_path) as f: ingest_template = f.read() ingest_template = ingest_template.replace("@uuid@", str(uuid.uuid4().hex)) @@ -414,7 +436,20 @@ def create_ingest( data_stream_conn = data_stream_conn.replace( "@source_config@", json.dumps(connector) ) - elif ingest_config.data_source.lower() == "local" or ingest_config.data_source.lower() == "remote": + elif ingest_config.data_source.lower() == "server": + folder_path = ingest_config.data_source_config.get("folder_path", None) + if folder_path is None: + raise Exception("Folder path not provided for server processing") + if ingest_config.file_format.lower() == "multi": + extractor = TextExtractor() + server_processing_result = extractor.process_folder(folder_path, graphname=graphname) + if server_processing_result.get("statusCode") != 200: + raise Exception(f"Server folder processing failed: {server_processing_result}") + documents = server_processing_result.get("documents", []) + res_ingest_config["server_jobs"] = documents + else: + raise Exception("Server data source supports only 'multi' file_format") + elif ingest_config.data_source.lower() == "remote": pass else: raise Exception("Data source not implemented") @@ -430,8 +465,13 @@ def create_ingest( res["data_path"] = ingest_config.data_source_config.get("output_bucket", "") # key name to be changed res["data_source_id"] = res_ingest_config - elif ingest_config.data_source.lower() == "local": - res["data_source_id"] = "DocumentContent" + elif ingest_config.data_source.lower() == "server" and ingest_config.file_format.lower() == "multi": + # Mirror S3 behavior: attach full ingest config (including server_jobs) to response + res_ingest_config["data_source_id"] = "DocumentContent" + # Use a placeholder path that doesn't start with "/" to avoid pyTigerGraph treating it as a file + # The actual folder path is stored in server_jobs, this is just for the API call + res["data_path"] = "server_multi" + res["data_source_id"] = res_ingest_config else: data_source_created = conn.gsql( "USE GRAPH {}\n".format(graphname) + data_stream_conn @@ -574,5 +614,42 @@ def ingest( "job_name": loader_info.load_job_id, "summary": processed_files } + elif ingest_config.get("data_source") == "server" and ingest_config.get("file_format") == "multi": + try: + processed_files = [] + data_source_id = ingest_config.get("data_source_id", "DocumentContent") + if ingest_config.get("server_jobs"): + for doc_data in ingest_config.get("server_jobs"): + if doc_data.get("image_data"): + payload = { + "doc_id": doc_data.get("doc_id", ""), + "doc_type": "image", + "image_data": doc_data.get("image_data", ""), + "image_format": doc_data.get("image_format", "jpg"), + "parent_doc": doc_data.get("parent_doc", ""), + "page_number": doc_data.get("page_number", 0), + "position": doc_data.get("position", 0), + "content": "" + } + else: + payload = { + "doc_id": doc_data.get("doc_id", ""), + "doc_type": doc_data.get("doc_type", "markdown"), + "content": doc_data.get("content", "") + } + payload_json = json.dumps(payload) + conn.runLoadingJobWithData(payload_json, data_source_id, loader_info.load_job_id) + processed_files.append({ + 'file_path': doc_data.get("doc_id", ""), + 'parent_doc': doc_data.get("parent_doc", ""), + }) + logger.info(f"Data uploading done for doc_id: {doc_data.get('doc_id', 'unknown')}") + except Exception as e: + raise Exception(f"Error during server markdown extraction and TigerGraph loading: {e}") + return { + "job_name": loader_info.load_job_id, + "summary": processed_files + } + else: - raise Exception("Data source and file format combination not implemented") + raise Exception("Data source and file format combination not implemented") \ No newline at end of file diff --git a/graphrag/app/supportai/supportai_ingest.py b/graphrag/app/supportai/supportai_ingest.py index b0d3909..f98c5ad 100644 --- a/graphrag/app/supportai/supportai_ingest.py +++ b/graphrag/app/supportai/supportai_ingest.py @@ -49,6 +49,19 @@ def chunk_document(self, document, chunker, chunker_params): chunker_params.get("breakpoint_threshold_type", "percentile"), chunker_params.get("breakpoint_threshold_amount", 0.95), ) + elif chunker.lower() == "html": + from common.chunkers.html_chunker import HTMLChunker + + chunker = HTMLChunker( + headers=chunker_params.get("headers", None) + ) + elif chunker.lower() == "markdown": + from common.chunkers.markdown_chunker import MarkdownChunker + + chunker = MarkdownChunker( + chunk_size=chunker_params.get("chunk_size", 0), + chunk_overlap=chunker_params.get("overlap_size", 0) + ) else: raise ValueError(f"Chunker {chunker} not supported") @@ -415,6 +428,8 @@ def ingest_blobs(self, doc_source: BatchDocumentIngest): blob_store = AzureBlobStore( doc_source.service_params["azure_connection_string"] ) + elif doc_source.service == "local": + raise ValueError("Local service should use direct file processing, not blob store") else: raise ValueError(f"Service {doc_source.service} not supported")