Class MTMDProcessor

examples/server/server.py:10534–11226

class MTMDProcessor:
    @dataclass
    class MediaChunk:
        kind: Literal["image", "audio", "video"]
        key: str
        chunk: Any
        n_tokens: int
        decode_n_pos: int
        non_causal: bool
        embeddings: Optional[np.ndarray] = None

    @dataclass
    class ParsedChunk:
        text_tokens: Optional[List[int]] = None
        media: Optional["MTMDProcessor.MediaChunk"] = None

    def __init__(
        self,
        *,
        model_path: str,
        llama_model: Any,
        chat_formatter: Jinja2ChatFormatter,
        tokenize: Callable[..., List[int]],
        n_embd_inp: int,
        n_batch: int,
        n_ubatch: int,
        n_threads_batch: int,
        mmproj_path: str,
        batch_max_tokens: int,
        embedding_cache: Optional[MTMDEmbeddingCache],
        allowed_media_domains: Optional[List[str]],
        allowed_local_media_path: Optional[str],
        image_max_bytes: int,
        audio_max_bytes: int,
        video_max_bytes: int,
        image_timeout_seconds: float,
    ) -> None:
        self.chat_formatter = chat_formatter
        self.tokenize = tokenize
        self.n_embd_inp = n_embd_inp
        self.n_batch = n_batch
        self.n_ubatch = n_ubatch
        self.mmproj_path = mmproj_path
        self.embedding_cache = embedding_cache
        self.batch_max_tokens = batch_max_tokens
        self.model_fingerprint = MTMDEmbeddingCache.fingerprint_file(model_path)
        self.mmproj_fingerprint = MTMDEmbeddingCache.fingerprint_file(mmproj_path)
        self.allowed_media_domains = (
            {domain.lower() for domain in allowed_media_domains}
            if allowed_media_domains is not None
            else set()
        )
        self.allowed_local_media_path = (
            Path(allowed_local_media_path).expanduser().resolve()
            if allowed_local_media_path is not None
            else None
        )
        self.image_max_bytes = image_max_bytes