Mappings between Stream and Async Generators

[awestlake87]

It seems like these conversions will be useful to users, so I think we should integrate some of the existing solutions into the library. @ThibaultLemaire has created bindings from Rust streams to async generators, and I've written up a simple function to forward items yielded from an async generator to an async_channel for #15. I think we have all the pieces, so all that's left is to try and integrate them into the API.

[omegablitz]

Does the example of bindings from Rust streams to Python async generators still exist? I saw that #21 was merged but that seems to be for the inverse use-case. Any example (even if rough) would be very useful as a starting point!

[ThibaultLemaire]

Ah I see @awestlake87 mentioned me, but didn't link #6 where the relevant discussion and links are.

Here is a direct link to my implementation for your convenience

[awestlake87]

Sorry for the delay, it's been a busy week. I'd give @ThibaultLemaire's implementation a try and see if that works for you!

This is something I'd consider integrating into the library since we already have the inverse available under an unstable feature. The reason I haven't yet is because I didn't know of anyone who had needed the Rust -> Stream conversion (and honestly I had forgotten about #6 due to other releases and fixes)

[ThibaultLemaire]

I didn't know of anyone who had needed the Rust -> Stream conversion

I've had to implement it for my toy project stirling to implement the find function of mongodb (https://gitlab.com/ThibaultLemaire/stirling/-/blob/master/src/lib.rs#L139) which returns an async generator. But well... it's only a toy project.

[omegablitz]

Thanks for the pointers! I ended up just using a synchronous python generator instead as it wasn't critical to get this to be async on the python side. I may end up trying the async generator approach again in the future, but we will see!

[elyase]

Hi guys, whats the latest on async generators Rust -> Python?

[awestlake87]

There's no official binding going from Rust -> Python just yet. I'd suggest taking a look at @ThibaultLemaire's repo as an example if you need the conversion. In the past I've also driven an asyncio.Queue off of a Rust stream.

Is this something that you think you'd want in pyo3-asyncio? I've been hesitant to add it since I don't know how many people would consistently use it. Let me know what you think!

[elyase]

yes I am trying to use it, at the moment exploring alternatives like pyo3_futures thanks for the suggestion.

[StuartHadfield]

Just chiming in here @awestlake87 - being able to stream results from an async fn in rust to a generator in python would be extremely helpful.

For a bit of context - imagine you are querying your DB in rust, most of the time you will want to stream results out (rather than a whole data set), and this tends to happen async in rust. Being able to send rows back over to Python one-by-one would be exceedingly helpful.

[awestlake87]

@StuartHadfield I think it would be a good thing to have these conversions for parity with Rust async Streams. It's a bit of a complex topic, and I think in order to get to a true solution it's probably going to involve a tricky implementation, and therefore a decent amount of testing, which is part of the reason I've been pretty hesitant to add them without a concrete use-case we can point to for testing.

Most conversations I've had with people on this tend to end up with them using a different solution, so it's been hard for me to tell whether people would actually use them. The unstable async generator -> Rust Stream has been usable in the unstable-streams feature for awhile now, but so far I haven't had any feedback on them yet. That either means it works well (fingers crossed) or nobody uses them. @elyase and @StuartHadfield (or anyone reading this!), do you have some concrete use-cases that would take advantage of these conversions?

Edit: To clarify a bit, by concrete use-cases, I mean pyo3 projects that might take advantage of them. It'd be great to get some immediate feedback from code as to how well they're working.

[awestlake87]

Just to add to that, it's also not clear to me that Stream -> async generator is actually the most useful conversion. Streams and Async Generators both tend to assume a single consumer, so would it be more valuable to enable a MPMC pattern like async_channel and asyncio.Queue?

Personally I've found that, at least for prototyping, MPMC is easier to reason with and MPSC tends to work until it doesn't and then it's actually pretty tedious to refactor.

Might be worth doing both because Stream has some useful extension traits, but my thought is that MPMC would cover more use-cases.

[marcustut]

I think that mapping Rust streams into Python's async generator actually does make sense, let's say we're building a WebSocket library in Rust and we wanted to let users use it through an async generator which would result in a much simpler API.

[awestlake87]

@marcustut I think it does make sense conceptually, but for awhile now I've really just wanted someone to say they're going to use the feature. I'm not even certain people use the Python async generator -> Rust Stream conversion that exists today TBH.

[s3rius]

Hi to everyone. So, I had the same problem of not having direct mapping between Rust streams and async iterators. But, after implementing it, I realized that it's not a problem with existing functionality.

I created a gist for all fellow developers who faced the same problem. I couldn't make a generic solution because pyo3 requires all pyclasses to be non-generic.

[XieJiSS]

Hi @awestlake87, I'm willing to provide a feasible usecase for this feature. We are trying to develop a rust program which need to consume a stream of LLM response and convert it to an async generator. The motivation is that the Python SDK provided by OpenAI wastes most of its time (~70%) constructing useless classes from JSON and doing type validations, even if we only need to fetch a chat completion response. Also, the underlying httpx package will become a bottleneck after fixing aforementioned performance issues, so we'd like to RIIR this part to gain speed improvements. Since the upper level of the software is already implemented in Python, which uses async generator as it is the OpenAI Python SDK's way of providing streaming data, we'd like to expose a similar interface to limit the impact of this modification. AFAIK many third party LLM SDKs also provides an async generator interface so this could be a somewhat common scenario.

[pbarker]

I have a similar use case to @XieJiSS, we are looking to do fairly labor intensive stream processing directly off LLMs using Transformers and similar tools. We hope to provide a python interface to this as most development happens there.

[kitty-eu-org]

Hi @awestlake87, I'm willing to provide a feasible usecase for this feature. We are trying to develop a rust program which need to consume a stream of LLM response and convert it to an async generator. The motivation is that the Python SDK provided by OpenAI wastes most of its time (~70%) constructing useless classes from JSON and doing type validations, even if we only need to fetch a chat completion response. Also, the underlying httpx package will become a bottleneck after fixing aforementioned performance issues, so we'd like to RIIR this part to gain speed improvements. Since the upper level of the software is already implemented in Python, which uses async generator as it is the OpenAI Python SDK's way of providing streaming data, we'd like to expose a similar interface to limit the impact of this modification. AFAIK many third party LLM SDKs also provides an async generator interface so this could be a somewhat common scenario.

+1