Skip to content

Commit 11e8822

Browse files
authored
Merge pull request #43 from EggerMarc/fix/bidi-streaming-ergonomics
Fix/bidi streaming ergonomics
2 parents d8e8cbe + 77ec47a commit 11e8822

39 files changed

Lines changed: 1304 additions & 931 deletions

Cargo.lock

Lines changed: 13 additions & 13 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

Cargo.toml

Lines changed: 19 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ schemars = "1.1.0"
3535
serde = { version = "1.0.228", features = ["derive"] }
3636
serde_json = "1.0.145"
3737
thiserror = "2.0.17"
38-
tokio = { version = "1.48.0", features = ["macros", "rt-multi-thread", "time", "sync"] }
38+
tokio = { version = "1.48.0", features = ["macros", "rt-multi-thread", "time", "sync", "io-std", "io-util"] }
3939
tools-rs = "0.3.3"
4040
tools_core = "0.3.3"
4141
tools_macros = "0.3.0"
@@ -45,7 +45,7 @@ url = { version = "2.5.8", features = ["serde"] }
4545

4646
[package]
4747
name = "chat-rs"
48-
version = "0.4.0"
48+
version = "0.5.0"
4949
description = "Build LLM clients with ease, attach them to your tools"
5050
autoexamples = false
5151
authors.workspace = true
@@ -59,18 +59,18 @@ name = "chat_rs"
5959
path = "src/lib.rs"
6060

6161
[dependencies]
62-
chat-core = { path = "core", version = "0.3.0" }
63-
chat-completions = { path = "providers/completions", version = "0.2.3", optional = true }
64-
chat-responses = { path = "providers/responses", version = "0.1.2", optional = true }
65-
chat-gemini = { path = "providers/gemini", version = "0.2.3", optional = true }
66-
chat-openai = { path = "providers/openai", version = "0.3.2", optional = true }
67-
chat-claude = { path = "providers/claude", version = "0.2.3", optional = true }
68-
chat-ollama = { path = "providers/ollama", version = "0.2.3", optional = true }
69-
chat-huggingface = { path = "providers/huggingface", version = "0.2.3", optional = true }
70-
chat-cerebras = { path = "providers/cerebras", version = "0.2.3", optional = true }
71-
chat-deepseek = { path = "providers/deepseek", version = "0.1.2", optional = true }
72-
chat-mistralrs = { path = "providers/mistralrs", version = "0.1.4", optional = true }
73-
chat-router = { path = "providers/router", version = "0.2.3", optional = true }
62+
chat-core = { path = "core", version = "0.4.0" }
63+
chat-completions = { path = "providers/completions", version = "0.2.4", optional = true }
64+
chat-responses = { path = "providers/responses", version = "0.1.3", optional = true }
65+
chat-gemini = { path = "providers/gemini", version = "0.2.4", optional = true }
66+
chat-openai = { path = "providers/openai", version = "0.3.3", optional = true }
67+
chat-claude = { path = "providers/claude", version = "0.2.4", optional = true }
68+
chat-ollama = { path = "providers/ollama", version = "0.2.4", optional = true }
69+
chat-huggingface = { path = "providers/huggingface", version = "0.2.4", optional = true }
70+
chat-cerebras = { path = "providers/cerebras", version = "0.2.4", optional = true }
71+
chat-deepseek = { path = "providers/deepseek", version = "0.1.3", optional = true }
72+
chat-mistralrs = { path = "providers/mistralrs", version = "0.1.5", optional = true }
73+
chat-router = { path = "providers/router", version = "0.2.4", optional = true }
7474
tokio.workspace = true
7575
schemars.workspace = true
7676
serde.workspace = true
@@ -303,6 +303,11 @@ name = "openai-input-stream"
303303
path = "./examples/openai/input_stream.rs"
304304
required-features = ["openai", "stream"]
305305

306+
[[example]]
307+
name = "openai-interactive"
308+
path = "./examples/openai/interactive.rs"
309+
required-features = ["openai", "stream"]
310+
306311
[[example]]
307312
name = "openai-hitl"
308313
path = "./examples/openai/hitl.rs"

README.md

Lines changed: 26 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ Add to your `Cargo.toml`:
2424

2525
```toml
2626
[dependencies]
27-
chat-rs = { version = "0.4.0", features = ["openai"] }
27+
chat-rs = { version = "0.5.0", features = ["openai"] }
2828
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
2929
```
3030

@@ -52,15 +52,15 @@ Enable providers via feature flags:
5252

5353
```toml
5454
# Pick one or more
55-
chat-rs = { version = "0.4.0", features = ["gemini"] }
56-
chat-rs = { version = "0.4.0", features = ["claude"] }
57-
chat-rs = { version = "0.4.0", features = ["openai"] }
58-
chat-rs = { version = "0.4.0", features = ["ollama"] }
59-
chat-rs = { version = "0.4.0", features = ["huggingface"] }
60-
chat-rs = { version = "0.4.0", features = ["cerebras"] }
61-
chat-rs = { version = "0.4.0", features = ["completions"] }
62-
chat-rs = { version = "0.4.0", features = ["router", "gemini", "claude"] }
63-
chat-rs = { version = "0.4.0", features = ["gemini", "claude", "openai", "stream"] }
55+
chat-rs = { version = "0.5.0", features = ["gemini"] }
56+
chat-rs = { version = "0.5.0", features = ["claude"] }
57+
chat-rs = { version = "0.5.0", features = ["openai"] }
58+
chat-rs = { version = "0.5.0", features = ["ollama"] }
59+
chat-rs = { version = "0.5.0", features = ["huggingface"] }
60+
chat-rs = { version = "0.5.0", features = ["cerebras"] }
61+
chat-rs = { version = "0.5.0", features = ["completions"] }
62+
chat-rs = { version = "0.5.0", features = ["router", "gemini", "claude"] }
63+
chat-rs = { version = "0.5.0", features = ["gemini", "claude", "openai", "stream"] }
6464
```
6565

6666
| Provider | Feature | API Key Env Var | Builder |
@@ -187,7 +187,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
187187
Load tools from Python scripts at runtime via the `python` feature (powered by `tools-rs` 0.3 + PyO3). Decorate functions with `@tool()` and point `ToolsBuilder` at a directory of `.py` files — they register alongside any native `#[tool]`s.
188188

189189
```toml
190-
chat-rs = { version = "0.4.0", features = ["gemini", "python"] }
190+
chat-rs = { version = "0.5.0", features = ["gemini", "python"] }
191191
```
192192

193193
```python
@@ -248,7 +248,7 @@ println!("Name: {}, Likes: {:?}", response.content.name, response.content.likes)
248248
Enable the `stream` feature flag:
249249

250250
```toml
251-
chat-rs = { version = "0.4.0", features = ["gemini", "stream"] }
251+
chat-rs = { version = "0.5.0", features = ["gemini", "stream"] }
252252
```
253253

254254
```rust
@@ -277,38 +277,41 @@ while let Some(chunk) = stream.next().await {
277277

278278
## Input Streaming (bidirectional)
279279

280-
Push input *into* the chat while the model is producing output — typed text, audio chunks, tool results, anything that fits a `PartEnum`. Useful for robotics, voice assistants, or any consumer where new context arrives during generation.
280+
Push input *into* the chat while the model is producing output — typed text, audio chunks (as `PartEnum::File`), tool results, anything that fits a `PartEnum`. Useful for robotics, voice assistants, or any consumer where new context arrives during generation.
281281

282-
Transition the builder into `InputStreamed<I>` via `.with_input_stream::<I>()`, then call `chat.stream(&mut messages, input)` with any `Stream<Item = PartEnum> + Send + Unpin + 'static`. On each input event the engine merges it into `Messages` per-variant (text/file/structured → push as user content; tool → resolve matching pending tool by call-id), drops the current provider stream, and re-enters with the updated state. For HTTP/SSE providers this is interrupt-and-restart; native-WS providers (planned OpenAI Realtime, Gemini Live) can hold their session open across calls in their client state — engine surface is identical either way.
282+
Transition the builder into `InputStreamed` via `.with_input_stream()`. `chat.stream(&mut messages)` then returns a **`ChatStream`**: it *is* the output stream you iterate with `.next()`, and it carries an input side you push to with `.send()`. `.send()` is the inverse of `.next()` — one verb for every input. Pushed input merges into `Messages` (coalescing into the trailing user turn), drops the current provider stream, and re-enters with the updated state. For HTTP/SSE providers this is interrupt-and-restart; native-WS providers (planned OpenAI Realtime, Gemini Live) can hold their session open — the surface is identical either way.
283+
284+
Completed work is never lost on an interrupt: every finished tool call and result stays in `Messages` and is re-sent; only the in-flight partial generation is discarded (tools execute *between* steps, never mid-stream, so an interrupt can't sever a running tool).
283285

284286
```rust
285-
use chat_rs::{ChatBuilder, PartEnum, StreamEvent, openai::OpenAIBuilder, types::messages};
286-
use futures::{StreamExt, channel::mpsc};
287+
use chat_rs::{ChatBuilder, StreamEvent, openai::OpenAIBuilder, types::messages};
288+
use futures::StreamExt;
287289

288290
let client = OpenAIBuilder::new().with_model("gpt-4o").build();
289291

290-
let (input_tx, input_rx) = mpsc::unbounded::<PartEnum>();
291-
292292
let mut chat = ChatBuilder::new()
293293
.with_model(client)
294-
.with_input_stream::<mpsc::UnboundedReceiver<PartEnum>>()
294+
.with_input_stream()
295295
.build();
296296

297297
let mut messages = messages::from_user(vec!["Tell me a long story about a rust crab."]);
298298

299+
let mut stream = chat.stream(&mut messages).await?;
300+
301+
// A producer handle — Clone + Send + 'static, so it drops into a task.
299302
// Interrupt mid-generation with a follow-up:
303+
let input = stream.input();
300304
tokio::spawn(async move {
301305
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
302-
let _ = input_tx.unbounded_send(PartEnum::from("Wait — make the crab wear a hat.".to_string()));
306+
let _ = input.send("Wait — make the crab wear a hat.");
303307
});
304308

305-
let mut stream = chat.stream(&mut messages, input_rx).await?;
306309
while let Some(event) = stream.next().await {
307310
if let StreamEvent::TextChunk(t) = event? { print!("{t}"); }
308311
}
309312
```
310313

311-
See `examples/openai/input_stream.rs` for a complete runnable example.
314+
For concurrent input and output, `split()` the stream into independent `(InputStream, OutputStream)` halves; `cancel()` tears the exchange down. See `examples/openai/input_stream.rs` for the minimal form and `examples/openai/interactive.rs` for an interactive CLI with mid-reply barge-in.
312315

313316
## Human in the Loop
314317

@@ -498,7 +501,7 @@ let client = OpenAIBuilder::new()
498501
To use WebSocket transport (e.g. for OpenAI's Responses API over WS):
499502

500503
```toml
501-
chat-rs = { version = "0.4.0", features = ["openai", "stream", "tokio-tungstenite"] }
504+
chat-rs = { version = "0.5.0", features = ["openai", "stream", "tokio-tungstenite"] }
502505
```
503506

504507
```rust

ROADMAP.md

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,7 @@ Tracking upcoming providers and features for chat-rs.
6060
- [x] **OpenAI WebSocket streaming**`AsyncWsTransport` with `.with_message_type("response.create")` connects to `wss://api.openai.com/v1/responses`, authenticates once on handshake, streams events. Connection reuse across calls, terminal event detection, error frame handling.
6161
- [x] **Image generation**`File` split into kind/source (`#[non_exhaustive]`). OpenAI `image_generation_call` and Gemini `inlineData` image parts decode into `PartEnum::File(File { kind: Image, .. })`. Claude has no image output upstream.
6262
- [x] **Mid-stream structured events**`StreamEvent::Structured(Value)` variant for providers that emit complete typed objects mid-stream (robotics action steps, etc.). Engine accumulates into `ChatResponse.content.parts` as `PartEnum::Structured` for non-streaming-equivalent semantics. (chat-core 0.3.0)
63-
- [x] **Input-stream type-state**`Chat<CP, InputStreamed<I>>::stream(&mut messages, input)` consumes a caller-supplied `Stream<Item = PartEnum>` alongside the model output. Audio/text/tool results merge case-by-case into `Messages`; engine restarts the provider stream on each input event. Same interrupt-and-restart pattern as HITL. Native-WS providers (planned Realtime/Live) keep their session open in client state; trait contract stays unchanged. (chat-core 0.3.0)
63+
- [x] **Input-stream type-state**`Chat<CP, InputStreamed>::stream(&mut messages)` returns a `ChatStream`: the output stream you iterate with `.next()`, carrying an input side you push to with `.send()` (the inverse of `.next()`), with `split()` into independent `(InputStream, OutputStream)` halves and `cancel()` to tear down. Pushed input rides as `PartEnum` (audio = `File`, mapped caller-side before `send`), coalesces into the trailing user turn, and restarts the provider stream — same interrupt-and-restart pattern as HITL, now push-driven. The producer handle is `Clone + Send + 'static`, so it drops into a task; completed tool work survives interrupts (only the in-flight partial is discarded). Native-WS providers (planned Realtime/Live) keep their session open in client state; trait contract stays unchanged. (redesigned in chat-core 0.4.0)
6464

6565
### Medium Term
6666

core/Cargo.toml

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,11 @@
11
[package]
22
name = "chat-core"
3-
version = "0.3.0"
3+
version = "0.4.0"
44
description = "Core library for chat-rs"
55
edition.workspace = true
66
authors.workspace = true
77
license.workspace = true
8+
repository.workspace = true
89

910
[dependencies]
1011
async-trait.workspace = true

core/README.md

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ You usually don't depend on this crate directly — instead, depend on a provide
88

99
```toml
1010
[dependencies]
11-
chat-core = "0.3.0"
11+
chat-core = "0.4.0"
1212
```
1313

1414
## What's in here
@@ -20,12 +20,17 @@ chat-core = "0.3.0"
2020
- **`types`**`Messages`, `Content`, `Parts`, `Tool`, `ChatOptions`, `ChatResponse`, `StreamEvent`, `Metadata`, etc.
2121
- **`error`**`ChatError`, `ChatFailure`
2222

23+
## What's new in 0.4
24+
25+
- **Bidirectional streaming, redesigned.** `Chat<CP, InputStreamed>::stream(&mut messages)` now returns a **`ChatStream`** instead of taking a caller-supplied input stream. It *is* the output stream you iterate with `.next()`, and it carries an input side you push to with `.send()` — the inverse of `.next()`, one verb for every input. `split()` peels it into independent `(InputStream, OutputStream)` halves (the `InputStream` is `Clone + Send + 'static`, so it drops into a task and clones into multiple producers); `cancel()` tears the exchange down. Builder transition: `ChatBuilder::with_input_stream()` (no longer generic).
26+
27+
Pushed input rides as `PartEnum` (audio = `File`, text = `Text`, tool result = `Tool`), mapped caller-side before `send`. It coalesces into the trailing user turn via `Messages::push` and restarts the provider stream — the same interrupt-and-restart pattern HITL uses, now push-driven. Completed tool work survives interrupts (tools run *between* steps, never mid-stream); only the in-flight partial generation is discarded.
28+
2329
## What's new in 0.3
2430

2531
- **`StreamEvent::Structured(Value)`** — providers can yield complete structured objects mid-stream (each event is a whole `serde_json::Value`, not a fragment). The engine accumulates them into `ChatResponse.content.parts` as `PartEnum::Structured` so non-streaming consumers see them too. Drop-in for robotics consumers that produce a stream of typed action steps.
26-
- **`InputStreamed<I>` type-state**`Chat<CP, InputStreamed<I>>::stream(&mut messages, input: I)` interleaves the model's output stream with a caller-supplied `Stream<Item = PartEnum>` input source. Audio bytes ride as `PartEnum::File`, text as `PartEnum::Text`, tool results as `PartEnum::Tool`. Each input event triggers a case-by-case merge into `Messages` and re-opens the provider stream with the updated state — same interrupt-and-restart pattern HITL already uses, just automated. Builder transition: `ChatBuilder::with_input_stream::<I>()`. Bounds: `I: Stream<Item = PartEnum> + Send + Unpin + 'static`.
2732

28-
Both are additive — existing `Chat<CP, Unstructured>::stream` callers see no behavior change.
33+
Both are additive over `Chat<CP, Unstructured>::stream`, which is unchanged.
2934

3035
## Feature Flags
3136

core/src/builder.rs

Lines changed: 9 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -89,24 +89,18 @@ impl<CP: CompletionProvider> ChatBuilder<CP, Unstructured> {
8989
}
9090

9191
/// Transition into the input-stream type-state. The resulting
92-
/// `Chat<CP, InputStreamed<I>>` exposes a `stream(&mut messages, input)`
93-
/// method that interleaves the model's output stream with a
94-
/// caller-supplied `Stream<Item = PartEnum>` input source. Audio
95-
/// chunks ride as `PartEnum::File`, text as `PartEnum::Text`,
96-
/// tool results as `PartEnum::Tool`, etc. — no parallel input enum.
97-
///
98-
/// `I` is just a type-marker at builder time; the actual stream
99-
/// instance is passed at `chat.stream(...)` call time, so a single
100-
/// `Chat` can be reused across multiple input streams of the same
101-
/// type.
92+
/// `Chat<CP, InputStreamed>` exposes a `stream(&mut messages)` method
93+
/// that returns a [`ChatStream`](crate::chat::input::ChatStream): the
94+
/// output stream you iterate with `.next()`, carrying an input side you
95+
/// push to with `.send()` (or `split()` into independent handles).
96+
/// Audio rides as `PartEnum::File`, text as `PartEnum::Text`, tool
97+
/// results as `PartEnum::Tool`, etc. — no parallel input enum, and a
98+
/// continuous producer is mapped into `PartEnum` caller-side before it
99+
/// reaches `send`.
102100
#[cfg(feature = "stream")]
103-
pub fn with_input_stream<I>(self) -> ChatBuilder<CP, InputStreamed<I>>
101+
pub fn with_input_stream(self) -> ChatBuilder<CP, InputStreamed>
104102
where
105103
CP: StreamProvider,
106-
I: futures::Stream<Item = crate::types::messages::parts::PartEnum>
107-
+ Send
108-
+ Unpin
109-
+ 'static,
110104
{
111105
if self.output_shape.is_some() {
112106
println!(

0 commit comments

Comments
 (0)