18. Vision Analysis System Project

Chapter 18 of 18 · 25 min

KEY INSIGHT

Building production vision systems requires integrating image preprocessing, model orchestration, caching layers, evaluation loops, and error handling into a coherent architectureΓÇönot just calling a model. This chapter synthesizes all previous concepts into a complete vision analysis system. The system handles image ingestion, route requests to appropriate models, maintain caches, and provide traceable outputs with confidence scores. ```python import asyncio import hashlib import json import time from dataclasses import dataclass, field from pathlib import Path from typing import Optional import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class AnalysisResult: request_id: str image_path: str model_used: str response: str confidence: float processing_time_ms: float cache_hit: bool = False error: Optional[str] = None class VisionAnalysisPipeline: def __init__( self, client, embedding_cache_size: int = 1000, preprocess_cache_size: int = 500 ): self.client = client self.embedding_cache_size = embedding_cache_size self.preprocess_cache_size = preprocess_cache_size # Caches self._preprocess_cache = {} self._result_cache = {} # Metrics self.cache_hits = 0 self.cache_misses = 0 self.total_requests = 0 def _generate_request_id(self, image_path: str, query: str) -> str: """Create deterministic request ID for caching""" content = f"{image_path}:{query}" return hashlib.sha256(content.encode()).hexdigest()[:16] def _preprocess_image( self, image_path: str, max_size: tuple[int, int] = (1024, 1024) ) -> bytes: """Preprocess image with caching""" cache_key = f"{image_path}:{max_size}" if cache_key in self._preprocess_cache: return self._preprocess_cache[cache_key] from PIL import Image import io img = Image.open(image_path) img.thumbnail(max_size, Image.LANCZOS) buffer = io.BytesIO() img.save(buffer, format="JPEG", quality=90) processed = buffer.getvalue() if len(self._preprocess_cache) >= self.preprocess_cache_size: # Simple eviction: remove oldest oldest_key = next(iter(self._preprocess_cache)) del self._preprocess_cache[oldest_key] self._preprocess_cache[cache_key] = processed return processed async def analyze_image( self, image_path: str, query: str, model: str = "gemini-2.0-flash-thinking", use_cache: bool = True ) -> AnalysisResult: """Main analysis entry point with full pipeline""" start_time = time.perf_counter() request_id = self._generate_request_id(image_path, query) # Check result cache if use_cache and request_id in self._result_cache: result = self._result_cache[request_id] result.cache_hit = True self.cache_hits += 1 logger.info(f"Cache hit for request {request_id}") return result self.cache_misses += 1 self.total_requests += 1 try: # Preprocess image processed = self._preprocess_image(image_path) img_b64 = __import__('base64').b64encode(processed).decode() # Route to appropriate model if "classify" in query.lower(): response_text, confidence = await self._classify_image( img_b64, query, model ) elif "describe" in query.lower(): response_text, confidence = await self._describe_image( img_b64, query, model ) elif "compare" in query.lower(): response_text, confidence = await self._compare_images( image_path, query, model ) else: response_text, confidence = await self._general_analysis( img_b64, query, model ) processing_time = (time.perf_counter() - start_time) * 1000 result = AnalysisResult( request_id=request_id, image_path=image_path, model_used=model, response=response_text, confidence=confidence, processing_time_ms=processing_time, cache_hit=False ) # Cache result if use_cache and len(self._result_cache) < 1000: self._result_cache[request_id] = result return result except Exception as e: logger.error(f"Analysis failed for {image_path}: {e}") processing_time = (time.perf_counter() - start_time) * 1000 return AnalysisResult( request_id=request_id, image_path=image_path, model_used=model, response="", confidence=0.0, processing_time_ms=processing_time, error=str(e) ) async def _classify_image( self, img_b64: str, query: str, model: str ) -> tuple[str, float]: """Classification analysis with confidence estimation""" prompt = f""" Analyze this image for classification purposes. Query: {query} Output structured response with categories and confidence levels. """ response = await self.client.messages.create( model=model, messages=[{ "role": "user", \n "content": [ {"type": "image", "source": {"type": "base64", "data": img_b64}}, {"type": "text", "text": prompt} ] }] ) # Estimate confidence from response language response_text = response.content[0].text confidence = self._estimate_confidence(response_text) return response_text, confidence async def _describe_image( self, img_b64: str, query: str, model: str ) -> tuple[str, float]: """Generate detailed image description""" prompt = f""" Provide a detailed description of this image. User interest: {query} Focus on relevant aspects per the user's interest. Include specific details that could be verified. """ response = await self.client.messages.create( model=model, messages=[{ "role": "user", "content": [ {"type": "image", "source": {"type": "base64", "data": img_b64}}, {"type": "text", "text": prompt} ] }] ) return response.content[0].text, 0.85 async def _general_analysis( self, img_b64: str, query: str, model: str ) -> tuple[str, float]: """General purpose visual question answering""" response = await self.client.messages.create( model=model, messages=[{ "role": "user", "content": [ {"type": "image", "source": {"type": "base64", "data": img_b64}}, {"type": "text", "text": query} ] }] ) response_text = response.content[0].text confidence = self._estimate_confidence(response_text) return response_text, confidence async def _compare_images( self, image_path: str, query: str, model: str ) -> tuple[str, float]: """Compare multiple images if referenced""" # Implementation for multi-image comparison return await self._general_analysis( self._preprocess_image(image_path), query, model ), 0.75 def _estimate_confidence(self, response: str) -> float: """Heuristic confidence estimation from response characteristics""" base_confidence = 0.7 # Hedging language suggests lower confidence hedging_words = ["might", "possibly", "perhaps", "maybe"] for word in hedging_words: if word in response.lower(): base_confidence -= 0.1 # Specific details suggest higher confidence specific_patterns = ["specifically", "exactly", "clearly", "distinctly"] for pattern in specific_patterns: if pattern in response.lower(): base_confidence += 0.05 return max(0.0, min(1.0, base_confidence)) def get_metrics(self) -> dict: """Return pipeline performance metrics""" return { "total_requests": self.total_requests, "cache_hits": self.cache_hits, "cache_misses": self.cache_misses, "cache_hit_rate": ( self.cache_hits / self.total_requests if self.total_requests > 0 else 0.0 ), "preprocess_cache_size": len(self._preprocess_cache), "result_cache_size": len(self._result_cache) } async def batch_analyze( self, image_queries: list[tuple[str, str]], batch_size: int = 10 ) -> list[AnalysisResult]: """Process batch of image analysis requests""" results = [] for i in range(0, len(image_queries), batch_size): batch = image_queries[i:i + batch_size] batch_tasks = [ self.analyze_image(image_path, query) for image_path, query in batch ] batch_results = await asyncio.gather(*batch_tasks) results.extend(batch_results) return results # Example usage async def main(): client = AsyncVertexAI() pipeline = VisionAnalysisPipeline(client) # Single analysis result = await pipeline.analyze_image( image_path="product_image.jpg", query="What are the key features of this product?" ) print(f"Analysis: {result.response}") print(f"Confidence: {result.confidence}") print(f"Processing time: {result.processing_time_ms:.2f}ms") print(f"Cache hit: {result.cache_hit}") # Batch analysis image_queries = [ ("product1.jpg", "List visible features"), ("product2.jpg", "Compare with similar products"), ("product3.jpg", "Identify brand and model"), ] batch_results = await pipeline.batch_analyze(image_queries) for res in batch_results: print(f"{res.request_id}: {res.confidence}") # Print metrics print(pipeline.get_metrics()) class AsyncVertexAI: async def messages(self): class StreamContext: async def __aenter__(self): return self async def __aexit__(self, *args): pass return StreamContext() async def create(self, **kwargs): from dataclasses import dataclass @dataclass class Content: text: str @dataclass class MockResponse: content: list return MockResponse(content=[Content(text="Analysis result")]) if __name__ == "__main__": asyncio.run(main()) ``` **Failure Modes:** - Cache invalidation when images update but same path reused. Include file hash or mtime in cache keys. - Batch queue backlog causing timeouts on large batches. Implement backpressure. - Memory leaks from unbounded caches. Monitor cache sizes and implement eviction.

EXERCISE

Extend this pipeline with:

Persistent disk cache for preprocessed images
Background re-analysis of low-confidence results
Automatic model selection based on query type classification
Structured logging with request tracing for production debugging