test_astream_fallback_to_ainvoke() — langchain Function Reference

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

Function python CoreAbstractions RunnableInterface calls 2

Entity Profile

CoreAbstractions→ RunnableInterface→ test_astream_fallback_to_ainvoke() — langchain Function Reference

Dependency Diagram

graph TD
  7509db04_79f0_aba1_2f13_5541131874c8["test_astream_fallback_to_ainvoke()"]
  4404310a_b16c_e9e6_1de2_90e5ea6f3ce3["test_base.py"]
  7509db04_79f0_aba1_2f13_5541131874c8 -->|defined in| 4404310a_b16c_e9e6_1de2_90e5ea6f3ce3
  6fde93f4_513d_1a3f_701f_1a4d762ee752["_generate()"]
  7509db04_79f0_aba1_2f13_5541131874c8 -->|calls| 6fde93f4_513d_1a3f_701f_1a4d762ee752
  5ddb6fca_46b7_b6c6_10aa_c263c14caa90["_llm_type()"]
  7509db04_79f0_aba1_2f13_5541131874c8 -->|calls| 5ddb6fca_46b7_b6c6_10aa_c263c14caa90
  style 7509db04_79f0_aba1_2f13_5541131874c8 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/language_models/llms/test_base.py lines 137–161

async def test_astream_fallback_to_ainvoke() -> None:
    """Test astream uses appropriate implementation."""

    class ModelWithGenerate(BaseLLM):
        @override
        def _generate(
            self,
            prompts: list[str],
            stop: list[str] | None = None,
            run_manager: CallbackManagerForLLMRun | None = None,
            **kwargs: Any,
        ) -> LLMResult:
            generations = [Generation(text="hello")]
            return LLMResult(generations=[generations])

        @property
        def _llm_type(self) -> str:
            return "fake-chat-model"

    model = ModelWithGenerate()
    chunks = list(model.stream("anything"))
    assert chunks == ["hello"]

    chunks = [chunk async for chunk in model.astream("anything")]
    assert chunks == ["hello"]