_iter_next_step() — langchain Function Reference

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

Function python AgentOrchestration AgentExcecutor calls 3 called by 1

Entity Profile

AgentOrchestration→ AgentExcecutor→ _iter_next_step() — langchain Function Reference

Dependency Diagram

graph TD
  9dcff3d2_e93f_4d1d_86be_c6326804d56b["_iter_next_step()"]
  ec1d7866_e6a8_7cbe_b150_f82463cf2c7f["AgentExecutor"]
  9dcff3d2_e93f_4d1d_86be_c6326804d56b -->|defined in| ec1d7866_e6a8_7cbe_b150_f82463cf2c7f
  37c42b3f_8517_1cf2_c825_b42e1f600be2["_take_next_step()"]
  37c42b3f_8517_1cf2_c825_b42e1f600be2 -->|calls| 9dcff3d2_e93f_4d1d_86be_c6326804d56b
  d1902c5d_fa4e_fa37_a8be_9cad6be0fbad["_prepare_intermediate_steps()"]
  9dcff3d2_e93f_4d1d_86be_c6326804d56b -->|calls| d1902c5d_fa4e_fa37_a8be_9cad6be0fbad
  8039fcf3_65c7_db8a_01b1_7048ea1e72b3["_perform_agent_action()"]
  9dcff3d2_e93f_4d1d_86be_c6326804d56b -->|calls| 8039fcf3_65c7_db8a_01b1_7048ea1e72b3
  84b1edcc_b4d7_5498_baf5_6ae75dc68ba8["tool_run_logging_kwargs()"]
  9dcff3d2_e93f_4d1d_86be_c6326804d56b -->|calls| 84b1edcc_b4d7_5498_baf5_6ae75dc68ba8
  style 9dcff3d2_e93f_4d1d_86be_c6326804d56b fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain/langchain_classic/agents/agent.py lines 1301–1378

    def _iter_next_step(
        self,
        name_to_tool_map: dict[str, BaseTool],
        color_mapping: dict[str, str],
        inputs: dict[str, str],
        intermediate_steps: list[tuple[AgentAction, str]],
        run_manager: CallbackManagerForChainRun | None = None,
    ) -> Iterator[AgentFinish | AgentAction | AgentStep]:
        """Take a single step in the thought-action-observation loop.

        Override this to take control of how the agent makes and acts on choices.
        """
        try:
            intermediate_steps = self._prepare_intermediate_steps(intermediate_steps)

            # Call the LLM to see what to do.
            output = self._action_agent.plan(
                intermediate_steps,
                callbacks=run_manager.get_child() if run_manager else None,
                **inputs,
            )
        except OutputParserException as e:
            if isinstance(self.handle_parsing_errors, bool):
                raise_error = not self.handle_parsing_errors
            else:
                raise_error = False
            if raise_error:
                msg = (
                    "An output parsing error occurred. "
                    "In order to pass this error back to the agent and have it try "
                    "again, pass `handle_parsing_errors=True` to the AgentExecutor. "
                    f"This is the error: {e!s}"
                )
                raise ValueError(msg) from e
            text = str(e)
            if isinstance(self.handle_parsing_errors, bool):
                if e.send_to_llm:
                    observation = str(e.observation)
                    text = str(e.llm_output)
                else:
                    observation = "Invalid or incomplete response"
            elif isinstance(self.handle_parsing_errors, str):
                observation = self.handle_parsing_errors
            elif callable(self.handle_parsing_errors):
                observation = self.handle_parsing_errors(e)
            else:
                msg = "Got unexpected type of `handle_parsing_errors`"  # type: ignore[unreachable]
                raise ValueError(msg) from e  # noqa: TRY004
            output = AgentAction("_Exception", observation, text)
            if run_manager:
                run_manager.on_agent_action(output, color="green")
            tool_run_kwargs = self._action_agent.tool_run_logging_kwargs()
            observation = ExceptionTool().run(
                output.tool_input,
                verbose=self.verbose,
                color=None,
                callbacks=run_manager.get_child() if run_manager else None,
                **tool_run_kwargs,
            )
            yield AgentStep(action=output, observation=observation)
            return

        # If the tool chosen is the finishing tool, then we end and return.
        if isinstance(output, AgentFinish):
            yield output
            return

        actions: list[AgentAction]
        actions = [output] if isinstance(output, AgentAction) else output
        for agent_action in actions:
            yield agent_action
        for agent_action in actions:
            yield self._perform_agent_action(
                name_to_tool_map,
                color_mapping,
                agent_action,
                run_manager,
            )