_aperform_agent_action() — langchain Function Reference

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

Function python AgentOrchestration ActionLogic calls 1 called by 1

Entity Profile

AgentOrchestration→ ActionLogic→ _aperform_agent_action() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/langchain/langchain_classic/agents/agent.py lines 1527–1568

    async def _aperform_agent_action(
        self,
        name_to_tool_map: dict[str, BaseTool],
        color_mapping: dict[str, str],
        agent_action: AgentAction,
        run_manager: AsyncCallbackManagerForChainRun | None = None,
    ) -> AgentStep:
        if run_manager:
            await run_manager.on_agent_action(
                agent_action,
                verbose=self.verbose,
                color="green",
            )
        # Otherwise we lookup the tool
        if agent_action.tool in name_to_tool_map:
            tool = name_to_tool_map[agent_action.tool]
            return_direct = tool.return_direct
            color = color_mapping[agent_action.tool]
            tool_run_kwargs = self._action_agent.tool_run_logging_kwargs()
            if return_direct:
                tool_run_kwargs["llm_prefix"] = ""
            # We then call the tool on the tool input to get an observation
            observation = await tool.arun(
                agent_action.tool_input,
                verbose=self.verbose,
                color=color,
                callbacks=run_manager.get_child() if run_manager else None,
                **tool_run_kwargs,
            )
        else:
            tool_run_kwargs = self._action_agent.tool_run_logging_kwargs()
            observation = await InvalidTool().arun(
                {
                    "requested_tool_name": agent_action.tool,
                    "available_tool_names": list(name_to_tool_map.keys()),
                },
                verbose=self.verbose,
                color=None,
                callbacks=run_manager.get_child() if run_manager else None,
                **tool_run_kwargs,
            )
        return AgentStep(action=agent_action, observation=observation)