Source code for langchain.agents.agent_iterator
from __future__ import annotations
import logging
import time
from abc import ABC, abstractmethod
from asyncio import CancelledError
from functools import wraps
from typing import (
    TYPE_CHECKING,
    Any,
    Callable,
    Dict,
    List,
    NoReturn,
    Optional,
    Tuple,
    Type,
    Union,
)
from langchain.callbacks.manager import (
    AsyncCallbackManager,
    AsyncCallbackManagerForChainRun,
    CallbackManager,
    CallbackManagerForChainRun,
    Callbacks,
)
from langchain.load.dump import dumpd
from langchain.schema import RUN_KEY, AgentAction, AgentFinish, RunInfo
from langchain.tools import BaseTool
from langchain.utilities.asyncio import asyncio_timeout
from langchain.utils.input import get_color_mapping
if TYPE_CHECKING:
    from langchain.agents.agent import AgentExecutor
logger = logging.getLogger(__name__)
[docs]class BaseAgentExecutorIterator(ABC):
    """Base class for AgentExecutorIterator."""
[docs]    @abstractmethod
    def build_callback_manager(self) -> None:
        pass
[docs]def rebuild_callback_manager_on_set(
    setter_method: Callable[..., None]
) -> Callable[..., None]:
    """Decorator to force setters to rebuild callback mgr"""
    @wraps(setter_method)
    def wrapper(self: BaseAgentExecutorIterator, *args: Any, **kwargs: Any) -> None:
        setter_method(self, *args, **kwargs)
        self.build_callback_manager()
    return wrapper
[docs]class AgentExecutorIterator(BaseAgentExecutorIterator):
    """Iterator for AgentExecutor."""
[docs]    def __init__(
        self,
        agent_executor: AgentExecutor,
        inputs: Any,