_acall() — langchain Function Reference

Architecture documentation for the _acall() function in base.py from the langchain codebase.

Function python AgentOrchestration ClassicChains calls 2

Entity Profile

AgentOrchestration→ ClassicChains→ _acall() — langchain Function Reference

Dependency Diagram

graph TD
  b128053c_ae67_13fa_39d2_7ed6b1273cf8["_acall()"]
  be2fab99_c17a_7327_31b8_3526c30ba2b8["BaseConversationalRetrievalChain"]
  b128053c_ae67_13fa_39d2_7ed6b1273cf8 -->|defined in| be2fab99_c17a_7327_31b8_3526c30ba2b8
  80d8511e_e74f_2e34_6d79_70d274ac463d["_aget_docs()"]
  b128053c_ae67_13fa_39d2_7ed6b1273cf8 -->|calls| 80d8511e_e74f_2e34_6d79_70d274ac463d
  a4ebed9c_c61f_770b_661e_da8b990ad251["_aget_docs()"]
  b128053c_ae67_13fa_39d2_7ed6b1273cf8 -->|calls| a4ebed9c_c61f_770b_661e_da8b990ad251
  style b128053c_ae67_13fa_39d2_7ed6b1273cf8 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain/langchain_classic/chains/conversational_retrieval/base.py lines 200–245

    async def _acall(
        self,
        inputs: dict[str, Any],
        run_manager: AsyncCallbackManagerForChainRun | None = None,
    ) -> dict[str, Any]:
        _run_manager = run_manager or AsyncCallbackManagerForChainRun.get_noop_manager()
        question = inputs["question"]
        get_chat_history = self.get_chat_history or _get_chat_history
        chat_history_str = get_chat_history(inputs["chat_history"])
        if chat_history_str:
            callbacks = _run_manager.get_child()
            new_question = await self.question_generator.arun(
                question=question,
                chat_history=chat_history_str,
                callbacks=callbacks,
            )
        else:
            new_question = question
        accepts_run_manager = (
            "run_manager" in inspect.signature(self._aget_docs).parameters
        )
        if accepts_run_manager:
            docs = await self._aget_docs(new_question, inputs, run_manager=_run_manager)
        else:
            docs = await self._aget_docs(new_question, inputs)  # type: ignore[call-arg]

        output: dict[str, Any] = {}
        if self.response_if_no_docs_found is not None and len(docs) == 0:
            output[self.output_key] = self.response_if_no_docs_found
        else:
            new_inputs = inputs.copy()
            if self.rephrase_question:
                new_inputs["question"] = new_question
            new_inputs["chat_history"] = chat_history_str
            answer = await self.combine_docs_chain.arun(
                input_documents=docs,
                callbacks=_run_manager.get_child(),
                **new_inputs,
            )
            output[self.output_key] = answer

        if self.return_source_documents:
            output["source_documents"] = docs
        if self.return_generated_question:
            output["generated_question"] = new_question
        return output