_astream() — langchain Function Reference

Architecture documentation for the _astream() function in fake_chat_models.py from the langchain codebase.

Function python LangChainCore LanguageModelBase

Entity Profile

LangChainCore→ LanguageModelBase→ _astream() — langchain Function Reference

Dependency Diagram

graph TD
  4aca5160_857e_b75e_f95a_9c6c1a4db44a["_astream()"]
  2790c9e0_a72e_5aac_c2c7_6702139c50e3["FakeListChatModel"]
  4aca5160_857e_b75e_f95a_9c6c1a4db44a -->|defined in| 2790c9e0_a72e_5aac_c2c7_6702139c50e3
  style 4aca5160_857e_b75e_f95a_9c6c1a4db44a fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/language_models/fake_chat_models.py lines 124–149

    async def _astream(
        self,
        messages: list[BaseMessage],
        stop: list[str] | None = None,
        run_manager: AsyncCallbackManagerForLLMRun | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[ChatGenerationChunk]:
        response = self.responses[self.i]
        if self.i < len(self.responses) - 1:
            self.i += 1
        else:
            self.i = 0
        for i_c, c in enumerate(response):
            if self.sleep is not None:
                await asyncio.sleep(self.sleep)
            if (
                self.error_on_chunk_number is not None
                and i_c == self.error_on_chunk_number
            ):
                raise FakeListChatModelError
            chunk_position: Literal["last"] | None = (
                "last" if i_c == len(response) - 1 else None
            )
            yield ChatGenerationChunk(
                message=AIMessageChunk(content=c, chunk_position=chunk_position)
            )