RunnableAgent Class — langchain Architecture

Architecture documentation for the RunnableAgent class in agent.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  d4aecf89_003b_4fb4_2c03_b614dbd33066["RunnableAgent"]
  2066c331_3189_1880_ba09_f5a3c375c553["BaseSingleActionAgent"]
  d4aecf89_003b_4fb4_2c03_b614dbd33066 -->|extends| 2066c331_3189_1880_ba09_f5a3c375c553
  0faae8c7_2812_be15_1073_b6537539cea8["agent.py"]
  d4aecf89_003b_4fb4_2c03_b614dbd33066 -->|defined in| 0faae8c7_2812_be15_1073_b6537539cea8
  aaac601d_c109_eda6_fdeb_34d4074dae1e["return_values()"]
  d4aecf89_003b_4fb4_2c03_b614dbd33066 -->|method| aaac601d_c109_eda6_fdeb_34d4074dae1e
  50066575_71e0_ce1f_91fe_d5004e62efe6["input_keys()"]
  d4aecf89_003b_4fb4_2c03_b614dbd33066 -->|method| 50066575_71e0_ce1f_91fe_d5004e62efe6
  f3744a6a_d6e6_3de9_26d0_166fe46f3862["plan()"]
  d4aecf89_003b_4fb4_2c03_b614dbd33066 -->|method| f3744a6a_d6e6_3de9_26d0_166fe46f3862
  2930f93d_205d_ca04_bcf2_65e6c3a766a2["aplan()"]
  d4aecf89_003b_4fb4_2c03_b614dbd33066 -->|method| 2930f93d_205d_ca04_bcf2_65e6c3a766a2

Relationship Graph

Source Code

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

class RunnableAgent(BaseSingleActionAgent):
    """Agent powered by Runnables."""

    runnable: Runnable[dict, AgentAction | AgentFinish]
    """Runnable to call to get agent action."""
    input_keys_arg: list[str] = []
    return_keys_arg: list[str] = []
    stream_runnable: bool = True
    """Whether to stream from the runnable or not.

    If `True` then 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`. If `False` then LLM is invoked in a non-streaming fashion and
        individual LLM tokens will not be available in stream_log.
    """

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
    )

    @property
    def return_values(self) -> list[str]:
        """Return values of the agent."""
        return self.return_keys_arg

    @property
    def input_keys(self) -> list[str]:
        """Return the input keys."""
        return self.input_keys_arg

    def plan(
        self,
        intermediate_steps: list[tuple[AgentAction, str]],
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> 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

    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: