test_simple_sequential_functionality_with_callbacks() — langchain Function Reference

Source Code

libs/langchain/tests/unit_tests/chains/test_sequential.py lines 209–242

async def test_simple_sequential_functionality_with_callbacks(
    *, is_async: bool
) -> None:
    """Test simple sequential functionality."""
    handler_1 = FakeCallbackHandler()
    handler_2 = FakeCallbackHandler()
    handler_3 = FakeCallbackHandler()
    chain_1 = FakeChain(
        input_variables=["foo"],
        output_variables=["bar"],
        callbacks=[handler_1],
    )
    chain_2 = FakeChain(
        input_variables=["bar"],
        output_variables=["baz"],
        callbacks=[handler_2],
    )
    chain_3 = FakeChain(
        input_variables=["jack"],
        output_variables=["baf"],
        callbacks=[handler_3],
    )
    chain = SimpleSequentialChain(chains=[chain_1, chain_2, chain_3])
    if is_async:
        output = await chain.ainvoke({"input": "123"})
    else:
        output = chain({"input": "123"})
    expected_output = {"output": "123foofoofoo", "input": "123"}
    assert output == expected_output
    # Check that each of the callbacks were invoked once per the entire run
    for handler in [handler_1, handler_2, handler_3]:
        assert handler.starts == 1
        assert handler.ends == 1
        assert handler.errors == 0