Loaders

Document Loaders — load data from different sources for RAG.

Resolution chain for loader imports: 1. Core classes (always available — defined directly in this module) 2. parrot_loaders (ai-parrot-loaders installed package) 3. plugins.loaders (user/deploy-time plugin directory) 4. LOADER_REGISTRY (declarative registry from ai-parrot-loaders) 5. Legacy dynamic_import_helper (backward-compat submodule resolution)

Submodule redirector

from parrot.loaders.audio import X is transparently redirected to from parrot_loaders.audio import X when no local submodule exists. This is done via a sys.meta_path finder installed at import time.

AbstractLoader

AbstractLoader(source: str | Path | list[str | Path] | None = None, *, tokenizer: str | Callable | None = None, text_splitter: str | Callable | None = None, source_type: str = 'file', language: str = 'en', **kwargs)

Bases: ABC

Base class for all loaders. Loaders are responsible for loading data from various sources.

Initialize the AbstractLoader.

PARAMETER	DESCRIPTION
source	Path, URL, or list of paths/URLs to load from TYPE: str | Path | list[str | Path] | None DEFAULT: None
tokenizer	Tokenizer to use (string model name or callable) TYPE: str | Callable | None DEFAULT: None
text_splitter	Text splitter to use TYPE: str | Callable | None DEFAULT: None
source_type	Type of source ('file', 'url', etc.) TYPE: str DEFAULT: 'file'
language	Default document language ISO code (e.g. "en", "es"). Propagated to every document_meta produced by this loader unless overridden per call. TYPE: str DEFAULT: 'en'
**kwargs	Additional keyword arguments for configuration DEFAULT: {}

Source code in packages/ai-parrot/src/parrot/loaders/abstract.py

def __init__(
    self,
    source: str | Path | list[str | Path] | None = None,
    *,
    tokenizer: str | Callable | None = None,
    text_splitter: str | Callable | None = None,
    source_type: str = "file",
    language: str = "en",
    **kwargs,
):
    """
    Initialize the AbstractLoader.

    Args:
        source: Path, URL, or list of paths/URLs to load from
        tokenizer: Tokenizer to use (string model name or callable)
        text_splitter: Text splitter to use
        source_type: Type of source ('file', 'url', etc.)
        language: Default document language ISO code (e.g. ``"en"``,
            ``"es"``).  Propagated to every ``document_meta`` produced
            by this loader unless overridden per call.
        **kwargs: Additional keyword arguments for configuration
    """
    self.chunk_size: int = kwargs.get("chunk_size", 2048)
    self.chunk_overlap: int = kwargs.get("chunk_overlap", 200)
    self.min_chunk_size: int = kwargs.get("min_chunk_size", 30)
    self.full_document: bool = kwargs.get("full_document", True)
    self.semaphore = asyncio.Semaphore(kwargs.get("semaphore", 10))
    self.extensions = kwargs.get("extensions", self.extensions)  # type: ignore[misc]
    self.skip_directories = kwargs.get("skip_directories", self.skip_directories)  # type: ignore[misc]
    self.encoding = kwargs.get("encoding", "utf-8")
    self._source_type = source_type
    self._recursive: bool = kwargs.get("recursive", False)
    self.category: str = kwargs.get("category", "document")
    self.doctype: str = kwargs.get("doctype", "text")
    self.language: str = language
    # Chunking configuration
    self._use_markdown_splitter: bool = kwargs.get("use_markdown_splitter", True)
    self._use_huggingface_splitter: bool = kwargs.get("use_huggingface_splitter", False)
    self._auto_detect_content_type: bool = kwargs.get("auto_detect_content_type", True)

    # Advanced features
    self._summarization = kwargs.get("summarization", False)
    self._summary_model: Any | None = kwargs.get("summary_model", None)
    self._use_summary_pipeline: bool = kwargs.get("use_summary_pipeline", False)
    self._use_translation_pipeline: bool = kwargs.get("use_translation_pipeline", False)
    self._translation = kwargs.get("translation", False)

    # Handle source/path initialization
    self.path = None
    if source is not None:
        self.path = source
    elif "path" in kwargs:
        self.path = kwargs["path"]

    # Normalize path if it's a string. URL-like strings must be kept
    # verbatim: ``Path("https://example.com").resolve()`` corrupts the
    # scheme to ``/cwd/https:/example.com`` and makes ``load()`` dispatch
    # to ``from_path`` instead of ``from_url``, yielding zero documents
    # from loaders that expect URL sources (e.g. WebScrapingLoader).
    # Lists of URLs are also preserved as-is.
    def _is_url(value: Any) -> bool:
        return isinstance(value, str) and value.startswith(("http://", "https://"))

    if self.path is not None:
        if isinstance(self.path, list):
            # Don't resolve any element that looks like a URL.
            if any(_is_url(item) for item in self.path):
                # Heterogeneous lists (mix of URLs and paths) are left
                # untouched — the caller is responsible for them.
                pass
            else:
                # All-path list: leave as-is (downstream handles it).
                pass
        elif _is_url(self.path):
            # Keep URL string untouched.
            pass
        elif isinstance(self.path, (str, Path, PurePath)):
            self.path = Path(self.path).resolve()

    # Tokenizer
    self.tokenizer = tokenizer
    # Text Splitter
    self.text_splitter = kwargs.get("text_splitter", None)
    self.markdown_splitter = kwargs.get("markdown_splitter", None)

    # Initialize text splitter based on configuration
    self._setup_text_splitters(tokenizer, text_splitter, kwargs)

    # Summarization Model:
    self.summarization_model = kwargs.get("summarizer", None)
    # LLM (if required)
    self._setup_llm(kwargs)
    # Logger
    self.logger = logging.getLogger(f"Parrot.Loaders.{self.__class__.__name__}")
    # JSON encoder:
    self._encoder = JSONContent()
    # Use CUDA if available:
    self._setup_device(kwargs)

get_default_llm

get_default_llm(model: str | None = None, model_kwargs: dict | None = None, use_groq: bool = False, use_openai: bool = False) -> Any

Return a AI Client instance.

Source code in packages/ai-parrot/src/parrot/loaders/abstract.py

def get_default_llm(
    self,
    model: str | None = None,
    model_kwargs: dict | None = None,
    use_groq: bool = False,
    use_openai: bool = False,
) -> Any:
    """Return a AI Client instance."""
    if not model_kwargs:
        model_kwargs = {
            "temperature": DEFAULT_LLM_TEMPERATURE,
            "top_k": 30,
            "top_p": 0.5,
        }
    if use_groq:
        return LLMFactory.create(
            llm=f"groq:{model or DEFAULT_GROQ_MODEL}" if model else "groq", model_kwargs=model_kwargs
        )
    elif use_openai:
        return LLMFactory.create(llm=f"openai:{model}" if model else "openai", model_kwargs=model_kwargs)
    return LLMFactory.create(llm=model or DEFAULT_LLM_MODEL, model_kwargs=model_kwargs)

supported_extensions

supported_extensions()

Get the supported file extensions.

Source code in packages/ai-parrot/src/parrot/loaders/abstract.py

def supported_extensions(self):
    """Get the supported file extensions."""
    return self.extensions

is_valid_path

is_valid_path(path: str | Path) -> bool

Check if a path is valid.

Source code in packages/ai-parrot/src/parrot/loaders/abstract.py

def is_valid_path(self, path: str | Path) -> bool:
    """Check if a path is valid."""
    if self.extensions == "*":
        return True
    if isinstance(path, str):
        path = Path(path)
    if not path.exists():
        return False
    if path.is_dir() and path.name in self.skip_directories:
        return False
    if path.is_file():
        if path.suffix not in self.extensions:
            return False
        if path.name.startswith("."):
            return False
        # check if file is empty
        if path.stat().st_size == 0:
            return False
        # check if file is inside of skip directories:
        for skip_dir in self.skip_directories:
            if path.is_relative_to(skip_dir):
                return False
    return True

from_path async

from_path(path: str | Path, recursive: bool = False, **kwargs) -> list[asyncio.Task]

Load data from a path.

Source code in packages/ai-parrot/src/parrot/loaders/abstract.py

async def from_path(self, path: str | Path, recursive: bool = False, **kwargs) -> list[asyncio.Task]:
    """
    Load data from a path.
    """
    tasks = []
    if isinstance(path, str):
        path = PurePath(path)
    if path.is_dir():
        for ext in self.extensions:
            glob_method = path.rglob if recursive else path.glob
            # Use glob to find all files with the specified extension
            for item in glob_method(f"*{ext}"):
                # Check if the item is a directory and if it should be skipped
                if set(item.parts).isdisjoint(self.skip_directories) and self.is_valid_path(item):
                    tasks.append(asyncio.create_task(self._load(item, **kwargs)))
    elif path.is_file():
        if self.is_valid_path(path):
            tasks.append(asyncio.create_task(self._load(path, **kwargs)))
    else:
        self.logger.warning(f"Path {path} is not valid.")
    return tasks

from_url async

from_url(url: str | list[str], **kwargs) -> list[asyncio.Task]

Load data from a URL.

Source code in packages/ai-parrot/src/parrot/loaders/abstract.py

async def from_url(self, url: str | list[str], **kwargs) -> list[asyncio.Task]:
    """
    Load data from a URL.
    """
    tasks = []
    if isinstance(url, str):
        url = [url]
    for item in url:
        tasks.append(asyncio.create_task(self._load(item, **kwargs)))
    return tasks

from_dataframe async

from_dataframe(source: DataFrame, **kwargs) -> list[asyncio.Task]

Load data from a pandas DataFrame.

Source code in packages/ai-parrot/src/parrot/loaders/abstract.py

async def from_dataframe(self, source: pd.DataFrame, **kwargs) -> list[asyncio.Task]:
    """
    Load data from a pandas DataFrame.
    """
    tasks = []
    try:
        import pandas as pd

        if isinstance(source, pd.DataFrame):
            tasks.append(asyncio.create_task(self._load(source, **kwargs)))
        else:
            self.logger.warning(f"Source {source} is not a valid pandas DataFrame.")
    except ImportError:
        self.logger.warning("Pandas not installed, cannot load from DataFrame")
    return tasks

chunkify

chunkify(lst: list[T], n: int = 50) -> Generator[list[T], None, None]

Split a List of objects into chunks of size n.

PARAMETER	DESCRIPTION
lst	The list to split into chunks TYPE: list[T]
n	The maximum size of each chunk TYPE: int DEFAULT: 50

YIELDS	DESCRIPTION
list[T]	List[T]: Chunks of the original list, each of size at most n

Source code in packages/ai-parrot/src/parrot/loaders/abstract.py

def chunkify(self, lst: list[T], n: int = 50) -> Generator[list[T], None, None]:
    """Split a List of objects into chunks of size n.

    Args:
        lst: The list to split into chunks
        n: The maximum size of each chunk

    Yields:
        List[T]: Chunks of the original list, each of size at most n
    """
    for i in range(0, len(lst), n):
        yield lst[i : i + n]

load async

load(source: Any | None = None, split_documents: bool = True, late_chunking: bool = False, vector_store=None, store_full_document: bool = True, auto_detect_content_type: bool | None = None, **kwargs) -> list[Document]

Load data from a source and return it as a list of Documents.

The source can be: - None: Uses self.path attribute if available - Path or str: Treated as file path or directory - List[str/Path]: Treated as list of file paths - URL string: Treated as a URL - List of URLs: Treated as list of URLs

PARAMETER	DESCRIPTION
source	The source of the data. TYPE: Optional[Any] DEFAULT: None
split_documents	Whether to split documents into chunks, defaults to True TYPE: bool DEFAULT: True
late_chunking	Whether to use late chunking strategy TYPE: bool DEFAULT: False
vector_store	Vector store instance (required for late chunking) DEFAULT: None
store_full_document	Whether to store full documents alongside chunks TYPE: bool DEFAULT: True
auto_detect_content_type	Override auto-detection setting TYPE: bool DEFAULT: None
**kwargs	Additional keyword arguments DEFAULT: {}

RETURNS	DESCRIPTION
list[Document]	List[Document]: A list of Documents (chunked if requested).

Source code in packages/ai-parrot/src/parrot/loaders/abstract.py

async def load(
    self,
    source: Any | None = None,
    split_documents: bool = True,
    late_chunking: bool = False,
    vector_store=None,
    store_full_document: bool = True,
    auto_detect_content_type: bool | None = None,
    **kwargs,
) -> list[Document]:
    """
    Load data from a source and return it as a list of Documents.

    The source can be:
    - None: Uses self.path attribute if available
    - Path or str: Treated as file path or directory
    - List[str/Path]: Treated as list of file paths
    - URL string: Treated as a URL
    - List of URLs: Treated as list of URLs

    Args:
        source (Optional[Any]): The source of the data.
        split_documents (bool): Whether to split documents into chunks, defaults to True
        late_chunking (bool): Whether to use late chunking strategy
        vector_store: Vector store instance (required for late chunking)
        store_full_document (bool): Whether to store full documents alongside chunks
        auto_detect_content_type (bool): Override auto-detection setting
        **kwargs: Additional keyword arguments

    Returns:
        List[Document]: A list of Documents (chunked if requested).
    """
    tasks = []
    # If no source is provided, use self.path
    if source is None:
        if self.path is None:
            raise ValueError(
                "No source provided and self.path is not set. "
                "Please provide a source parameter or set path during initialization."
            )
        source = self.path

    if isinstance(source, (str, Path, PosixPath, PurePath)):
        # Check if it's a URL
        if isinstance(source, str) and source.startswith(("http://", "https://")):
            tasks = await self.from_url(source, **kwargs)
        else:
            # Assume it's a file path or directory
            tasks = await self.from_path(source, recursive=self._recursive, **kwargs)
    elif isinstance(source, list):
        # Check if it's a list of URLs or paths
        if all(isinstance(item, str) and item.startswith(("http://", "https://")) for item in source):
            tasks = await self.from_url(source, **kwargs)
        else:
            # Assume it's a list of file paths
            path_tasks = []
            for path in source:
                path_tasks.extend(await self.from_path(path, recursive=self._recursive, **kwargs))
            tasks = path_tasks
    else:
        # Check for DataFrame lazily
        is_dataframe = False
        try:
            import pandas as pd

            if isinstance(source, pd.DataFrame):
                is_dataframe = True
                tasks = await self.from_dataframe(source, **kwargs)
        except ImportError:
            pass

        if not is_dataframe:
            raise ValueError(f"Unsupported source type: {type(source)}")
    # Load tasks and get raw documents
    documents = []
    if tasks:
        results = await self._load_tasks(tasks)
        documents = results

    # Apply chunking if requested
    if split_documents and documents:
        self.logger.debug(f"Splitting {len(documents)} documents into chunks...")

        if late_chunking and vector_store is None:
            raise ValueError("Vector store is required when using late_chunking=True")

        documents = await self.chunk_documents(
            documents=documents,
            use_late_chunking=late_chunking,
            vector_store=vector_store,
            store_full_document=store_full_document,
            auto_detect_content_type=auto_detect_content_type,
        )

        self.logger.debug(f"Document chunking complete: {len(documents)} final documents")

    return documents

create_metadata

create_metadata(path: str | PurePath, doctype: str = 'document', source_type: str = 'source', doc_metadata: dict | None = None, *, language: str | None = None, title: str | None = None, **kwargs) -> dict

Build a canonical Document.metadata dict.

The returned dict always contains:

• Top-level: url, source, filename, type, source_type, created_at, category, document_meta, plus any extra **kwargs (loader-specific extras).
• document_meta (closed shape): exactly {source_type, category, type, language, title}.

Legacy callers that pass doc_metadata continue to work: canonical keys in that dict are folded into document_meta; non-canonical keys are hoisted to the top-level metadata.

PARAMETER	DESCRIPTION
path	Filesystem path or URL of the source document. TYPE: str | PurePath
doctype	Document type identifier (e.g. "pdf", "audio_transcript"). TYPE: str DEFAULT: 'document'
source_type	High-level source kind (e.g. "file", "url", "video"). TYPE: str DEFAULT: 'source'
doc_metadata	Legacy dict of additional metadata. Canonical keys (source_type, category, type, language, title) are merged into document_meta; all others become top-level metadata keys. TYPE: dict | None DEFAULT: None
language	Document language ISO code. Defaults to self.language when None. TYPE: str | None DEFAULT: None
title	Human-readable title. Defaults to self._derive_title(path) when None. TYPE: str | None DEFAULT: None
**kwargs	Loader-specific extras — become top-level keys in the returned metadata dict. Must not be canonical document_meta keys. DEFAULT: {}

RETURNS	DESCRIPTION
dict	A metadata dict with canonical top-level keys and a
dict	closed-shape document_meta sub-dict.

Source code in packages/ai-parrot/src/parrot/loaders/abstract.py

def create_metadata(
    self,
    path: str | PurePath,
    doctype: str = "document",
    source_type: str = "source",
    doc_metadata: dict | None = None,
    *,
    language: str | None = None,
    title: str | None = None,
    **kwargs,
) -> dict:
    """Build a canonical ``Document.metadata`` dict.

    The returned dict always contains:

    * Top-level: ``url``, ``source``, ``filename``, ``type``,
      ``source_type``, ``created_at``, ``category``, ``document_meta``,
      plus any extra ``**kwargs`` (loader-specific extras).
    * ``document_meta`` (closed shape): exactly
      ``{source_type, category, type, language, title}``.

    Legacy callers that pass ``doc_metadata`` continue to work:
    canonical keys in that dict are folded into ``document_meta``;
    non-canonical keys are hoisted to the top-level metadata.

    Args:
        path: Filesystem path or URL of the source document.
        doctype: Document type identifier (e.g. ``"pdf"``,
            ``"audio_transcript"``).
        source_type: High-level source kind (e.g. ``"file"``,
            ``"url"``, ``"video"``).
        doc_metadata: Legacy dict of additional metadata.  Canonical
            keys (``source_type``, ``category``, ``type``,
            ``language``, ``title``) are merged into ``document_meta``;
            all others become top-level metadata keys.
        language: Document language ISO code.  Defaults to
            ``self.language`` when ``None``.
        title: Human-readable title.  Defaults to
            ``self._derive_title(path)`` when ``None``.
        **kwargs: Loader-specific extras — become top-level keys in
            the returned metadata dict.  Must **not** be canonical
            ``document_meta`` keys.

    Returns:
        A metadata dict with canonical top-level keys and a
        closed-shape ``document_meta`` sub-dict.
    """
    if not doc_metadata:
        doc_metadata = {}

    # Resolve path-derived fields
    if isinstance(path, PurePath):
        origin = path.name
        url = f"file://{path}"
        filename = str(path)
    else:
        origin = path
        url = path
        filename = f"file://{path}"

    # Resolve language and title
    resolved_language = language if language is not None else self.language
    resolved_title = title if title is not None else self._derive_title(path)

    # Separate canonical keys in legacy doc_metadata from extras
    canonical_from_legacy: dict = {}
    extra_from_legacy: dict = {}
    for key, val in doc_metadata.items():
        if key in self._CANONICAL_DOC_META_KEYS:
            canonical_from_legacy[key] = val
        else:
            extra_from_legacy[key] = val

    # Canonical doc_metadata keys override derived values when present
    resolved_language = canonical_from_legacy.get("language", resolved_language)
    resolved_title = canonical_from_legacy.get("title", resolved_title)
    resolved_doctype = canonical_from_legacy.get("type", doctype)
    resolved_source_type = canonical_from_legacy.get("source_type", source_type or self._source_type)
    resolved_category = canonical_from_legacy.get("category", self.category)

    document_meta = {
        "source_type": resolved_source_type,
        "category": resolved_category,
        "type": resolved_doctype,
        "language": resolved_language,
        "title": resolved_title,
    }

    metadata = {
        "url": url,
        "source": origin,
        "filename": filename,
        "type": resolved_doctype,
        "source_type": resolved_source_type,
        "created_at": datetime.now(tz=UTC).strftime("%Y-%m-%d, %H:%M:%S"),
        "category": resolved_category,
        "document_meta": document_meta,
        # Extras from legacy doc_metadata — top level
        **extra_from_legacy,
        # Caller-supplied extras — top level
        **kwargs,
    }
    return metadata

create_document

create_document(content: Any, path: str | PurePath, metadata: dict | None = None, **kwargs) -> Document

Create a Parrot Document from content.

If metadata is None, create_metadata is called with path, self.doctype, and self._source_type (plus any **kwargs). The resulting metadata is validated via _validate_metadata before the Document is constructed.

PARAMETER	DESCRIPTION
content	The text content of the document. TYPE: Any
path	Source path or URL (used when metadata is None). TYPE: str | PurePath
metadata	Pre-built metadata dict. When provided it is still passed through _validate_metadata so canonical fields are auto-filled if missing. TYPE: dict | None DEFAULT: None
**kwargs	Forwarded to create_metadata when metadata is None. DEFAULT: {}

RETURNS	DESCRIPTION
Document	A Document with validated canonical metadata.

Source code in packages/ai-parrot/src/parrot/loaders/abstract.py

def create_document(self, content: Any, path: str | PurePath, metadata: dict | None = None, **kwargs) -> Document:
    """Create a Parrot Document from content.

    If *metadata* is ``None``, ``create_metadata`` is called with
    *path*, ``self.doctype``, and ``self._source_type`` (plus any
    ``**kwargs``).  The resulting metadata is validated via
    ``_validate_metadata`` before the ``Document`` is constructed.

    Args:
        content: The text content of the document.
        path: Source path or URL (used when *metadata* is ``None``).
        metadata: Pre-built metadata dict.  When provided it is still
            passed through ``_validate_metadata`` so canonical fields
            are auto-filled if missing.
        **kwargs: Forwarded to ``create_metadata`` when *metadata* is
            ``None``.

    Returns:
        A ``Document`` with validated canonical metadata.
    """
    if metadata:
        _meta = metadata
    else:
        _meta = self.create_metadata(path=path, doctype=self.doctype, source_type=self._source_type, **kwargs)
    _meta = self._validate_metadata(_meta)
    return Document(page_content=content, metadata=_meta)

summary_from_text async

summary_from_text(text: str, max_length: int = 500, min_length: int = 50) -> str

Get a summary of a text.

Source code in packages/ai-parrot/src/parrot/loaders/abstract.py

async def summary_from_text(self, text: str, max_length: int = 500, min_length: int = 50) -> str:
    """
    Get a summary of a text.
    """
    if not text:
        return ""
    try:
        summarizer = self.get_summarization_model()
        if self._use_summary_pipeline:
            # Use Huggingface pipeline
            content = summarizer(
                text, max_length=max_length, min_length=min_length, do_sample=False, truncation=True
            )
            return content[0].get("summary_text", "")
        # Use Summarize Method from GoogleGenAIClient (synchronous)
        summary = await asyncio.to_thread(
            summarizer.summarize_text,
            text=text,
            model="gemini-2.5-flash-lite",
            temperature=0.1,
        )
        return summary.output
    except Exception as e:  # noqa: BLE001
        self.logger.error(f"ERROR on summary_from_text: {e}")
        return ""

translate_text

translate_text(text: str, source_lang: str | None = None, target_lang: str = 'es') -> str

Translate text from source language to target language.

PARAMETER	DESCRIPTION
text	Text to translate TYPE: str
source_lang	Source language code (default: 'en') TYPE: str | None DEFAULT: None
target_lang	Target language code (default: 'es') TYPE: str DEFAULT: 'es'

RETURNS	DESCRIPTION
str	Translated text

Source code in packages/ai-parrot/src/parrot/loaders/abstract.py

def translate_text(self, text: str, source_lang: str | None = None, target_lang: str = "es") -> str:
    """
    Translate text from source language to target language.

    Args:
        text: Text to translate
        source_lang: Source language code (default: 'en')
        target_lang: Target language code (default: 'es')

    Returns:
        Translated text
    """
    if not text:
        return ""
    try:
        translator = self.get_translation_model(source_lang, target_lang)
        if self._use_translation_pipeline:
            # Use Huggingface pipeline
            content = translator(
                text, max_length=len(text) * 2, truncation=True  # Allow for expansion in target language
            )
            return content[0].get("translation_text", "")
        else:
            # Use LLM for translation
            translation = translator.translate_text(
                text=text,
                source_lang=source_lang,
                target_lang=target_lang,
                model="gemini-2.5-flash-lite",
                temperature=0.1,
                max_tokens=1000,
            )
            return translation.output if hasattr(translation, "output") else str(translation)
    except Exception as e:  # noqa: BLE001
        self.logger.error(f"ERROR on translate_text: {e}")
        return ""

get_translation_model

get_translation_model(source_lang: str = 'en', target_lang: str = 'es', model_name: str | None = None)

Get or create a translation model.

PARAMETER	DESCRIPTION
source_lang	Source language code TYPE: str DEFAULT: 'en'
target_lang	Target language code TYPE: str DEFAULT: 'es'
model_name	Optional model name override TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
	Translation model/chain

Source code in packages/ai-parrot/src/parrot/loaders/abstract.py

def get_translation_model(self, source_lang: str = "en", target_lang: str = "es", model_name: str | None = None):
    """
    Get or create a translation model.

    Args:
        source_lang: Source language code
        target_lang: Target language code
        model_name: Optional model name override

    Returns:
        Translation model/chain
    """
    # Create a cache key for the language pair
    cache_key = f"{source_lang}_{target_lang}"

    # Check if we already have a model for this language pair
    if not hasattr(self, "_translation_models"):
        self._translation_models = {}

    if cache_key not in self._translation_models:
        if self._use_translation_pipeline:
            from transformers import AutoModelForSeq2SeqLM, AutoTokenizer

            # Select appropriate model based on language pair if not specified
            if model_name is None:
                if source_lang == "en" and target_lang in ["es", "fr", "de", "it", "pt", "ru"]:
                    model_name = "Helsinki-NLP/opus-mt-en-ROMANCE"
                elif source_lang in ["es", "fr", "de", "it", "pt"] and target_lang == "en":
                    model_name = "Helsinki-NLP/opus-mt-ROMANCE-en"
                else:
                    # Default to a specific model for the language pair
                    model_name = f"Helsinki-NLP/opus-mt-{source_lang}-{target_lang}"

            try:
                translate_model = AutoModelForSeq2SeqLM.from_pretrained(model_name)
                translate_tokenizer = AutoTokenizer.from_pretrained(model_name)

                self._translation_models[cache_key] = Seq2SeqGenerator(
                    translate_model, translate_tokenizer, output_key="translation_text"
                )
            except Exception as e:  # noqa: BLE001
                self.logger.error(f"Error loading translation model {model_name}: {e}")
                # Fallback to using LLM for translation
                self._use_translation_pipeline = False

        if not self._use_translation_pipeline:
            # Use LLM for translation
            translation_model = self.get_default_llm(model="gemini-2.5-flash-lite")
            self._translation_models[cache_key] = translation_model

    return self._translation_models[cache_key]

create_translated_document

create_translated_document(content: str, metadata: dict, source_lang: str = 'en', target_lang: str = 'es') -> Document

Create a document with translated content.

PARAMETER	DESCRIPTION
content	Original content TYPE: str
metadata	Document metadata TYPE: dict
source_lang	Source language code TYPE: str DEFAULT: 'en'
target_lang	Target language code TYPE: str DEFAULT: 'es'

RETURNS	DESCRIPTION
Document	Document with translated content

Source code in packages/ai-parrot/src/parrot/loaders/abstract.py

def create_translated_document(
    self, content: str, metadata: dict, source_lang: str = "en", target_lang: str = "es"
) -> Document:
    """
    Create a document with translated content.

    Args:
        content: Original content
        metadata: Document metadata
        source_lang: Source language code
        target_lang: Target language code

    Returns:
        Document with translated content
    """
    translated_content = self.translate_text(content, source_lang, target_lang)

    # Clone the metadata and add translation info
    translation_metadata = metadata.copy()
    translation_metadata.update({"original_language": source_lang, "language": target_lang, "is_translation": True})

    return Document(page_content=translated_content, metadata=translation_metadata)

saving_file

saving_file(filename: PurePath, data: Any)

Save data to a file.

PARAMETER	DESCRIPTION
filename	The path to the file. TYPE: PurePath
data	The data to save. TYPE: Any

Source code in packages/ai-parrot/src/parrot/loaders/abstract.py

def saving_file(self, filename: PurePath, data: Any):
    """Save data to a file.

    Args:
        filename (PurePath): The path to the file.
        data (Any): The data to save.
    """
    with open(filename, "wb") as f:
        f.write(data)
        f.flush()
    self.logger.info("Saved file on %s", filename)

chunk_documents async

chunk_documents(documents: list[Document], use_late_chunking: bool = False, vector_store=None, store_full_document: bool = True, auto_detect_content_type: bool | None = None) -> list[Document]

Chunk documents using the configured text splitter or late chunking strategy.

PARAMETER	DESCRIPTION
documents	List of documents to chunk TYPE: list[Document]
use_late_chunking	Whether to use late chunking strategy TYPE: bool DEFAULT: False
vector_store	Vector store instance (required for late chunking) DEFAULT: None
store_full_document	Whether to store full documents alongside chunks (late chunking only) TYPE: bool DEFAULT: True
auto_detect_content_type	Override auto-detection setting TYPE: bool | None DEFAULT: None

RETURNS	DESCRIPTION
list[Document]	List of chunked documents

Source code in packages/ai-parrot/src/parrot/loaders/abstract.py

async def chunk_documents(
    self,
    documents: list[Document],
    use_late_chunking: bool = False,
    vector_store=None,
    store_full_document: bool = True,
    auto_detect_content_type: bool | None = None,
) -> list[Document]:
    """
    Chunk documents using the configured text splitter or late chunking strategy.

    Args:
        documents: List of documents to chunk
        use_late_chunking: Whether to use late chunking strategy
        vector_store: Vector store instance (required for late chunking)
        store_full_document: Whether to store full documents alongside chunks (late chunking only)
        auto_detect_content_type: Override auto-detection setting

    Returns:
        List of chunked documents
    """
    if use_late_chunking:
        return await self._chunk_with_late_chunking(documents, vector_store, store_full_document)
    else:
        return self._chunk_with_text_splitter(documents, auto_detect_content_type)

Document

Bases: BaseModel

A simple document model for adding data to the vector store. This replaces langchain.docstore.document.Document.