langchain Architecture Docs
Home / Function/ astream() — langchain Function Reference

astream() — langchain Function Reference

Architecture documentation for the astream() function in test_runnable_events_v2.py from the langchain codebase.

Function python LangChainCore Runnables called by 2

Entity Profile

Dependency Diagram

graph TD
  858b85ba_6d9e_d209_9eb5_2469bad0adca["astream()"]
  ec849ced_49cd_d2f5_60e0_db0d12879deb["StreamingRunnable"]
  858b85ba_6d9e_d209_9eb5_2469bad0adca -->|defined in| ec849ced_49cd_d2f5_60e0_db0d12879deb
  ee416508_9e2a_a50d_f6d0_f03000a99329["test_async_in_async_stream_lambdas()"]
  ee416508_9e2a_a50d_f6d0_f03000a99329 -->|calls| 858b85ba_6d9e_d209_9eb5_2469bad0adca
  20a2cdd2_25f8_e563_51d8_eecf2fc98b01["test_astream_events_from_custom_runnable()"]
  20a2cdd2_25f8_e563_51d8_eecf2fc98b01 -->|calls| 858b85ba_6d9e_d209_9eb5_2469bad0adca
  style 858b85ba_6d9e_d209_9eb5_2469bad0adca fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py lines 2148–2182

    async def astream(
        self,
        input: Any,
        config: RunnableConfig | None = None,
        **kwargs: Any | None,
    ) -> AsyncIterator[Addable]:
        config = ensure_config(config)
        callback_manager = get_async_callback_manager_for_config(config)
        run_manager = await callback_manager.on_chain_start(
            None,
            input,
            name=config.get("run_name", self.get_name()),
            run_id=config.get("run_id"),
        )

        try:
            final_output = None
            for element in self.iterable:
                if isinstance(element, BaseException):
                    raise element  # noqa: TRY301
                yield element

                if final_output is None:
                    final_output = element
                else:
                    try:
                        final_output = final_output + element
                    except TypeError:
                        final_output = element

            # set final channel values as run output
            await run_manager.on_chain_end(final_output)
        except BaseException as e:
            await run_manager.on_chain_error(e)
            raise

Domain

Subdomains

Defined In

Called By

Source

Frequently Asked Questions

What does astream() do?
astream() is a function in the langchain codebase, defined in libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py.
Where is astream() defined?
astream() is defined in libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py at line 2148.
What calls astream()?
astream() is called by 2 function(s): test_astream_events_from_custom_runnable, test_async_in_async_stream_lambdas.

Analyze Your Own Codebase

Get architecture documentation, dependency graphs, and domain analysis for your codebase in minutes.

Try Supermodel Free