_astream() — langchain Function Reference

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

Function python LangChainCore LanguageModelBase calls 1 called by 1

Entity Profile

LangChainCore→ LanguageModelBase→ _astream() — langchain Function Reference

Dependency Diagram

graph TD
  cde77cc5_1e1f_9bb4_ed01_e6c99050c598["_astream()"]
  e6b830e0_e492_47f6_6d0a_65d69c36da10["HuggingFaceEndpoint"]
  cde77cc5_1e1f_9bb4_ed01_e6c99050c598 -->|defined in| e6b830e0_e492_47f6_6d0a_65d69c36da10
  56f985d3_9992_f0cb_ad40_e11190fde3bb["_acall()"]
  56f985d3_9992_f0cb_ad40_e11190fde3bb -->|calls| cde77cc5_1e1f_9bb4_ed01_e6c99050c598
  d3d02905_01f3_3fb8_f596_5694650af72c["_invocation_params()"]
  cde77cc5_1e1f_9bb4_ed01_e6c99050c598 -->|calls| d3d02905_01f3_3fb8_f596_5694650af72c
  style cde77cc5_1e1f_9bb4_ed01_e6c99050c598 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/partners/huggingface/langchain_huggingface/llms/huggingface_endpoint.py lines 421–455

    async def _astream(
        self,
        prompt: str,
        stop: list[str] | None = None,
        run_manager: AsyncCallbackManagerForLLMRun | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[GenerationChunk]:
        invocation_params = self._invocation_params(stop, **kwargs)
        async for response in await self.async_client.text_generation(
            prompt, **invocation_params, stream=True
        ):
            # identify stop sequence in generated text, if any
            stop_seq_found: str | None = None
            for stop_seq in invocation_params["stop"]:
                if stop_seq in response:
                    stop_seq_found = stop_seq

            # identify text to yield
            text: str | None = None
            if stop_seq_found:
                text = response[: response.index(stop_seq_found)]
            else:
                text = response

            # yield text, if any
            if text:
                chunk = GenerationChunk(text=text)

                if run_manager:
                    await run_manager.on_llm_new_token(chunk.text)
                yield chunk

            # break if stop sequence found
            if stop_seq_found:
                break