_acall() — langchain Function Reference

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

Function python AgentOrchestration ClassicChains calls 1 called by 1

Entity Profile

AgentOrchestration→ ClassicChains→ _acall() — langchain Function Reference

Dependency Diagram

graph TD
  ccc83700_c56e_645f_a5ce_9e44f26e609a["_acall()"]
  b9a82d83_700a_c3c4_a4fd_29ab9cbe4112["APIChain"]
  ccc83700_c56e_645f_a5ce_9e44f26e609a -->|defined in| b9a82d83_700a_c3c4_a4fd_29ab9cbe4112
  0393cea2_16ff_2ed6_ebe5_2e7f4968843e["langchain_community()"]
  0393cea2_16ff_2ed6_ebe5_2e7f4968843e -->|calls| ccc83700_c56e_645f_a5ce_9e44f26e609a
  357f19d8_3422_6a9d_533d_c62f93023ce8["_check_in_allowed_domain()"]
  ccc83700_c56e_645f_a5ce_9e44f26e609a -->|calls| 357f19d8_3422_6a9d_533d_c62f93023ce8
  style ccc83700_c56e_645f_a5ce_9e44f26e609a fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain/langchain_classic/chains/api/base.py lines 315–358

        async def _acall(
            self,
            inputs: dict[str, Any],
            run_manager: AsyncCallbackManagerForChainRun | None = None,
        ) -> dict[str, str]:
            _run_manager = (
                run_manager or AsyncCallbackManagerForChainRun.get_noop_manager()
            )
            question = inputs[self.question_key]
            api_url = await self.api_request_chain.apredict(
                question=question,
                api_docs=self.api_docs,
                callbacks=_run_manager.get_child(),
            )
            await _run_manager.on_text(
                api_url,
                color="green",
                end="\n",
                verbose=self.verbose,
            )
            api_url = api_url.strip()
            if self.limit_to_domains and not _check_in_allowed_domain(
                api_url,
                self.limit_to_domains,
            ):
                msg = (
                    f"{api_url} is not in the allowed domains: {self.limit_to_domains}"
                )
                raise ValueError(msg)
            api_response = await self.requests_wrapper.aget(api_url)
            await _run_manager.on_text(
                str(api_response),
                color="yellow",
                end="\n",
                verbose=self.verbose,
            )
            answer = await self.api_answer_chain.apredict(
                question=question,
                api_docs=self.api_docs,
                api_url=api_url,
                api_response=api_response,
                callbacks=_run_manager.get_child(),
            )
            return {self.output_key: answer}