dosubot[bot]
commented
2 months ago Hey there, @tomi-bigpi! I'm here to help you out with any bugs, questions, or contributions you have in mind. Let's tackle this together!
To implement streaming support for the RunnableWithFallbacks class in LangChain.js, you need to override the stream method to handle the streaming logic, including the fallback mechanism. Here is how you can do it:
- Override the
streammethod to handle the primary runnable and fallbacks. - Ensure that the method yields chunks from the primary runnable and, if it fails, from the fallback runnables.
Here is the implementation:
export class RunnableWithFallbacks<RunInput, RunOutput> extends Runnable<
RunInput,
RunOutput
> {
static lc_name() {
return "RunnableWithFallbacks";
}
lc_namespace = ["langchain_core", "runnables"];
lc_serializable = true;
runnable: Runnable<RunInput, RunOutput>;
fallbacks: Runnable<RunInput, RunOutput>[];
constructor(fields: {
runnable: Runnable<RunInput, RunOutput>;
fallbacks: Runnable<RunInput, RunOutput>[];
}) {
super(fields);
this.runnable = fields.runnable;
this.fallbacks = fields.fallbacks;
}
*runnables() {
yield this.runnable;
for (const fallback of this.fallbacks) {
yield fallback;
}
}
async *stream(
input: RunInput,
options?: Partial<RunnableConfig>
): AsyncGenerator<RunOutput> {
const runnables = this.runnables();
for (const runnable of runnables) {
try {
const stream = await runnable.stream(input, options);
for await (const chunk of stream) {
yield chunk;
}
return; // Exit if the primary runnable succeeds
} catch (e) {
// Log the error and continue to the next fallback
console.error(`Runnable failed: ${e.message}`);
}
}
throw new Error("All runnables failed.");
}
}This implementation ensures that the RunnableWithFallbacks class attempts to stream output from the primary runnable first and falls back to the other runnables if the primary one fails [1].
To continue talking to Dosu, mention @dosu.
tomi-bigpi
jacoblee93
jsegeren
Checked other resources
Example Code
Sample code similar to the following should stream the results:
However, currently
invokewill be called on the model(s) and no streaming happens.A sample patch to resolves the issue:
Error Message and Stack Trace (if applicable)
No response
Description
(Modified description of the situation based on dosubot's description from a similar issue:) The stream method implementation provided in the base class (from which other runnables inherit) is designed to work with individual Runnable instances. It leverages an asynchronous generator (_streamIterator) to yield results as they become available.
The
RunnableWithFallbacksand similar constructs do not override or provide their own implementations of the stream method that would allow them to handle streaming appropriately across all constituent runnables. Without such an implementation, attempting to stream through aRunnableWithFallbacksdefaults to the base class's streaming behavior, which does not implement streaming/call streaming on the actual primary or fallback models.We did a quick local patch of the built files. I've added the patch code above even though it's not the best way to fix the issue. The patch code is just for reference.
System Info
Platform: mac