_call() — langchain Function Reference

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

Function python AgentOrchestration ClassicChains calls 3

Entity Profile

AgentOrchestration→ ClassicChains→ _call() — langchain Function Reference

Dependency Diagram

graph TD
  16ea3b30_e483_d5fc_b053_6173b3e0b010["_call()"]
  daa08d88_1d2b_5650_6257_d688074121b7["ElasticsearchDatabaseChain"]
  16ea3b30_e483_d5fc_b053_6173b3e0b010 -->|defined in| daa08d88_1d2b_5650_6257_d688074121b7
  39259671_095c_7a18_7716_e943cacf5276["_list_indices()"]
  16ea3b30_e483_d5fc_b053_6173b3e0b010 -->|calls| 39259671_095c_7a18_7716_e943cacf5276
  bb96d282_1bc8_a6dc_3a48_1083a4096d66["_get_indices_infos()"]
  16ea3b30_e483_d5fc_b053_6173b3e0b010 -->|calls| bb96d282_1bc8_a6dc_3a48_1083a4096d66
  243f4640_a981_e59e_8932_e88ee66a31de["_search()"]
  16ea3b30_e483_d5fc_b053_6173b3e0b010 -->|calls| 243f4640_a981_e59e_8932_e88ee66a31de
  style 16ea3b30_e483_d5fc_b053_6173b3e0b010 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain/langchain_classic/chains/elasticsearch_database/base.py lines 116–170

    def _call(
        self,
        inputs: dict[str, Any],
        run_manager: CallbackManagerForChainRun | None = None,
    ) -> dict[str, Any]:
        _run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager()
        input_text = f"{inputs[self.input_key]}\nESQuery:"
        _run_manager.on_text(input_text, verbose=self.verbose)
        indices = self._list_indices()
        indices_info = self._get_indices_infos(indices)
        query_inputs: dict = {
            "input": input_text,
            "top_k": str(self.top_k),
            "indices_info": indices_info,
            "stop": ["\nESResult:"],
        }
        intermediate_steps: list = []
        try:
            intermediate_steps.append(query_inputs)  # input: es generation
            es_cmd = self.query_chain.invoke(
                query_inputs,
                config={"callbacks": _run_manager.get_child()},
            )

            _run_manager.on_text(es_cmd, color="green", verbose=self.verbose)
            intermediate_steps.append(
                es_cmd,
            )  # output: elasticsearch dsl generation (no checker)
            intermediate_steps.append({"es_cmd": es_cmd})  # input: ES search
            result = self._search(indices=indices, query=es_cmd)
            intermediate_steps.append(str(result))  # output: ES search

            _run_manager.on_text("\nESResult: ", verbose=self.verbose)
            _run_manager.on_text(result, color="yellow", verbose=self.verbose)

            _run_manager.on_text("\nAnswer:", verbose=self.verbose)
            answer_inputs: dict = {"data": result, "input": input_text}
            intermediate_steps.append(answer_inputs)  # input: final answer
            final_result = self.answer_chain.invoke(
                answer_inputs,
                config={"callbacks": _run_manager.get_child()},
            )

            intermediate_steps.append(final_result)  # output: final answer
            _run_manager.on_text(final_result, color="green", verbose=self.verbose)
            chain_result: dict[str, Any] = {self.output_key: final_result}
            if self.return_intermediate_steps:
                chain_result[INTERMEDIATE_STEPS_KEY] = intermediate_steps
        except Exception as exc:
            # Append intermediate steps to exception, to aid in logging and later
            # improvement of few shot prompt seeds
            exc.intermediate_steps = intermediate_steps  # type: ignore[attr-defined]
            raise

        return chain_result