plan() — langchain Function Reference

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

Function python AgentOrchestration ToolInterface calls 2

Entity Profile

AgentOrchestration→ ToolInterface→ plan() — langchain Function Reference

Dependency Diagram

graph TD
  dd36f14f_cc3c_6f86_0cda_844be39584a3["plan()"]
  43de2fb1_2cba_0c7f_1a66_b91200c3611f["RunnableMultiActionAgent"]
  dd36f14f_cc3c_6f86_0cda_844be39584a3 -->|defined in| 43de2fb1_2cba_0c7f_1a66_b91200c3611f
  fff15887_2a1c_dee2_d5a0_3a5aa4536868["stream()"]
  dd36f14f_cc3c_6f86_0cda_844be39584a3 -->|calls| fff15887_2a1c_dee2_d5a0_3a5aa4536868
  ff3ae374_f735_73ee_26e0_538ba433911c["plan()"]
  dd36f14f_cc3c_6f86_0cda_844be39584a3 -->|calls| ff3ae374_f735_73ee_26e0_538ba433911c
  style dd36f14f_cc3c_6f86_0cda_844be39584a3 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain/langchain_classic/agents/agent.py lines 531–565

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

        Args:
            intermediate_steps: Steps the LLM has taken to date,
                along with the 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.
            for chunk in self.runnable.stream(inputs, config={"callbacks": callbacks}):
                if final_output is None:
                    final_output = chunk
                else:
                    final_output += chunk
        else:
            final_output = self.runnable.invoke(inputs, config={"callbacks": callbacks})

        return final_output