8. Project: Use Case Selector

Description

In this project you build a Use Case Selector: an intelligent router that receives an input (PDF, image, audio, video, text), automatically detects its type, and applies the correct pipeline. It's the "brain" that connects all the design patterns you learned in this module into a single system.

Why it matters: In a real multimodal system, the user doesn't say "apply the Document Q&A pipeline". They upload a file and expect the system to do the right thing. The Use Case Selector is that layer of intelligence: it receives any input, decides what to do with it, and runs the appropriate pipeline. It's the most critical component of the Document Analyzer you'll build in Module 8.

What you're going to build:

┌──────────────┐     ┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│  Input       │────▶│  Type        │────▶│  Pipeline    │────▶│  Router      │
│  (any)       │     │  detector    │     │  Registry    │     │  (execute)   │
└──────────────┘     └──────────────┘     └──────────────┘     └──────┬───────┘
                                                                       │
                                                                       ▼
                                                               ┌──────────────┐
                                                               │  Result      │
                                                               │  + tracking  │
                                                               └──────────────┘

Specifications

Input

• Path to a file — PDF, image, audio, video, text
• Optional question — If the user wants Q&A about the file
• Configuration — Preferred model, maximum budget, desired quality

Output

• Detected type — What type of input it is
• Selected pipeline — Which pipeline was applied
• Result — Output of the executed pipeline
• Metadata — Cost, duration, model used

Components

Step 1: Configuration

from dataclasses import dataclass, field

@dataclass
class PipelineConfig:
    model: str = "gpt-4o-mini"
    max_tokens: int = 500
    temperature: float = 0.2
    max_cost_per_request: float = 0.50
    enable_cache: bool = True
    enable_logging: bool = True

@dataclass
class UseCaseSelectorConfig:
    default_pipeline_config: PipelineConfig = field(default_factory=PipelineConfig)
    supported_types: set = field(default_factory=lambda: {
        "document", "image", "audio", "video", "text"
    })
    max_file_size_mb: float = 100.0
    fallback_enabled: bool = True

    type_extensions: dict = field(default_factory=lambda: {
        ".pdf": "document",
        ".doc": "document",
        ".docx": "document",
        ".png": "image",
        ".jpg": "image",
        ".jpeg": "image",
        ".gif": "image",
        ".webp": "image",
        ".bmp": "image",
        ".mp3": "audio",
        ".wav": "audio",
        ".m4a": "audio",
        ".ogg": "audio",
        ".flac": "audio",
        ".mp4": "video",
        ".avi": "video",
        ".mov": "video",
        ".mkv": "video",
        ".webm": "video",
        ".txt": "text",
        ".md": "text",
        ".csv": "text",
        ".json": "text",
    })

Step 2: Type Detector

from pathlib import Path
import mimetypes

class TypeDetector:
    def __init__(self, config: UseCaseSelectorConfig):
        self.config = config

    def detect(self, path: str) -> dict:
        p = Path(path)

        if not p.exists():
            return {
                "type": "unknown",
                "error": "file_not_found",
                "path": path
            }

        ext = p.suffix.lower()
        size_mb = p.stat().st_size / (1024 * 1024)

        if size_mb > self.config.max_file_size_mb:
            return {
                "type": "unknown",
                "error": "file_too_large",
                "size_mb": round(size_mb, 2),
                "max_mb": self.config.max_file_size_mb,
                "path": path
            }

        detected_type = self.config.type_extensions.get(ext)

        if not detected_type:
            mime_type, _ = mimetypes.guess_type(path)
            if mime_type:
                detected_type = self._mime_to_type(mime_type)

        if not detected_type:
            detected_type = "unknown"

        return {
            "type": detected_type,
            "extension": ext,
            "size_mb": round(size_mb, 2),
            "path": path,
            "mime_type": mimetypes.guess_type(path)[0]
        }

    def _mime_to_type(self, mime: str) -> str | None:
        if mime.startswith("image/"):
            return "image"
        if mime.startswith("audio/"):
            return "audio"
        if mime.startswith("video/"):
            return "video"
        if mime.startswith("text/"):
            return "text"
        if mime == "application/pdf":
            return "document"
        return None

    def detect_multiple(self, paths: list[str]) -> list[dict]:
        return [self.detect(path) for path in paths]

    def validate(self, path: str) -> dict:
        detection = self.detect(path)

        if detection["type"] == "unknown":
            return {
                "valid": False,
                "detection": detection,
                "message": detection.get("error", "Unsupported type")
            }

        if detection["type"] not in self.config.supported_types:
            return {
                "valid": False,
                "detection": detection,
                "message": f"Type '{detection['type']}' is not enabled"
            }

        return {
            "valid": True,
            "detection": detection
        }

Step 3: Pipeline Registry

from typing import Callable, Any

class Pipeline:
    def __init__(
        self,
        name: str,
        input_type: str,
        handler: Callable,
        description: str = "",
        supports_question: bool = False,
        estimated_cost_per_call: float = 0.01
    ):
        self.name = name
        self.input_type = input_type
        self.handler = handler
        self.description = description
        self.supports_question = supports_question
        self.estimated_cost_per_call = estimated_cost_per_call


class PipelineRegistry:
    def __init__(self):
        self.pipelines: dict[str, list[Pipeline]] = {}

    def register(self, pipeline: Pipeline) -> None:
        if pipeline.input_type not in self.pipelines:
            self.pipelines[pipeline.input_type] = []
        self.pipelines[pipeline.input_type].append(pipeline)

    def get_pipeline(self, input_type: str, has_question: bool = False) -> Pipeline | None:
        candidates = self.pipelines.get(input_type, [])

        if not candidates:
            return None

        if has_question:
            qa_pipelines = [p for p in candidates if p.supports_question]
            if qa_pipelines:
                return qa_pipelines[0]

        return candidates[0]

    def list_pipelines(self) -> dict:
        result = {}
        for input_type, pipelines in self.pipelines.items():
            result[input_type] = [
                {
                    "name": p.name,
                    "description": p.description,
                    "supports_question": p.supports_question,
                    "estimated_cost": p.estimated_cost_per_call
                }
                for p in pipelines
            ]
        return result

    def get_all_supported_types(self) -> set[str]:
        return set(self.pipelines.keys())

Register pipelines

from openai import OpenAI
import base64
import fitz

client = OpenAI()

def document_extract_handler(path: str, question: str = None, config: PipelineConfig = None) -> dict:
    config = config or PipelineConfig()
    doc = fitz.open(path)
    pages = []
    for i in range(len(doc)):
        text = doc[i].get_text().strip()
        if text:
            pages.append({"page": i + 1, "text": text})
    doc.close()

    full_text = "\n\n".join(p["text"] for p in pages)

    if question:
        response = client.chat.completions.create(
            model=config.model,
            messages=[
                {
                    "role": "system",
                    "content": "Answer ONLY based on the context. If you can't find the answer, say so. Cite the page."
                },
                {
                    "role": "user",
                    "content": f"Document:\n{full_text[:8000]}\n\nQuestion: {question}"
                }
            ],
            max_tokens=config.max_tokens,
            temperature=config.temperature
        )
        return {
            "type": "document_qa",
            "answer": response.choices[0].message.content,
            "pages_processed": len(pages),
            "model": config.model
        }

    return {
        "type": "document_extraction",
        "pages": pages,
        "total_pages": len(pages),
        "total_characters": len(full_text)
    }


def image_analyze_handler(path: str, question: str = None, config: PipelineConfig = None) -> dict:
    config = config or PipelineConfig()
    with open(path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()

    prompt = question or "Analyze this image. Describe what it contains, extract any visible text, and classify the content type."

    response = client.chat.completions.create(
        model=config.model,
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": prompt},
                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
            ]
        }],
        max_tokens=config.max_tokens,
        temperature=config.temperature
    )

    return {
        "type": "image_analysis",
        "analysis": response.choices[0].message.content,
        "model": config.model
    }


def audio_transcribe_handler(path: str, question: str = None, config: PipelineConfig = None) -> dict:
    config = config or PipelineConfig()
    with open(path, "rb") as f:
        transcript = client.audio.transcriptions.create(
            model="whisper-1",
            file=f,
            response_format="text"
        )

    if question:
        response = client.chat.completions.create(
            model=config.model,
            messages=[
                {
                    "role": "system",
                    "content": "Answer based on the audio transcript."
                },
                {
                    "role": "user",
                    "content": f"Transcript:\n{transcript[:6000]}\n\nQuestion: {question}"
                }
            ],
            max_tokens=config.max_tokens,
            temperature=config.temperature
        )
        return {
            "type": "audio_qa",
            "answer": response.choices[0].message.content,
            "transcript": transcript,
            "model": config.model
        }

    return {
        "type": "audio_transcription",
        "transcript": transcript,
        "transcript_length": len(transcript)
    }


def video_analyze_handler(path: str, question: str = None, config: PipelineConfig = None) -> dict:
    config = config or PipelineConfig()
    import cv2
    import os

    cap = cv2.VideoCapture(path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    duration = total_frames / fps if fps > 0 else 0

    target_frames = min(15, max(5, int(duration / 10)))
    interval = max(1, int(total_frames / target_frames))

    os.makedirs("/tmp/uc_selector_frames", exist_ok=True)
    descriptions = []
    frame_id = 0
    captured = 0

    while True:
        ret, frame = cap.read()
        if not ret:
            break
        if frame_id % interval == 0 and captured < target_frames:
            frame_path = f"/tmp/uc_selector_frames/frame_{frame_id}.jpg"
            cv2.imwrite(frame_path, frame)

            with open(frame_path, "rb") as f:
                b64 = base64.b64encode(f.read()).decode()

            ts = frame_id / fps
            minutes = int(ts // 60)
            seconds = int(ts % 60)

            response = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{
                    "role": "user",
                    "content": [
                        {"type": "text", "text": "Describe this scene in 1-2 sentences."},
                        {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
                    ]
                }],
                max_tokens=100
            )
            descriptions.append(f"{minutes}:{seconds:02d} - {response.choices[0].message.content}")
            captured += 1

        frame_id += 1

    cap.release()

    timeline = "\n".join(descriptions)

    summary_prompt = f"Video frame analysis:\n\n{timeline}\n\n"
    if question:
        summary_prompt += f"Question: {question}"
    else:
        summary_prompt += "Generate a summary of the video with the key points."

    response = client.chat.completions.create(
        model=config.model,
        messages=[{"role": "user", "content": summary_prompt}],
        max_tokens=config.max_tokens
    )

    return {
        "type": "video_analysis",
        "summary": response.choices[0].message.content,
        "frames_analyzed": captured,
        "duration_seconds": round(duration, 2),
        "model": config.model
    }


def text_analyze_handler(path: str, question: str = None, config: PipelineConfig = None) -> dict:
    config = config or PipelineConfig()
    with open(path, "r", encoding="utf-8") as f:
        text = f.read()

    prompt = question or "Analyze this text. Summarize the key points and extract relevant information."

    response = client.chat.completions.create(
        model=config.model,
        messages=[
            {
                "role": "user",
                "content": f"Text:\n{text[:8000]}\n\n{prompt}"
            }
        ],
        max_tokens=config.max_tokens,
        temperature=config.temperature
    )

    return {
        "type": "text_analysis",
        "result": response.choices[0].message.content,
        "text_length": len(text),
        "model": config.model
    }


def create_default_registry() -> PipelineRegistry:
    registry = PipelineRegistry()

    registry.register(Pipeline(
        name="document_extractor",
        input_type="document",
        handler=document_extract_handler,
        description="Extracts text from PDFs and answers questions about documents",
        supports_question=True,
        estimated_cost_per_call=0.01
    ))

    registry.register(Pipeline(
        name="image_analyzer",
        input_type="image",
        handler=image_analyze_handler,
        description="Analyzes images: describes content, extracts text, classifies",
        supports_question=True,
        estimated_cost_per_call=0.005
    ))

    registry.register(Pipeline(
        name="audio_transcriber",
        input_type="audio",
        handler=audio_transcribe_handler,
        description="Transcribes audio and answers questions about the content",
        supports_question=True,
        estimated_cost_per_call=0.02
    ))

    registry.register(Pipeline(
        name="video_analyzer",
        input_type="video",
        handler=video_analyze_handler,
        description="Analyzes video by extracting frames and generating a summary",
        supports_question=True,
        estimated_cost_per_call=0.10
    ))

    registry.register(Pipeline(
        name="text_analyzer",
        input_type="text",
        handler=text_analyze_handler,
        description="Analyzes text files and answers questions",
        supports_question=True,
        estimated_cost_per_call=0.005
    ))

    return registry

Step 4: Router

import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("use_case_selector")

class UseCaseRouter:
    def __init__(
        self,
        config: UseCaseSelectorConfig = None,
        registry: PipelineRegistry = None,
        cost_tracker = None
    ):
        self.config = config or UseCaseSelectorConfig()
        self.detector = TypeDetector(self.config)
        self.registry = registry or create_default_registry()
        self.cost_tracker = cost_tracker
        self.history: list[dict] = []

    def route(self, path: str, question: str = None) -> dict:
        start_time = time.time()

        validation = self.detector.validate(path)
        if not validation["valid"]:
            return {
                "status": "error",
                "error": validation["message"],
                "detection": validation["detection"]
            }

        detection = validation["detection"]
        input_type = detection["type"]
        has_question = question is not None and question.strip() != ""

        pipeline = self.registry.get_pipeline(input_type, has_question)
        if not pipeline:
            return {
                "status": "error",
                "error": f"No pipeline registered for type '{input_type}'",
                "detection": detection
            }

        logger.info(f"Routing: {path} → {pipeline.name} (type={input_type}, question={has_question})")

        try:
            result = pipeline.handler(
                path,
                question=question,
                config=self.config.default_pipeline_config
            )

            duration_ms = round((time.time() - start_time) * 1000, 2)

            entry = {
                "status": "success",
                "detection": detection,
                "pipeline": pipeline.name,
                "result": result,
                "duration_ms": duration_ms,
                "estimated_cost": pipeline.estimated_cost_per_call,
                "question": question
            }

            if self.cost_tracker:
                self.cost_tracker.track_chat(
                    self.config.default_pipeline_config.model,
                    input_tokens=1000,
                    output_tokens=500
                )

            self.history.append(entry)
            logger.info(f"Success: {pipeline.name} in {duration_ms}ms")

            return entry

        except Exception as e:
            duration_ms = round((time.time() - start_time) * 1000, 2)

            error_entry = {
                "status": "error",
                "detection": detection,
                "pipeline": pipeline.name,
                "error": str(e),
                "error_type": type(e).__name__,
                "duration_ms": duration_ms
            }

            self.history.append(error_entry)
            logger.error(f"Error in {pipeline.name}: {e}")

            if self.config.fallback_enabled:
                return self._try_fallback(path, question, error_entry)

            return error_entry

    def _try_fallback(self, path: str, question: str, original_error: dict) -> dict:
        logger.info("Attempting fallback...")

        try:
            with open(path, "rb") as f:
                content = f.read()

            if len(content) < 10000:
                text_content = content.decode("utf-8", errors="replace")
                prompt = f"Analyze this content:\n{text_content[:6000]}"
                if question:
                    prompt += f"\n\nQuestion: {question}"

                response = client.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=[{"role": "user", "content": prompt}],
                    max_tokens=500
                )

                return {
                    "status": "fallback",
                    "original_error": original_error,
                    "result": {
                        "type": "fallback_analysis",
                        "analysis": response.choices[0].message.content
                    }
                }
        except Exception:
            pass

        return original_error

    def route_multiple(self, paths: list[str], question: str = None) -> list[dict]:
        return [self.route(path, question) for path in paths]

    def get_stats(self) -> dict:
        total = len(self.history)
        successes = sum(1 for h in self.history if h["status"] == "success")
        errors = sum(1 for h in self.history if h["status"] == "error")
        fallbacks = sum(1 for h in self.history if h["status"] == "fallback")

        durations = [h["duration_ms"] for h in self.history if "duration_ms" in h]
        total_cost = sum(h.get("estimated_cost", 0) for h in self.history)

        pipeline_counts = {}
        for h in self.history:
            p = h.get("pipeline", "unknown")
            pipeline_counts[p] = pipeline_counts.get(p, 0) + 1

        return {
            "total_requests": total,
            "successes": successes,
            "errors": errors,
            "fallbacks": fallbacks,
            "success_rate": round(successes / total * 100, 1) if total > 0 else 0,
            "avg_duration_ms": round(sum(durations) / len(durations), 2) if durations else 0,
            "total_estimated_cost": round(total_cost, 4),
            "by_pipeline": pipeline_counts
        }

Step 5: Integrated Cost Tracker

class UseCaseCostTracker:
    def __init__(self):
        self.records: list[dict] = []

    def track(self, pipeline_name: str, input_type: str, cost: float, duration_ms: float):
        self.records.append({
            "pipeline": pipeline_name,
            "input_type": input_type,
            "cost": cost,
            "duration_ms": duration_ms,
            "timestamp": time.time()
        })

    def summary(self) -> dict:
        total_cost = sum(r["cost"] for r in self.records)
        by_pipeline = {}
        for r in self.records:
            p = r["pipeline"]
            if p not in by_pipeline:
                by_pipeline[p] = {"cost": 0, "count": 0}
            by_pipeline[p]["cost"] += r["cost"]
            by_pipeline[p]["count"] += 1

        return {
            "total_cost": round(total_cost, 4),
            "total_requests": len(self.records),
            "by_pipeline": {
                k: {"cost": round(v["cost"], 4), "count": v["count"]}
                for k, v in by_pipeline.items()
            }
        }

    def check_budget(self, max_daily_cost: float) -> dict:
        today_records = [
            r for r in self.records
            if r["timestamp"] > time.time() - 86400
        ]
        today_cost = sum(r["cost"] for r in today_records)

        return {
            "today_cost": round(today_cost, 4),
            "daily_budget": max_daily_cost,
            "remaining": round(max_daily_cost - today_cost, 4),
            "within_budget": today_cost < max_daily_cost,
            "usage_percent": round(today_cost / max_daily_cost * 100, 1) if max_daily_cost > 0 else 0
        }

Step 6: Main — Complete Orchestration

class UseCaseSelector:
    def __init__(self, config: UseCaseSelectorConfig = None):
        self.config = config or UseCaseSelectorConfig()
        self.registry = create_default_registry()
        self.cost_tracker = UseCaseCostTracker()
        self.router = UseCaseRouter(
            config=self.config,
            registry=self.registry,
            cost_tracker=self.cost_tracker
        )

    def process(self, path: str, question: str = None) -> dict:
        return self.router.route(path, question)

    def process_batch(self, paths: list[str], question: str = None) -> list[dict]:
        return self.router.route_multiple(paths, question)

    def info(self) -> dict:
        return {
            "supported_types": list(self.config.supported_types),
            "available_pipelines": self.registry.list_pipelines(),
            "config": {
                "model": self.config.default_pipeline_config.model,
                "max_file_size_mb": self.config.max_file_size_mb,
                "fallback_enabled": self.config.fallback_enabled,
                "cache_enabled": self.config.default_pipeline_config.enable_cache
            }
        }

    def stats(self) -> dict:
        return {
            "routing": self.router.get_stats(),
            "costs": self.cost_tracker.summary()
        }

    def add_pipeline(self, pipeline: Pipeline) -> None:
        self.registry.register(pipeline)

Complete usage

selector = UseCaseSelector()

print(selector.info())

result = selector.process("contract.pdf")
print(f"Type: {result['detection']['type']}")
print(f"Pipeline: {result['pipeline']}")
print(f"Result: {result['result']}")

result = selector.process(
    "contract.pdf",
    question="What is the penalty clause?"
)
print(f"Answer: {result['result']['answer']}")

result = selector.process("product.jpg")
print(f"Analysis: {result['result']['analysis']}")

result = selector.process("meeting.mp3")
print(f"Transcript: {result['result']['transcript'][:200]}")

result = selector.process("class.mp4")
print(f"Summary: {result['result']['summary']}")

print(selector.stats())

Extension 1: Auto-Pipeline Selection with an LLM

Instead of only using the file extension, use an LLM to decide which pipeline to apply based on the content:

def smart_pipeline_selection(
    path: str,
    question: str,
    detection: dict,
    available_pipelines: dict
) -> str:
    pipelines_desc = "\n".join(
        f"- {p['name']}: {p['description']}"
        for type_pipelines in available_pipelines.values()
        for p in type_pipelines
    )

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": (
                f"File: {path}\n"
                f"Detected type: {detection['type']}\n"
                f"Extension: {detection['extension']}\n"
                f"Size: {detection['size_mb']}MB\n"
                f"User's question: {question or 'None'}\n\n"
                f"Available pipelines:\n{pipelines_desc}\n\n"
                "Which pipeline is the most appropriate? Reply ONLY with the pipeline name."
            )
        }],
        max_tokens=30,
        temperature=0
    )
    return response.choices[0].message.content.strip()

Extension 2: Batch Routing with Priorities

To process multiple files with different priorities:

from dataclasses import dataclass
from typing import Optional

@dataclass
class RoutingRequest:
    path: str
    question: Optional[str] = None
    priority: int = 5

class BatchRouter:
    def __init__(self, selector: UseCaseSelector):
        self.selector = selector
        self.queue: list[RoutingRequest] = []

    def add(self, path: str, question: str = None, priority: int = 5):
        self.queue.append(RoutingRequest(path=path, question=question, priority=priority))

    def process_all(self) -> list[dict]:
        sorted_queue = sorted(self.queue, key=lambda r: r.priority)

        results = []
        for request in sorted_queue:
            result = self.selector.process(request.path, request.question)
            result["priority"] = request.priority
            results.append(result)

        self.queue.clear()
        return results

    def process_by_type(self) -> dict[str, list[dict]]:
        by_type: dict[str, list[RoutingRequest]] = {}
        for req in self.queue:
            detection = self.selector.router.detector.detect(req.path)
            input_type = detection["type"]
            if input_type not in by_type:
                by_type[input_type] = []
            by_type[input_type].append(req)

        results = {}
        for input_type, requests in by_type.items():
            results[input_type] = []
            for req in sorted(requests, key=lambda r: r.priority):
                result = self.selector.process(req.path, req.question)
                results[input_type].append(result)

        self.queue.clear()
        return results

Usage:

batch = BatchRouter(selector)

batch.add("urgent.pdf", priority=1)
batch.add("product1.jpg", priority=5)
batch.add("product2.jpg", priority=5)
batch.add("meeting.mp3", question="What decisions were made?", priority=3)

results = batch.process_all()
for r in results:
    print(f"[P{r['priority']}] {r['pipeline']}: {r['status']}")

Troubleshooting

Problem 1: File type not detected

Symptom: TypeDetector.detect returns "unknown" for a valid file.

Solution: Add the extension to the config:

config = UseCaseSelectorConfig()
config.type_extensions[".xlsx"] = "document"
config.type_extensions[".pptx"] = "document"

Problem 2: Wrong pipeline selected

Symptom: A scanned PDF is routed to document_extractor but there's no text to extract.

Solution: Add content detection after the type:

def detect_pdf_subtype(path: str) -> str:
    import fitz
    doc = fitz.open(path)
    total_text = sum(len(doc[i].get_text().strip()) for i in range(len(doc)))
    total_images = sum(len(doc[i].get_images()) for i in range(len(doc)))
    doc.close()

    if total_text < 100 and total_images > 0:
        return "scanned_document"
    return "text_document"

Problem 3: Long videos consume too much

Symptom: A 2-hour video generates 100+ calls to the Vision API.

Solution: The video_analyze_handler already limits frames with target_frames = min(15, ...). For more control:

config = PipelineConfig()
config.max_tokens = 300

selector = UseCaseSelector(UseCaseSelectorConfig(
    default_pipeline_config=config,
    max_file_size_mb=50.0
))

Problem 4: Accumulated cost without control

Symptom: The cost tracker shows high spending but there are no alerts.

Solution:

budget = selector.cost_tracker.check_budget(max_daily_cost=5.0)
if not budget["within_budget"]:
    print(f"ALERT: Budget exceeded. Spent: ${budget['today_cost']}")

Completeness Checklist

Core functionality

• Detects PDF and routes to the document pipeline
• Detects image and routes to the image pipeline
• Detects audio and routes to the audio pipeline
• Detects video and routes to the video pipeline
• Detects text and routes to the text pipeline
• Handles unsupported files with a clear error
• Supports an optional question for Q&A

Robustness

• Validates that the file exists
• Checks the maximum size
• Handles API errors with fallback
• Logs a history of operations
• Tracks estimated costs

Extensibility

• New pipelines can be registered
• The configuration is flexible
• Supports batch processing

Exercises

Exercise 1: Custom pipeline

Add a pipeline for CSV files that reads the file, analyzes the columns with an LLM, and generates a statistical summary.

import csv

def csv_analyze_handler(path: str, question: str = None, config: PipelineConfig = None) -> dict:
    config = config or PipelineConfig()

    with open(path, "r", encoding="utf-8") as f:
        reader = csv.reader(f)
        headers = next(reader)
        rows = list(reader)

    sample = rows[:20]
    sample_text = "\n".join([",".join(headers)] + [",".join(row) for row in sample])

    prompt = (
        f"CSV file with {len(rows)} rows and {len(headers)} columns.\n"
        f"Columns: {', '.join(headers)}\n\n"
        f"Data sample:\n{sample_text}\n\n"
    )

    if question:
        prompt += f"Question: {question}"
    else:
        prompt += "Analyze this data: describe the columns, identify patterns, and generate a summary."

    response = client.chat.completions.create(
        model=config.model,
        messages=[{"role": "user", "content": prompt}],
        max_tokens=config.max_tokens
    )

    return {
        "type": "csv_analysis",
        "result": response.choices[0].message.content,
        "rows": len(rows),
        "columns": headers
    }

selector.add_pipeline(Pipeline(
    name="csv_analyzer",
    input_type="text",
    handler=csv_analyze_handler,
    description="Analyzes CSV files",
    supports_question=True,
    estimated_cost_per_call=0.005
))

Exercise 2: Multi-file routing

Extend the selector so it takes multiple files and combines them into a single analysis when they're about the same topic.

def process_related_files(
    selector: UseCaseSelector,
    paths: list[str],
    question: str = None
) -> dict:
    individual_results = []
    for path in paths:
        result = selector.process(path)
        if result["status"] == "success":
            individual_results.append({
                "path": path,
                "type": result["detection"]["type"],
                "summary": str(result["result"])[:500]
            })

    if len(individual_results) > 1:
        summaries = "\n\n".join(
            f"[{r['type']}] {r['path']}:\n{r['summary']}"
            for r in individual_results
        )

        prompt = f"Analysis of {len(individual_results)} files:\n\n{summaries}\n\n"
        if question:
            prompt += f"Question: {question}"
        else:
            prompt += "Generate an integrated analysis that connects the information from all the files."

        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            max_tokens=600
        )

        return {
            "type": "multi_file_analysis",
            "integrated_result": response.choices[0].message.content,
            "individual_results": individual_results,
            "files_processed": len(individual_results)
        }

    return {
        "type": "single_file",
        "results": individual_results
    }

Summary

The Use Case Selector is:

• A router that detects the input type and selects the correct pipeline
• A registry that manages the available pipelines and lets you add new ones
• A tracker that monitors costs and performance
• The foundation of Module 8's Document Analyzer

Components implemented:

Next module: Module 8 — Final Project: Multimodal Document Analyzer. The Use Case Selector you built here integrates as the routing component of a complete system that processes documents, extracts data with vision, indexes for Q&A with RAG, and generates audio with TTS.