Source code for langchain_core.documents.compressor
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Optional, Sequence
from langchain_core.callbacks import Callbacks
from langchain_core.documents import Document
from langchain_core.pydantic_v1 import BaseModel
from langchain_core.runnables import run_in_executor
[docs]class BaseDocumentCompressor(BaseModel, ABC):
"""Base class for document compressors."""
[docs] @abstractmethod
def compress_documents(
self,
documents: Sequence[Document],
query: str,
callbacks: Optional[Callbacks] = None,
) -> Sequence[Document]:
"""Compress retrieved documents given the query context."""
[docs] async def acompress_documents(
self,
documents: Sequence[Document],
query: str,
callbacks: Optional[Callbacks] = None,
) -> Sequence[Document]:
"""Compress retrieved documents given the query context."""
return await run_in_executor(
None, self.compress_documents, documents, query, callbacks
)