test_stream_error_callback() — langchain Function Reference

Architecture documentation for the test_stream_error_callback() function in test_base.py from the langchain codebase.

Function python CoreAbstractions MessageSchema

Entity Profile

CoreAbstractions→ MessageSchema→ test_stream_error_callback() — langchain Function Reference

Dependency Diagram

graph TD
  f3e5df8e_96ac_e58c_d6b7_74882da12cdf["test_stream_error_callback()"]
  8830054d_ac1e_daa9_c6c5_ff55b10d0bf3["test_base.py"]
  f3e5df8e_96ac_e58c_d6b7_74882da12cdf -->|defined in| 8830054d_ac1e_daa9_c6c5_ff55b10d0bf3
  style f3e5df8e_96ac_e58c_d6b7_74882da12cdf fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/language_models/chat_models/test_base.py lines 149–180

async def test_stream_error_callback() -> None:
    message = "test"

    def eval_response(callback: BaseFakeCallbackHandler, i: int) -> None:
        assert callback.errors == 1
        assert len(callback.errors_args) == 1
        llm_result: LLMResult = callback.errors_args[0]["kwargs"]["response"]
        if i == 0:
            assert llm_result.generations == []
        else:
            assert llm_result.generations[0][0].text == message[:i]

    for i in range(len(message)):
        llm = FakeListChatModel(
            responses=[message],
            error_on_chunk_number=i,
        )
        cb_async = FakeAsyncCallbackHandler()
        llm_astream = llm.astream("Dummy message", config={"callbacks": [cb_async]})
        for _ in range(i):
            await anext(llm_astream)
        with pytest.raises(FakeListChatModelError):
            await anext(llm_astream)
        eval_response(cb_async, i)

        cb_sync = FakeCallbackHandler()
        llm_stream = llm.stream("Dumy message", config={"callbacks": [cb_sync]})
        for _ in range(i):
            next(llm_stream)
        with pytest.raises(FakeListChatModelError):
            next(llm_stream)
        eval_response(cb_sync, i)