Guides

Image Analysis - Advanced

Copy page

Advanced configuration and technical details for the image analysis system.

Architecture

System Components

The image analysis system consists of:

1. Vision Analyzer Service - Core extraction and analysis logic
2. Document Processor - Integration with document processing pipeline
3. Configuration - Vision model settings and fallbacks
4. Storage - Image chunks in Neo4j database

Data Flow

Code
 
Upload API
    ↓
DocumentProcessor._process_document()
    ↓
DoclingConverter.run() → DoclingDocument (with do_picture_description=True)
    ↓
VisionAnalyzer.analyze_all_images()
    ↓
    ├─ extract_images_from_document()
    │   └─ _get_pil_image_from_picture()
    │
    ├─ analyze_image() [for each image]
    │   ├─ analyze_image_with_vision_model() [if configured]
    │   ├─ analyze_image_with_docling() [uses pre-generated description]
    │   └─ fallback metadata
    │
    └─ Create DocumentChunk with image analysis
        ↓
    Embed image description
        ↓
    Store in Neo4j

Note: Docling generates image descriptions during the conversion step when do_picture_description=True is enabled. The analyze_image_with_docling() method retrieves these pre-generated descriptions.

Vision Analyzer Service

Core Methods

Extract Images

Code
 
def extract_images_from_document(
    self, 
    docling_doc: DoclingDocument
) -> list[ExtractedImage]:
    """Extract all images from a DoclingDocument.
    
    Returns:
        List of ExtractedImage objects with:
        - image_id: Unique identifier
        - pil_image: PIL Image object
        - page_number: Page location
        - bbox: Bounding box coordinates
        - caption: Document caption (if any)
        - existing_description: Docling description (if any)
    """

Analyze with Vision Model

Code
 
async def analyze_image_with_vision_model(
    self,
    pil_image: Image.Image,
    prompt: Optional[str] = None
) -> Optional[str]:
    """Analyze image using configured vision model.
    
    Process:
    1. Convert PIL image to base64 data URL
    2. Prepare OpenAI-compatible request
    3. Call vision API
    4. Return description or None on failure
    
    Timeout: 60 seconds
    """

Smart Analysis

Code
 
async def analyze_image(
    self,
    extracted_image: ExtractedImage,
    force_vision_model: bool = False,
    custom_prompt: Optional[str] = None
) -> ImageAnalysisResult:
    """Analyze with automatic method selection.
    
    Priority:
    1. Vision model (if configured and available)
    2. Docling description (pre-generated during conversion)
    3. Basic metadata (fallback)
    
    Returns:
        ImageAnalysisResult with description, method, and metadata
    """

Docling Fallback

Code
 
def analyze_image_with_docling(
    self, 
    extracted_image: ExtractedImage
) -> Optional[str]:
    """Return Docling's pre-generated image description from conversion.
    
    NOTE: This requires do_picture_description=True in the Docling pipeline
    options. If not enabled during conversion, this will always return None.
    
    The description is generated by Docling's built-in vision model (SmolDocling)
    during the document conversion step, not on-demand.
    """

Image Extraction Details

PIL Image Extraction

Code
 
def _get_pil_image_from_picture(
    picture: PictureItem, 
    doc: DoclingDocument
) -> Optional[Image.Image]:
    """Extract PIL image from Docling PictureItem.
    
    Methods:
    1. picture.get_image(doc) - Direct extraction
    2. Base64 data URL decoding
    3. File path loading
    
    Handles:
    - Embedded base64 images
    - File references
    - Multiple image formats
    """

Supported Image Sources

• Embedded base64: data:image/png;base64,...
• File references: file:///path/to/image.png
• Direct extraction: Via Docling's get_image() method

Base64 Encoding

Code
 
def _pil_to_data_url(pil_image: Image.Image) -> str:
    """Convert PIL image to data URL.
    
    - PNG format for RGBA images
    - JPEG format for RGB images
    - Automatic mode conversion
    
    Returns: data:image/{type};base64,{data}
    """

Vision API Integration

Request Format

Code
 
{
  "model": "gpt-4o",
  "messages": [
    {
      "role": "user",
      "content": [
        {
          "type": "text",
          "text": "Analyze this image..."
        },
        {
          "type": "image_url",
          "image_url": {
            "url": "data:image/jpeg;base64,..."
          }
        }
      ]
    }
  ],
  "max_tokens": 1000
}

Response Handling

Code
 
# Success
{
    "choices": [{
        "message": {
            "content": "This image shows..."
        }
    }]
}

# Error handling
if response.status_code != 200:
    logger.error(f"Vision API error: {response.status_code}")
    return None

Timeout Configuration

Code
 
async with httpx.AsyncClient(timeout=60.0) as client:
    response = await client.post(...)

Adjust timeout based on:

• Image size
• Vision model speed
• Network conditions

Custom Analysis Prompts

Default Prompt

Code
 
DEFAULT_PROMPT = """
Analyze this image in detail. Describe what you see, including:
1. Main objects, people, or elements visible
2. Text visible in the image (if any)
3. Charts, diagrams, or data visualizations (if any)
4. Overall context and purpose of the image
5. Any relevant details that would help understand the document

Provide a comprehensive description suitable for document retrieval and understanding.
"""

Custom Prompts by Use Case

Business Charts

Code
 
CHART_PROMPT = """
Analyze this business chart. Provide:
1. Chart type (bar, line, pie, scatter, etc.)
2. Title and axis labels
3. Key data trends and patterns
4. Notable data points or anomalies
5. Overall business insights

Format for easy text search and retrieval.
"""

Technical Diagrams

Code
 
DIAGRAM_PROMPT = """
Analyze this technical diagram. Include:
1. Diagram type (flowchart, architecture, schema, etc.)
2. Components and their relationships
3. Data flow or process steps
4. Technical specifications visible
5. Purpose and context

Describe for technical documentation retrieval.
"""

Text-Heavy Images

Code
 
TEXT_PROMPT = """
Extract and analyze text from this image:
1. All visible text (OCR)
2. Document type and format
3. Layout and structure
4. Key information and data
5. Context and purpose

Optimize for text search and citation.
"""

Implementing Custom Prompts

Code
 
# In your code
result = await analyzer.analyze_image(
    extracted_image,
    custom_prompt=CHART_PROMPT
)

Storage Integration

Image Chunk Structure

Code
 
image_chunk = DocumentChunk(
    id=f"{doc_id}_image_{idx}",
    document_id=doc_id,
    content=f"[Image Analysis]\n{analysis.description}",
    embedding=embedded_vector,
    chunk_index=1000 + idx,  # Separate from text chunks
    metadata={
        "type": "image_analysis",
        "image_id": analysis.image_id,
        "analysis_method": analysis.analysis_method,
        "page_number": extracted_image.page_number,
        "bbox": extracted_image.bbox,
        "caption": extracted_image.caption
    }
)

Chunk Indexing

• Text chunks: 0 to N
• Image chunks: 1000 to 1000+M
• Prevents overlap between text and image chunks

Neo4j Storage

Code
 
CREATE (c:Chunk {
    id: $chunk_id,
    document_id: $doc_id,
    content: $content,
    embedding: $embedding,
    chunk_index: $index,
    metadata: $metadata
})

Performance Optimization

Image Preprocessing

Code
 
def preprocess_image(pil_image: Image.Image) -> Image.Image:
    """Optimize image before vision API call."""
    
    # Resize large images
    max_size = (1024, 1024)
    if pil_image.size > max_size:
        pil_image.thumbnail(max_size, Image.Resampling.LANCZOS)
    
    # Convert mode for compatibility
    if pil_image.mode == "RGBA":
        pil_image = pil_image.convert("RGB")
    
    return pil_image

Batch Processing

Code
 
# Images are processed concurrently, controlled by VISION_MAX_CONCURRENT (default 2).
# The built-in analyze_all_images() uses asyncio.gather with a global semaphore.
# Example of the same pattern for custom usage:
async def analyze_images_batch(
    images: list[ExtractedImage],
    max_concurrent: int = 2
) -> list[ImageAnalysisResult]:
    """Process images with controlled concurrency."""

    semaphore = asyncio.Semaphore(max_concurrent)

    async def analyze_with_limit(img):
        async with semaphore:
            return await analyzer.analyze_image(img)

    return await asyncio.gather(*[
        analyze_with_limit(img) for img in images
    ])

Caching (Future Enhancement)

Code
 
import hashlib
from functools import lru_cache

@lru_cache(maxsize=100)
def cached_vision_analysis(image_hash: str) -> str:
    """Cache vision model responses."""
    # Implementation for caching
    pass

def get_image_hash(pil_image: Image.Image) -> str:
    """Generate hash for image."""
    buffer = io.BytesIO()
    pil_image.save(buffer, format='PNG')
    return hashlib.sha256(buffer.getvalue()).hexdigest()

Monitoring and Debugging

Logging

Code
 
# Enable detailed logging
import logging

logging.getLogger('app.services.vision_analyzer').setLevel(logging.DEBUG)

# Log messages include:
# - Image extraction success/failure
# - Vision API call status
# - Analysis method selected
# - Processing time per image

Performance Metrics

Code
 
import time

async def analyze_with_metrics(
    extracted_image: ExtractedImage
) -> ImageAnalysisResult:
    start = time.time()
    
    result = await analyzer.analyze_image(extracted_image)
    
    duration = time.time() - start
    logger.info(
        f"Image analysis: {result.image_id} "
        f"method={result.analysis_method} "
        f"duration={duration:.2f}s"
    )
    
    return result

Error Tracking

Code
 
from sentry_sdk import capture_exception

try:
    result = await analyzer.analyze_image(img)
except Exception as e:
    logger.error(f"Image analysis failed: {e}")
    capture_exception(e)
    # Fall back to basic metadata
    result = create_fallback_result(img)

Testing

Unit Tests

Code
 
import pytest
from PIL import Image

@pytest.mark.asyncio
async def test_analyze_image_with_vision_model():
    analyzer = VisionAnalyzer()
    
    # Create test image
    test_image = Image.new('RGB', (100, 100), color='red')
    
    result = await analyzer.analyze_image_with_vision_model(
        test_image,
        prompt="What color is this image?"
    )
    
    assert result is not None
    assert "red" in result.lower()

Integration Tests

Code
 
@pytest.mark.asyncio
async def test_document_processing_with_images():
    processor = DocumentProcessor()
    
    # Process document with images
    doc_id = await processor.process_file(
        file_path="test_document.pdf",
        filename="test.pdf",
        file_size=1024
    )
    
    # Verify image chunks created
    chunks = neo4j.get_document_chunks(doc_id)
    image_chunks = [c for c in chunks if c.metadata.get('type') == 'image_analysis']
    
    assert len(image_chunks) > 0

Extending the System

Custom Analyzers

Code
 
class CustomVisionAnalyzer(VisionAnalyzer):
    """Custom analyzer with specialized processing."""
    
    async def analyze_image(
        self,
        extracted_image: ExtractedImage,
        **kwargs
    ) -> ImageAnalysisResult:
        # Custom preprocessing
        preprocessed = self.preprocess(extracted_image.pil_image)
        
        # Custom analysis logic
        if self.is_chart(preprocessed):
            description = await self.analyze_chart(preprocessed)
        else:
            description = await super().analyze_image(
                extracted_image,
                **kwargs
            )
        
        return ImageAnalysisResult(
            image_id=extracted_image.image_id,
            description=description,
            analysis_method="custom"
        )

Multiple Vision Models

Code
 
class MultiModelAnalyzer(VisionAnalyzer):
    """Use different models for different image types."""
    
    async def analyze_image(
        self,
        extracted_image: ExtractedImage,
        **kwargs
    ) -> ImageAnalysisResult:
        # Detect image type
        image_type = self.classify_image(extracted_image.pil_image)
        
        # Select appropriate model
        if image_type == "chart":
            model = "gpt-4o"
        elif image_type == "text":
            model = "claude-3-5-sonnet"
        else:
            model = "llava"
        
        # Use selected model
        return await self.analyze_with_model(
            extracted_image,
            model=model,
            **kwargs
        )

Best Practices

Production Deployment

1. Rate Limiting: Implement API rate limits
2. Error Handling: Graceful fallbacks for failures
3. Monitoring: Track API usage and costs
4. Caching: Cache frequent analyses
5. Queue Processing: Process images asynchronously

Cost Management

1. Use Docling for simple images
2. Select cost-effective models for scale
3. Implement image deduplication
4. Monitor API usage per document
5. Set spending alerts

Quality Assurance

1. Test with diverse document types
2. Validate extraction accuracy
3. Review analysis quality regularly
4. Collect user feedback
5. Fine-tune prompts as needed

Related Resources

• Image Analysis Guide - Basic usage and configuration
• Document Processing - Processing pipeline details
• API Reference - Upload and query endpoints