test_agent_iterator.py — langchain Source File

Source Code

from uuid import UUID

import pytest
from langchain_core.language_models import FakeListLLM
from langchain_core.tools import Tool
from langchain_core.tracers.context import collect_runs

from langchain_classic.agents import (
    AgentExecutor,
    AgentExecutorIterator,
    AgentType,
    initialize_agent,
)
from langchain_classic.schema import RUN_KEY
from tests.unit_tests.agents.test_agent import _get_agent
from tests.unit_tests.callbacks.fake_callback_handler import FakeCallbackHandler


def test_agent_iterator_bad_action() -> None:
    """Test react chain iterator when bad action given."""
    agent = _get_agent()
    agent_iter = agent.iter(inputs="when was langchain made")

    outputs = list(agent_iter)

    assert isinstance(outputs[-1], dict)
    assert outputs[-1]["output"] == "curses foiled again"


def test_agent_iterator_stopped_early() -> None:
    """Test react chain iterator when stopped early.

    Test react chain iterator when max iterations or
    max execution time is exceeded.
    """
    # iteration limit
    agent = _get_agent(max_iterations=1)
    agent_iter = agent.iter(inputs="when was langchain made")

    outputs = list(agent_iter)
    # NOTE: we don't use agent.run like in the test for the regular agent executor,
    # so the dict structure for outputs stays intact
    assert isinstance(outputs[-1], dict)
    assert (
        outputs[-1]["output"] == "Agent stopped due to iteration limit or time limit."
    )

    # execution time limit
    agent = _get_agent(max_execution_time=1e-5)
    agent_iter = agent.iter(inputs="when was langchain made")

    outputs = []
    for step in agent_iter:
        outputs.append(step)
    assert isinstance(outputs[-1], dict)
    assert (
        outputs[-1]["output"] == "Agent stopped due to iteration limit or time limit."
    )


// ... (327 more lines)