AgentExecutorIterator Class — langchain Architecture

Source Code

libs/langchain/langchain_classic/agents/agent_iterator.py lines 40–432

class AgentExecutorIterator:
    """Iterator for AgentExecutor."""

    def __init__(
        self,
        agent_executor: AgentExecutor,
        inputs: Any,
        callbacks: Callbacks = None,
        *,
        tags: list[str] | None = None,
        metadata: dict[str, Any] | None = None,
        run_name: str | None = None,
        run_id: UUID | None = None,
        include_run_info: bool = False,
        yield_actions: bool = False,
    ):
        """Initialize the `AgentExecutorIterator`.

        Initialize the `AgentExecutorIterator` with the given `AgentExecutor`,
        inputs, and optional callbacks.

        Args:
            agent_executor: The `AgentExecutor` to iterate over.
            inputs: The inputs to the `AgentExecutor`.
            callbacks: The callbacks to use during iteration.
            tags: The tags to use during iteration.
            metadata: The metadata to use during iteration.
            run_name: The name of the run.
            run_id: The ID of the run.
            include_run_info: Whether to include run info in the output.
            yield_actions: Whether to yield actions as they are generated.
        """
        self._agent_executor = agent_executor
        self.inputs = inputs
        self.callbacks = callbacks
        self.tags = tags
        self.metadata = metadata
        self.run_name = run_name
        self.run_id = run_id
        self.include_run_info = include_run_info
        self.yield_actions = yield_actions
        self.reset()

    _inputs: dict[str, str]
    callbacks: Callbacks
    tags: list[str] | None
    metadata: dict[str, Any] | None
    run_name: str | None
    run_id: UUID | None
    include_run_info: bool
    yield_actions: bool

    @property
    def inputs(self) -> dict[str, str]:
        """The inputs to the `AgentExecutor`."""
        return self._inputs

    @inputs.setter
    def inputs(self, inputs: Any) -> None:
        self._inputs = self.agent_executor.prep_inputs(inputs)

    @property
    def agent_executor(self) -> AgentExecutor:
        """The `AgentExecutor` to iterate over."""
        return self._agent_executor

    @agent_executor.setter
    def agent_executor(self, agent_executor: AgentExecutor) -> None:
        self._agent_executor = agent_executor
        # force re-prep inputs in case agent_executor's prep_inputs fn changed
        self.inputs = self.inputs

    @property
    def name_to_tool_map(self) -> dict[str, BaseTool]:
        """A mapping of tool names to tools."""
        return {tool.name: tool for tool in self.agent_executor.tools}

    @property
    def color_mapping(self) -> dict[str, str]:
        """A mapping of tool names to colors."""
        return get_color_mapping(