_aiter_next_step() — langchain Function Reference

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

Function python AgentOrchestration ToolInterface calls 3 called by 1

Entity Profile

AgentOrchestration→ ToolInterface→ _aiter_next_step() — langchain Function Reference

Dependency Diagram

graph TD
  c61b7dc2_6e22_7735_1fa0_744874e0c701["_aiter_next_step()"]
  493d4ce4_2303_08e0_2337_2bee34fe2662["AgentExecutor"]
  c61b7dc2_6e22_7735_1fa0_744874e0c701 -->|defined in| 493d4ce4_2303_08e0_2337_2bee34fe2662
  1ddbcac4_c10b_0975_4498_6162d936691c["_atake_next_step()"]
  1ddbcac4_c10b_0975_4498_6162d936691c -->|calls| c61b7dc2_6e22_7735_1fa0_744874e0c701
  ec593113_cca6_1902_3a34_7e9c8b098e7b["_prepare_intermediate_steps()"]
  c61b7dc2_6e22_7735_1fa0_744874e0c701 -->|calls| ec593113_cca6_1902_3a34_7e9c8b098e7b
  85fa3938_f549_0ce3_b3f9_68b32bbb3d27["_aperform_agent_action()"]
  c61b7dc2_6e22_7735_1fa0_744874e0c701 -->|calls| 85fa3938_f549_0ce3_b3f9_68b32bbb3d27
  c88bc836_f328_3f17_0fd2_8dcc7203a6ca["tool_run_logging_kwargs()"]
  c61b7dc2_6e22_7735_1fa0_744874e0c701 -->|calls| c88bc836_f328_3f17_0fd2_8dcc7203a6ca
  style c61b7dc2_6e22_7735_1fa0_744874e0c701 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain/langchain_classic/agents/agent.py lines 1440–1525

    async def _aiter_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: AsyncCallbackManagerForChainRun | None = None,
    ) -> AsyncIterator[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 = await self._action_agent.aplan(
                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)
            tool_run_kwargs = self._action_agent.tool_run_logging_kwargs()
            observation = await ExceptionTool().arun(
                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

        # Use asyncio.gather to run multiple tool.arun() calls concurrently
        result = await asyncio.gather(
            *[
                self._aperform_agent_action(
                    name_to_tool_map,
                    color_mapping,
                    agent_action,
                    run_manager,
                )
                for agent_action in actions
            ],