Source code for langchain.chains.combine_documents.reduce
"""Combine many documents together by recursively reducing them."""
from __future__ import annotations
from typing import Any, Callable, List, Optional, Protocol, Tuple
from langchain_core.documents import Document
from langchain_core.pydantic_v1 import Extra
from langchain.callbacks.manager import Callbacks
from langchain.chains.combine_documents.base import BaseCombineDocumentsChain
[docs]class CombineDocsProtocol(Protocol):
    """Interface for the combine_docs method."""
    def __call__(self, docs: List[Document], **kwargs: Any) -> str:
        """Interface for the combine_docs method."""
[docs]class AsyncCombineDocsProtocol(Protocol):
    """Interface for the combine_docs method."""
    async def __call__(self, docs: List[Document], **kwargs: Any) -> str:
        """Async interface for the combine_docs method."""
[docs]def split_list_of_docs(
    docs: List[Document], length_func: Callable, token_max: int, **kwargs: Any
) -> List[List[Document]]:
    """Split Documents into subsets that each meet a cumulative length constraint.
    Args:
        docs: The full list of Documents.
        length_func: Function for computing the cumulative length of a set of Documents.
        token_max: The maximum cumulative length of any subset of Documents.
        **kwargs: Arbitrary additional keyword params to pass to each call of the
            length_func.
    Returns:
        A List[List[Document]].
    """
    new_result_doc_list = []
    _sub_result_docs = []
    for doc in docs:
        _sub_result_docs.append(doc)
        _num_tokens = length_func(_sub_result_docs, **kwargs)
        if _num_tokens > token_max: