_astream() — langchain Function Reference

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

Function python LangChainCore LanguageModelBase calls 2 called by 1

Entity Profile

LangChainCore→ LanguageModelBase→ _astream() — langchain Function Reference

Dependency Diagram

graph TD
  82b5a500_280a_9bd8_9cfb_8d941c28f7c6["_astream()"]
  d5ca3c3a_3c29_0cb2_a156_35c92a31f5fd["ChatGroq"]
  82b5a500_280a_9bd8_9cfb_8d941c28f7c6 -->|defined in| d5ca3c3a_3c29_0cb2_a156_35c92a31f5fd
  a59adcb5_ffbb_8e47_2ed1_52a3de26c52e["_agenerate()"]
  a59adcb5_ffbb_8e47_2ed1_52a3de26c52e -->|calls| 82b5a500_280a_9bd8_9cfb_8d941c28f7c6
  4149b891_9f6b_d1d3_0159_d2a1de034bf3["_create_message_dicts()"]
  82b5a500_280a_9bd8_9cfb_8d941c28f7c6 -->|calls| 4149b891_9f6b_d1d3_0159_d2a1de034bf3
  943d2ee7_1f7a_632d_e35f_f168f64894fe["_convert_chunk_to_message_chunk()"]
  82b5a500_280a_9bd8_9cfb_8d941c28f7c6 -->|calls| 943d2ee7_1f7a_632d_e35f_f168f64894fe
  style 82b5a500_280a_9bd8_9cfb_8d941c28f7c6 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/partners/groq/langchain_groq/chat_models.py lines 697–751

    async def _astream(
        self,
        messages: list[BaseMessage],
        stop: list[str] | None = None,
        run_manager: AsyncCallbackManagerForLLMRun | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[ChatGenerationChunk]:
        message_dicts, params = self._create_message_dicts(messages, stop)

        params = {**params, **kwargs, "stream": True}

        default_chunk_class: type[BaseMessageChunk] = AIMessageChunk
        async for chunk in await self.async_client.create(
            messages=message_dicts, **params
        ):
            if not isinstance(chunk, dict):
                chunk = chunk.model_dump()  # noqa: PLW2901
            if len(chunk["choices"]) == 0:
                continue
            choice = chunk["choices"][0]
            message_chunk = _convert_chunk_to_message_chunk(chunk, default_chunk_class)
            generation_info = {}
            if finish_reason := choice.get("finish_reason"):
                generation_info["finish_reason"] = finish_reason
                generation_info["model_name"] = self.model_name
                if system_fingerprint := chunk.get("system_fingerprint"):
                    generation_info["system_fingerprint"] = system_fingerprint
                service_tier = params.get("service_tier") or self.service_tier
                generation_info["service_tier"] = service_tier
                reasoning_effort = (
                    params.get("reasoning_effort") or self.reasoning_effort
                )
                if reasoning_effort:
                    generation_info["reasoning_effort"] = reasoning_effort
            logprobs = choice.get("logprobs")
            if logprobs:
                generation_info["logprobs"] = logprobs

            if generation_info:
                message_chunk = message_chunk.model_copy(
                    update={"response_metadata": generation_info}
                )

            default_chunk_class = message_chunk.__class__
            generation_chunk = ChatGenerationChunk(
                message=message_chunk, generation_info=generation_info or None
            )

            if run_manager:
                await run_manager.on_llm_new_token(
                    token=generation_chunk.text,
                    chunk=generation_chunk,
                    logprobs=logprobs,
                )
            yield generation_chunk