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: Optional[Union[str, Path, List[Union[str, Path]]]] = None, *, tokenizer: Union[str, Callable] = None, text_splitter: Union[str, Callable] = 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: Optional[Union[str, Path, List[Union[str, Path]]]] DEFAULT: None
tokenizer	Tokenizer to use (string model name or callable) TYPE: Union[str, Callable] DEFAULT: None
text_splitter	Text splitter to use TYPE: Union[str, Callable] 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: Optional[Union[str, Path, List[Union[str, Path]]]] = None,
    *,
    tokenizer: Union[str, Callable] = None,
    text_splitter: Union[str, Callable] = 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)
    self.skip_directories = kwargs.get(
        'skip_directories',
        self.skip_directories
    )
    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: Optional[Any] = 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://') or value.startswith('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):
            self.path = Path(self.path).resolve()
        elif isinstance(self.path, (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, model_kwargs: dict = 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,
    model_kwargs: dict = 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: Union[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: Union[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: Union[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: Union[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):
                    if 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: Union[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: Union[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: Optional[Any] = None, split_documents: bool = True, late_chunking: bool = False, vector_store=None, store_full_document: bool = True, auto_detect_content_type: bool = 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: Optional[Any] = None,
    split_documents: bool = True,
    late_chunking: bool = False,
    vector_store=None,
    store_full_document: bool = True,
    auto_detect_content_type: bool = 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://') or source.startswith('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://') or item.startswith('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: Union[str, PurePath], doctype: str = 'document', source_type: str = 'source', doc_metadata: Optional[dict] = None, *, language: Optional[str] = None, title: Optional[str] = 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: Union[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: Optional[dict] DEFAULT: None
language	Document language ISO code. Defaults to self.language when None. TYPE: Optional[str] DEFAULT: None
title	Human-readable title. Defaults to self._derive_title(path) when None. TYPE: Optional[str] 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: Union[str, PurePath],
    doctype: str = 'document',
    source_type: str = 'source',
    doc_metadata: Optional[dict] = None,
    *,
    language: Optional[str] = None,
    title: Optional[str] = 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().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: Union[str, PurePath], metadata: Optional[dict] = 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: Union[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: Optional[dict] 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: Union[str, PurePath],
    metadata: Optional[dict] = 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 GroqClient
            system_prompt = f"""
Your job is to produce a final summary from the following text and identify the main theme.
- The summary should be concise and to the point.
- The summary should be no longer than {max_length} characters and no less than {min_length} characters.
- The summary should be in a single paragraph.
"""
            # Ensure the LLM client is initialized for the current loop
            await summarizer._ensure_client()
            summary = await summarizer.summarize_text(
                text=text,
                model=GroqModel.LLAMA_3_3_70B_VERSATILE,
                system_prompt=system_prompt,
                temperature=0.1,
                max_tokens=1000,
                top_p=0.5
            )
            return summary.output
        except Exception as e:
            self.logger.error(
                f'ERROR on summary_from_text: {e}'
            )
            return ""

translate_text

translate_text(text: str, source_lang: str = 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 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,
    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,  # Allow for expansion in target language
                truncation=True
            )
            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=GoogleModel.GEMINI_2_5_FLASH_LITE_PREVIEW,
                temperature=0.1,
                max_tokens=1000
            )
            return translation.output if hasattr(translation, 'output') else str(translation)
    except Exception as e:
        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)

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 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
):
    """
    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,
                pipeline
            )
            # 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] = pipeline(
                    "translation",
                    model=translate_model,
                    tokenizer=translate_tokenizer
                )
            except Exception as e:
                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=GoogleModel.GEMINI_2_5_FLASH_LITE_PREVIEW
            )
            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()
    print(f':: Saved File on {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) -> 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 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
) -> 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.