streamEvents does not properly work on RemoteRunnable

[jumski]

Checked other resources

• I added a very descriptive title to this issue.
• I searched the LangChain.js documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.
• I am sure that this is a bug in LangChain.js rather than my code.
• The bug is not resolved by updating to the latest stable version of LangChain (or the specific integration package).

Example Code

Chain

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai.chat_models import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

OPENAI_API_KEY = "your API key"

chain = (
    ChatPromptTemplate.from_template("Tell me a joke about {input}")
    | ChatOpenAI(model="gpt-3.5-turbo-1106", api_key=OPENAI_API_KEY)
    | StrOutputParser()
)

LangServe app

from fastapi import FastAPI
from langserve import add_routes
import chain

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["*"],
)

add_routes(app, chain, path='/remote-chain')

if __name__ == "__main__":
    import os
    import uvicorn

    HTTP_PORT = 8081
    uvicorn.run(app, host="0.0.0.0", port=HTTP_PORT)

RemoteRunnable#streamEvents in JS

import { RemoteRunnable } from '@langchain/core/runnables/remote';

const remoteChain = new RemoteRunnable('remote-chain');

async function streamEvents() {
  const stream = remoteChain.streamEvents({ input: "programming" }, { version: "v1" });
  
  for await (const event of stream) {
    console.log('event', event);  
  }
}

Expected this code to produce stream of events that one can expect from streamEvents and a POST request to /stream_events, but it hits POST /stream and the event stream/response looks like a regular stream response that uses SSE.

import chain
import asyncio

if __name__ == "__main__":
    async def run():
        events = chain.astream_events(dict(input="programming"), version='v1')

        async for event in events:
            print(event)

    asyncio.run(run());

This code properly streams all the events, proving that the chain itself is streamable.

Error Message and Stack Trace (if applicable)

No response

Description

streamEvents called on RemoteRunnable does not seem to properly stream the events and instead posts regular /stream endpoint.

The chain is streamable for sure (verified in python).

I also noticed that the chain.streamEvents function definition that i inspect on real chain instance in the browser debug console does not look like the definition i verify in code, but instead looks like the definition for Runnable._streamEvents:

chain.streamEvents

Runnable._streamEvents

RemoteRunnable.streamEvents

System Info

pnpm show langchain

langchain@0.1.36 | MIT | deps: 18 | versions: 269
Typescript bindings for langchain
https://github.com/langchain-ai/langchainjs/tree/main/langchain/

keywords: llm, ai, gpt3, chain, prompt, prompt engineering, chatgpt, machine learning, ml, openai, embeddings, vectorstores

dist
.tarball: https://registry.npmjs.org/langchain/-/langchain-0.1.36.tgz
.shasum: 2f201f681d83fb265716e28e7dfcfe61fbeef2c2
.integrity: sha512-NTbnCL/jKWIeEI//Nm1oG8nhW3vkYWvEMr1MPotmTThTfeKfO87eV/OAzAyh6Ruy6GFs/qofRgQZGIe6XvXTNQ==
.unpackedSize: 4.0 MB

dependencies:
@anthropic-ai/sdk: ^0.9.1
@langchain/community: ~0.0.47
@langchain/core: ~0.1.60
@langchain/openai: ~0.0.28
@langchain/textsplitters: ~0.0.0
binary-extensions: ^2.2.0
js-tiktoken: ^1.0.7
js-yaml: ^4.1.0
jsonpointer: ^5.0.1
langchainhub: ~0.0.8
langsmith: ~0.1.7
ml-distance: ^4.0.0
openapi-types: ^12.1.3
p-retry: 4
uuid: ^9.0.0
yaml: ^2.2.1
zod-to-json-schema: ^3.22.3
zod: ^3.22.4

maintainers:
- nfcampos <nuno@boringbits.io>
- jacoblee93 <jacoblee93@gmail.com>
- andrewnguonly <andrewnguonly@gmail.com>
- davidduong <david@duong.cz>
- hwchase17 <hw.chase.17@gmail.com>
- basproul <braceasproul@gmail.com>

dist-tags:
latest: 0.1.36
next: 0.2.0-rc.1

published a week ago by jacoblee93 <jacoblee93@gmail.com>

Platform: Linux

Node: v21.5.0

pnpm: 8.14.1

[dosubot]

It appears you're encountering an issue where the streamEvents method on RemoteRunnable is not behaving as expected, specifically it's making a POST request to /stream instead of /stream_events. This behavior is unusual since streamEvents is designed to interact with the /stream_events endpoint for streaming events.

The difference between streamEvents in RemoteRunnable and _streamEvents in the base Runnable class is significant. The former is a public API for clients to monitor runnable execution, while the latter is a protected method for internal event generation based on runnable logs. It's unexpected for streamEvents in RemoteRunnable to act like _streamEvents.

This issue might be related to a version mismatch or a bug in the LangChain.js version you're using (0.1.36). However, there's no known issue related to this in the version you mentioned. Another potential cause could be compatibility with your Node.js version (21.5.0). LangChain.js tests are run on Node.js versions 18.x and 20.x, so using a version within this range might avoid compatibility issues.

If the problem persists, providing more details about the error messages or the exact output when calling streamEvents could help in diagnosing the issue further.

Details

Sources

• langchain-core/src/runnables/base.ts
• langchain-core/src/runnables/tests/runnable_stream_events.test.ts
• package.json
• .github/workflows/unit-tests-langchain-community.yml
• langchain-core/src/runnables/remote.ts
• examples/src/ecosystem/langsmith_stream_events.ts

About Dosu
This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

To continue the conversation, mention @dosu.

[jumski]

@dosu can you confirm that RemoteRunnable should have full support for streamEvents and it should include all the events from the sub-runnables from the remote runnable?

[jumski]

@dosu in what circumstances one cannot use streamEvents or it is unfeasible to use it?

[dosubot]

Yes, RemoteRunnable should indeed have full support for streamEvents, and it is designed to include all events from the sub-runnables within the remote runnable. This means that when you use streamEvents with a RemoteRunnable, it should stream events as they occur in real-time, including those from any sub-runnables involved in the chain. The intention is for it to make a POST request to the /stream_events endpoint, allowing for a comprehensive stream of events that reflect the execution of the entire chain.

Regarding the circumstances where streamEvents might be unfeasible or not recommended, there are a few specific scenarios to consider:

• Compatibility issues with the specified version in the options, as streamEvents is designed to work with a specific version ("v1"). If there's a mismatch, it might not function as expected.
• Situations where the runnable or chain does not emit meaningful events or is not set up to support event streaming effectively. This could make the use of streamEvents less beneficial.
• Environments where long-lived connections are problematic or not supported, such as certain network configurations or platforms that restrict streaming protocols. This could limit the ability to use streamEvents effectively.

It's important to ensure that your RemoteRunnable and the environment it operates in are correctly configured to support streamEvents. This includes using the correct version and understanding the limitations of your specific setup. If you continue to experience issues, it might be necessary to review the implementation details and compatibility with your environment and the version of LangChain.js you're using.

Details

About Dosu
This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

To continue the conversation, mention @dosu.

[jumski]

@dosu i am pretty sure we are onto something.

When remote chain is RunnableSequence, its sub-runnables will not be included in the streamEvents, and only events with name==RunnableSequence will be streamed.

how one can target the inner rrunnables of RunnableSequence, when the sequence chain is exposed as langserve remote runnable and consumed in JS?

[dosubot]

To target the inner runnables of a RunnableSequence when the sequence chain is exposed as a LangServe remote runnable and consumed in JavaScript, it's crucial to ensure that the streamEvents method is implemented in a way that captures events from all sub-runnables within the sequence. This involves leveraging the capabilities of the RunnableSequence and the streamEvents method to listen for and process events from every part of the execution chain, including nested or sub-runnables.

Here's a more detailed approach to ensure you capture events from sub-runnables:

1. Ensure Proper Implementation of RunnableSequence: When constructing your RunnableSequence, make sure that each sub-runnable is correctly added to the sequence. This setup is crucial for the streamEvents method to be able to capture events from these sub-runnables.

2. Use streamEvents Correctly: When calling streamEvents on your RemoteRunnable that represents a RunnableSequence, ensure that you're using the method in a way that's designed to capture and stream events from all runnables within the sequence. This might involve correctly setting up the event listeners and ensuring that the streamEvents method is implemented to handle events from sub-runnables.

3. Adjust Your JavaScript Implementation: In your JavaScript code that consumes the LangServe remote runnable, make sure that your implementation of the event streaming logic is set up to handle and process events from all parts of the runnable sequence. This might involve adjusting how you iterate over the streamed events and how you process each event.

By following these steps, you should be able to capture and process events from all sub-runnables within a RunnableSequence when it's exposed as a LangServe remote runnable and consumed in JavaScript. This approach ensures comprehensive event streaming and processing, allowing you to fully leverage the capabilities of the LangChain.js library and the RunnableSequence construct for your applications.

If you're still facing issues or have specific questions about implementing this in your project, feel free to provide more details or ask for further clarification.

Details

About Dosu
This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

To continue the conversation, mention @dosu.

[jumski]

@dosu you tripped into the vague mumbo jumbo

tell me details about implementation of streamEvents for RunnableSequence so i can understand its limitations

[dosubot]

The streamEvents method for RunnableSequence is designed to aggregate and stream events from all its sub-runnables. This means that during the execution of a RunnableSequence, it should emit events at the start and end of the sequence, as well as for each sub-runnable's execution. This setup is crucial for capturing a comprehensive stream of events, including those from nested sequences or complex runnables, which should similarly aggregate and emit events from their components.

When a RunnableSequence is exposed as a LangServe remote runnable, it's important to ensure the remote service is correctly set up to handle streaming responses. This often involves using server-sent events (SSE) or WebSockets, with the streamEvents output formatted to be compatible with these streaming technologies. On the client side, using EventSource (for SSE) or a WebSocket client to consume the streamed events is necessary. Implementing event handlers to process the incoming events may involve parsing event data and responding accordingly.

The issue of hitting the /stream endpoint instead of /stream_events could be related to the specific implementation within the RemoteRunnable class. It's essential to ensure that the streamEvents method is correctly used to target the /stream_events endpoint for streaming structured events, rather than raw data or logs. The RemoteRunnable class is responsible for converting the streamEvents method's output into a suitable format for streaming, ensuring compatibility with SSE or WebSockets by converting the response body into an iterable readable data stream and parsing each chunk of data as JSON into structured events.

This detailed explanation aims to clarify the implementation and limitations of streamEvents for RunnableSequence in LangChain.js. If there are more questions or further clarification is needed, feel free to ask.

Details

About Dosu
This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

To continue the conversation, mention @dosu.

[jumski]

@dosu it is still not helpful.

search for code that implements streamEvents for RunnableSequence and explain to me step by step how it works

[dosubot]

The implementation of streamEvents for a RunnableSequence in LangChain.js is designed to handle the streaming of events from all the sub-runnables within the sequence. This process involves several key steps to ensure that events are captured, aggregated, and streamed effectively:

1. Initialization and Event Capture: When a RunnableSequence starts, it initializes the streaming process. As the sequence progresses, it captures events from each sub-runnable. These events include the start, execution details, and completion of each sub-runnable, providing a detailed view of the sequence's execution.

2. Event Aggregation: The streamEvents method aggregates these events, ensuring that they are collected in a manner that maintains the sequence and context of the execution flow. This aggregation is crucial for understanding the execution order and the outcome of each sub-runnable within the sequence.

3. Streaming Mechanism: The method then streams these aggregated events to the client. This streaming is typically done over HTTP, using technologies such as Server-Sent Events (SSE) or WebSockets, depending on the implementation specifics and the client's capabilities. This ensures real-time or near-real-time transmission of events from the server to the client.

4. Client-Side Handling: On the client side, these streamed events are consumed and processed. This might involve updating the UI in real-time, logging, or triggering other processes based on the events' content. The client implementation needs to handle the stream correctly, parsing the events and taking appropriate actions based on the event data.

Regarding the specific issue of hitting the /stream endpoint instead of /stream_events, it's important to note that this behavior might be tied to the implementation details within the LangChain.js framework or the server-side setup. The framework's design for handling event streams, including the naming and functionality of endpoints, can vary between versions or configurations. Ensuring compatibility with the expected endpoint and the correct handling of streamed events is essential for the effective use of streamEvents with RunnableSequence.

If you're encountering issues with streaming events from sub-runnables or if the events are not being streamed as expected, reviewing the specific implementation details of the RunnableSequence and the streamEvents method in your version of LangChain.js is recommended. This review can help identify any discrepancies or areas for adjustment to align with the expected behavior of streaming comprehensive event data from all parts of the sequence.

Details

About Dosu
This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

To continue the conversation, mention @dosu.

[kikohs]

I have this issue as well with an AgentExecutor, the /stream endpoint is called but I get all the chunks at once, like invoke. steamEvents doesn't work either.
If I use a local model, all is fine.