_collapse() — langchain Function Reference

Architecture documentation for the _collapse() function in reduce.py from the langchain codebase.

Function python AgentOrchestration ClassicChains calls 2 called by 1

Entity Profile

AgentOrchestration→ ClassicChains→ _collapse() — langchain Function Reference

Dependency Diagram

graph TD
  5ec5e85d_0e34_733e_07b9_d13ee52ef47f["_collapse()"]
  1d761688_0112_f998_eecc_6712fc22bf37["ReduceDocumentsChain"]
  5ec5e85d_0e34_733e_07b9_d13ee52ef47f -->|defined in| 1d761688_0112_f998_eecc_6712fc22bf37
  cc4540ae_400a_89bd_9f59_c90678c2a044["combine_docs()"]
  cc4540ae_400a_89bd_9f59_c90678c2a044 -->|calls| 5ec5e85d_0e34_733e_07b9_d13ee52ef47f
  c98dc7a1_c3dc_2557_0a0f_d34b8e2c7323["split_list_of_docs()"]
  5ec5e85d_0e34_733e_07b9_d13ee52ef47f -->|calls| c98dc7a1_c3dc_2557_0a0f_d34b8e2c7323
  e5e85969_ef5a_d0e5_f524_145250c67cf1["collapse_docs()"]
  5ec5e85d_0e34_733e_07b9_d13ee52ef47f -->|calls| e5e85969_ef5a_d0e5_f524_145250c67cf1
  style 5ec5e85d_0e34_733e_07b9_d13ee52ef47f fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain/langchain_classic/chains/combine_documents/reduce.py lines 309–346

    def _collapse(
        self,
        docs: list[Document],
        token_max: int | None = None,
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> tuple[list[Document], dict]:
        result_docs = docs
        length_func = self.combine_documents_chain.prompt_length
        num_tokens = length_func(result_docs, **kwargs)

        def _collapse_docs_func(docs: list[Document], **kwargs: Any) -> str:
            return self._collapse_chain.run(
                input_documents=docs,
                callbacks=callbacks,
                **kwargs,
            )

        _token_max = token_max or self.token_max
        retries: int = 0
        while num_tokens is not None and num_tokens > _token_max:
            new_result_doc_list = split_list_of_docs(
                result_docs,
                length_func,
                _token_max,
                **kwargs,
            )
            result_docs = [
                collapse_docs(docs_, _collapse_docs_func, **kwargs)
                for docs_ in new_result_doc_list
            ]
            num_tokens = length_func(result_docs, **kwargs)
            retries += 1
            if self.collapse_max_retries and retries == self.collapse_max_retries:
                msg = f"Exceed {self.collapse_max_retries} tries to \
                        collapse document to {_token_max} tokens."
                raise ValueError(msg)
        return result_docs, {}