core: add a no_collect: bool=False param to RunnableLambda, enables per-chunk processing instead of whole collected input

[rhighs]

This PR introduces a new param no_collect: bool to RunnableLambda.__init__. This flags allows for the _transform and _atransform methods to not collect input before processing but rather to process each individual chunk via self.func/self.afunc. This is particularly useful when we want to keep the input stream flowing as it gets transformed. One specific issue this implementation has solved on my part is being able to have a streamed conversation chain whilst being able to save the final chain output in a context memory. This is a problem some other ppl have encountered in this issue: #11945

Example usage:

def generate_values(input: Iterator[Any]) -> Iterator[int]:
    yield 1
    yield 2
    yield 3
    yield 4

def square(input: Input) -> int:
    n = cast(int, input)
    return n * n

expected_values = [1, 4, 9, 16]
runnable = RunnableGenerator(generate_values) | RunnableLambda(
   square, no_collect=True
)
for i, value in enumerate(runnable.stream({"input": "some_input"})):
    assert value == expected_values[i]

twitter: @rhighs_

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

1 Ignored Deployment

Name	Status	Preview	Comments	Updated (UTC)
langchain	⬜️ Ignored (Inspect)	Visit Preview		May 8, 2024 8:15am

[ccurme]

cc @nfcampos