aplan() — langchain Function Reference

Architecture documentation for the aplan() function in agent.py from the langchain codebase.

Function python AgentOrchestration ActionLogic calls 2 called by 1

Entity Profile

AgentOrchestration→ ActionLogic→ aplan() — langchain Function Reference

Dependency Diagram

graph TD
  2930f93d_205d_ca04_bcf2_65e6c3a766a2["aplan()"]
  d4aecf89_003b_4fb4_2c03_b614dbd33066["RunnableAgent"]
  2930f93d_205d_ca04_bcf2_65e6c3a766a2 -->|defined in| d4aecf89_003b_4fb4_2c03_b614dbd33066
  5741fa2f_805c_1a46_0802_547d3beef2d0["aplan()"]
  5741fa2f_805c_1a46_0802_547d3beef2d0 -->|calls| 2930f93d_205d_ca04_bcf2_65e6c3a766a2
  330b61d2_ee14_f0e8_72b9_0f8c88b2a488["astream()"]
  2930f93d_205d_ca04_bcf2_65e6c3a766a2 -->|calls| 330b61d2_ee14_f0e8_72b9_0f8c88b2a488
  5741fa2f_805c_1a46_0802_547d3beef2d0["aplan()"]
  2930f93d_205d_ca04_bcf2_65e6c3a766a2 -->|calls| 5741fa2f_805c_1a46_0802_547d3beef2d0
  style 2930f93d_205d_ca04_bcf2_65e6c3a766a2 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain/langchain_classic/agents/agent.py lines 455–494

    async def aplan(
        self,
        intermediate_steps: list[tuple[AgentAction, str]],
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> AgentAction | AgentFinish:
        """Async based on past history and current inputs, decide what to do.

        Args:
            intermediate_steps: Steps the LLM has taken to date,
                along with observations.
            callbacks: Callbacks to run.
            **kwargs: User inputs.

        Returns:
            Action specifying what tool to use.
        """
        inputs = {**kwargs, "intermediate_steps": intermediate_steps}
        final_output: Any = None
        if self.stream_runnable:
            # Use streaming to make sure that the underlying LLM is invoked in a
            # streaming
            # fashion to make it possible to get access to the individual LLM tokens
            # when using stream_log with the AgentExecutor.
            # Because the response from the plan is not a generator, we need to
            # accumulate the output into final output and return that.
            async for chunk in self.runnable.astream(
                inputs,
                config={"callbacks": callbacks},
            ):
                if final_output is None:
                    final_output = chunk
                else:
                    final_output += chunk
        else:
            final_output = await self.runnable.ainvoke(
                inputs,
                config={"callbacks": callbacks},
            )
        return final_output