Stream Batch Events

[adreo00]

Checked

• I searched existing ideas and did not find a similar one
• I added a very descriptive title
• I've clearly described the feature request and motivation for it

Feature request

Extend the astream_events runnable method (https://python.langchain.com/docs/modules/agents/how_to/streaming/#custom-streaming-with-events)
to something like astream_batch_events, which accepts inputs rather than an input.

Motivation

This would help me run the same chain on batches of inputs where I rely on getting intermediate steps as metadata. It also feels like a natural extension considering we have invoke(input={}) and batch(inputs=[{}]) interfaces.

Proposal (If applicable)

I would propose that a new interface that serves as a wrapper to astream_events be created (i.e. astream_batch_events). This wrapper would be the same as astream_events, except:

• it would accept inputs:list[dict] rather than input:dict
• it would accept a new arguement: something like input_id_attr which would allow users to determine the originating input object from a list of streamed events. The example below should help.

Extending this example of a astream_events run:

from langchain_core.runnables import RunnableLambda


async def reverse(s: str) -> str:
    return s["text"][::-1]


chain = RunnableLambda(func=reverse)


events = [event async for event in chain.astream_events({"text": "hello"}, version="v1")]

# will produce the following events
[
    {"data": {"input": "hello"}, "event": "on_chain_start", "metadata": {}, "name": "reverse", "tags": []},
    {"data": {"chunk": "olleh"}, "event": "on_chain_stream", "metadata": {}, "name": "reverse", "tags": []},
    {"data": {"output": "olleh"}, "event": "on_chain_end", "metadata": {}, "name": "reverse", "tags": []},
]

slightly modified from source, to use a dict as an input instead of a str

I might expect to see something like this from astream_batch_events:

from langchain_core.runnables import RunnableLambda


async def reverse(s: str) -> str:
    return s["text"][::-1]


chain = RunnableLambda(func=reverse)


events = [
    event
    async for event in chain.astream_batch_events(
        inputs=[{"id": 1, "text": "hello"}, {"id": 2, "text": "world"}], input_id_attr="id", version="v1"
    )
]


# will produce the following events
[
    {
        "data": {"input": "hello"},
        "event": "on_chain_start",
        "metadata": {},
        "name": "reverse",
        "tags": [],
        "input_id": 1,
    },
    {
        "data": {"chunk": "olleh"},
        "event": "on_chain_stream",
        "metadata": {},
        "name": "reverse",
        "tags": [],
        "input_id": 1,
    },
    {
        "data": {"output": "olleh"},
        "event": "on_chain_end",
        "metadata": {},
        "name": "reverse",
        "tags": [],
        "input_id": 1,
    },
    {
        "data": {"input": "world"},
        "event": "on_chain_start",
        "metadata": {},
        "name": "reverse",
        "tags": [],
        "input_id": 2,
    },
    {
        "data": {"chunk": "dlrow"},
        "event": "on_chain_stream",
        "metadata": {},
        "name": "reverse",
        "tags": [],
        "input_id": 2,
    },
    {
        "data": {"output": "dlrow"},
        "event": "on_chain_end",
        "metadata": {},
        "name": "reverse",
        "tags": [],
        "input_id": 2,
    },
]

[adreo00]

Hi @eyurtsev, would love to get your feedback on whether or not this idea makes sense, since you did such a fantastic job on the original astream_events implementation.

[casbra13]

I would love to have this functionality!