langchain Architecture Docs
Home / File/ base.py — langchain Source File

base.py — langchain Source File

Architecture documentation for base.py, a python file in the langchain codebase. 5 imports, 0 dependents.

File python CoreAbstractions RunnableInterface 5 imports 1 classes

Entity Profile

Dependency Diagram

graph LR
  75de190b_0731_ecff_0670_12db0d5e9795["base.py"]
  cfe2bde5_180e_e3b0_df2b_55b3ebaca8e7["collections.abc"]
  75de190b_0731_ecff_0670_12db0d5e9795 --> cfe2bde5_180e_e3b0_df2b_55b3ebaca8e7
  614e7b9f_ed51_0780_749c_ff40b74963fc["inspect"]
  75de190b_0731_ecff_0670_12db0d5e9795 --> 614e7b9f_ed51_0780_749c_ff40b74963fc
  f3bc7443_c889_119d_0744_aacc3620d8d2["langchain_core.callbacks"]
  75de190b_0731_ecff_0670_12db0d5e9795 --> f3bc7443_c889_119d_0744_aacc3620d8d2
  c554676d_b731_47b2_a98f_c1c2d537c0aa["langchain_core.documents"]
  75de190b_0731_ecff_0670_12db0d5e9795 --> c554676d_b731_47b2_a98f_c1c2d537c0aa
  6e58aaea_f08e_c099_3cc7_f9567bfb1ae7["pydantic"]
  75de190b_0731_ecff_0670_12db0d5e9795 --> 6e58aaea_f08e_c099_3cc7_f9567bfb1ae7
  style 75de190b_0731_ecff_0670_12db0d5e9795 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

from collections.abc import Sequence
from inspect import signature

from langchain_core.callbacks import Callbacks
from langchain_core.documents import (
    BaseDocumentCompressor,
    BaseDocumentTransformer,
    Document,
)
from pydantic import ConfigDict


class DocumentCompressorPipeline(BaseDocumentCompressor):
    """Document compressor that uses a pipeline of Transformers."""

    transformers: list[BaseDocumentTransformer | BaseDocumentCompressor]
    """List of document filters that are chained together and run in sequence."""

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
    )

    def compress_documents(
        self,
        documents: Sequence[Document],
        query: str,
        callbacks: Callbacks | None = None,
    ) -> Sequence[Document]:
        """Transform a list of documents."""
        for _transformer in self.transformers:
            if isinstance(_transformer, BaseDocumentCompressor):
                accepts_callbacks = (
                    signature(_transformer.compress_documents).parameters.get(
                        "callbacks",
                    )
                    is not None
                )
                if accepts_callbacks:
                    documents = _transformer.compress_documents(
                        documents,
                        query,
                        callbacks=callbacks,
                    )
                else:
                    documents = _transformer.compress_documents(documents, query)
            elif isinstance(_transformer, BaseDocumentTransformer):
                documents = _transformer.transform_documents(documents)
            else:
                msg = f"Got unexpected transformer type: {_transformer}"  # type: ignore[unreachable]
                raise ValueError(msg)  # noqa: TRY004
        return documents

    async def acompress_documents(
        self,
        documents: Sequence[Document],
        query: str,
        callbacks: Callbacks | None = None,
    ) -> Sequence[Document]:
        """Compress retrieved documents given the query context."""
        for _transformer in self.transformers:
            if isinstance(_transformer, BaseDocumentCompressor):
                accepts_callbacks = (
                    signature(_transformer.acompress_documents).parameters.get(
                        "callbacks",
                    )
                    is not None
                )
                if accepts_callbacks:
                    documents = await _transformer.acompress_documents(
                        documents,
                        query,
                        callbacks=callbacks,
                    )
                else:
                    documents = await _transformer.acompress_documents(documents, query)
            elif isinstance(_transformer, BaseDocumentTransformer):
                documents = await _transformer.atransform_documents(documents)
            else:
                msg = f"Got unexpected transformer type: {_transformer}"  # type: ignore[unreachable]
                raise ValueError(msg)  # noqa: TRY004
        return documents

Domain

Subdomains

Classes

Dependencies

  • collections.abc
  • inspect
  • langchain_core.callbacks
  • langchain_core.documents
  • pydantic

Source

Frequently Asked Questions

What does base.py do?
base.py is a source file in the langchain codebase, written in python. It belongs to the CoreAbstractions domain, RunnableInterface subdomain.
What does base.py depend on?
base.py imports 5 module(s): collections.abc, inspect, langchain_core.callbacks, langchain_core.documents, pydantic.
Where is base.py in the architecture?
base.py is located at libs/langchain/langchain_classic/retrievers/document_compressors/base.py (domain: CoreAbstractions, subdomain: RunnableInterface, directory: libs/langchain/langchain_classic/retrievers/document_compressors).

Analyze Your Own Codebase

Get architecture documentation, dependency graphs, and domain analysis for your codebase in minutes.

Try Supermodel Free