RunnableGenerator Class — langchain Architecture

Architecture documentation for the RunnableGenerator class in base.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb["RunnableGenerator"]
  e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb["RunnableGenerator"]
  e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb -->|extends| e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb
  5f3c1f1c_6f8a_e293_7cb5_97c21b4bf214["base.py"]
  e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb -->|defined in| 5f3c1f1c_6f8a_e293_7cb5_97c21b4bf214
  814a39a3_d37e_aefa_df96_5c5b9f76d247["__init__()"]
  e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb -->|method| 814a39a3_d37e_aefa_df96_5c5b9f76d247
  d9e94308_0c24_a630_4103_96627d8ad625["InputType()"]
  e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb -->|method| d9e94308_0c24_a630_4103_96627d8ad625
  1fde5d98_52c5_efe1_1a02_331fa1709cb3["get_input_schema()"]
  e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb -->|method| 1fde5d98_52c5_efe1_1a02_331fa1709cb3
  18d557c3_d23d_f559_18ee_c3326d8459bd["OutputType()"]
  e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb -->|method| 18d557c3_d23d_f559_18ee_c3326d8459bd
  1daab5de_de3b_ecb4_8ae7_af1698bb03d2["get_output_schema()"]
  e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb -->|method| 1daab5de_de3b_ecb4_8ae7_af1698bb03d2
  dab5a86e_4d35_1620_084e_f6b4d9e824fc["__eq__()"]
  e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb -->|method| dab5a86e_4d35_1620_084e_f6b4d9e824fc
  4d85e554_6949_26d9_819c_6ad5bda1ac75["__repr__()"]
  e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb -->|method| 4d85e554_6949_26d9_819c_6ad5bda1ac75
  27f4e534_10e4_c281_de23_226bdb867e10["transform()"]
  e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb -->|method| 27f4e534_10e4_c281_de23_226bdb867e10
  775a6a7e_7537_8fb6_cbfe_8b940ddaf430["stream()"]
  e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb -->|method| 775a6a7e_7537_8fb6_cbfe_8b940ddaf430
  91e2969d_a13e_0c64_478d_1174abeff5b8["invoke()"]
  e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb -->|method| 91e2969d_a13e_0c64_478d_1174abeff5b8
  cb0b1535_4933_3a35_cb65_9d60e5fff35b["atransform()"]
  e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb -->|method| cb0b1535_4933_3a35_cb65_9d60e5fff35b
  63469165_5eef_b5a2_f2d6_a9e7f07712e4["astream()"]
  e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb -->|method| 63469165_5eef_b5a2_f2d6_a9e7f07712e4
  e9693136_4f07_2492_9811_e1c477b33401["ainvoke()"]
  e9574e58_87b9_d5c5_d57e_ad2e1f0c0dfb -->|method| e9693136_4f07_2492_9811_e1c477b33401

Relationship Graph

Source Code

libs/core/langchain_core/runnables/base.py lines 4096–4396

class RunnableGenerator(Runnable[Input, Output]):
    """`Runnable` that runs a generator function.

    `RunnableGenerator`s can be instantiated directly or by using a generator within
    a sequence.

    `RunnableGenerator`s can be used to implement custom behavior, such as custom
    output parsers, while preserving streaming capabilities. Given a generator function
    with a signature `Iterator[A] -> Iterator[B]`, wrapping it in a
    `RunnableGenerator` allows it to emit output chunks as soon as they are streamed
    in from the previous step.

    !!! note
        If a generator function has a `signature A -> Iterator[B]`, such that it
        requires its input from the previous step to be completed before emitting chunks
        (e.g., most LLMs need the entire prompt available to start generating), it can
        instead be wrapped in a `RunnableLambda`.

    Here is an example to show the basic mechanics of a `RunnableGenerator`:

        ```python
        from typing import Any, AsyncIterator, Iterator

        from langchain_core.runnables import RunnableGenerator


        def gen(input: Iterator[Any]) -> Iterator[str]:
            for token in ["Have", " a", " nice", " day"]:
                yield token


        runnable = RunnableGenerator(gen)
        runnable.invoke(None)  # "Have a nice day"
        list(runnable.stream(None))  # ["Have", " a", " nice", " day"]
        runnable.batch([None, None])  # ["Have a nice day", "Have a nice day"]


        # Async version:
        async def agen(input: AsyncIterator[Any]) -> AsyncIterator[str]:
            for token in ["Have", " a", " nice", " day"]:
                yield token


        runnable = RunnableGenerator(agen)
        await runnable.ainvoke(None)  # "Have a nice day"
        [p async for p in runnable.astream(None)]  # ["Have", " a", " nice", " day"]
        ```

    `RunnableGenerator` makes it easy to implement custom behavior within a streaming
    context. Below we show an example:

        ```python
        from langchain_core.prompts import ChatPromptTemplate
        from langchain_core.runnables import RunnableGenerator, RunnableLambda
        from langchain_openai import ChatOpenAI
        from langchain_core.output_parsers import StrOutputParser


        model = ChatOpenAI()
        chant_chain = (
            ChatPromptTemplate.from_template("Give me a 3 word chant about {topic}")
            | model
            | StrOutputParser()
        )


        def character_generator(input: Iterator[str]) -> Iterator[str]:
            for token in input:
                if "," in token or "." in token:
                    yield "👏" + token
                else:
                    yield token


        runnable = chant_chain | character_generator
        assert type(runnable.last) is RunnableGenerator
        "".join(runnable.stream({"topic": "waste"}))  # Reduce👏, Reuse👏, Recycle👏.


        # Note that RunnableLambda can be used to delay streaming of one step in a
        # sequence until the previous step is finished: