Add SSE streaming for chat responses #8 - #70
Conversation
WalkthroughAdds ChangesStreaming chat endpoint
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant chat_stream
participant ChatSession
Client->>chat_stream: POST /chat/stream
chat_stream->>ChatSession: send_message_async(message, stream=True)
ChatSession-->>chat_stream: text chunks
chat_stream-->>Client: delta SSE events
chat_stream-->>Client: done or error SSE event
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@main.py`:
- Around line 189-191: In main.py lines 189-191, update the exception path in
the streaming handler to log the original exception server-side and emit only a
stable, non-sensitive public error code/message instead of str(exc). In
tests/test_chat_streaming.py lines 67-70, update the streaming error assertion
to expect that stable public contract rather than “upstream failed”.
In `@README.md`:
- Line 83: Update the streaming endpoint event documentation in README.md to
include the error event, describing its failure payload and that it terminates
the stream. Keep the existing delta and done event descriptions intact.
In `@requirements.txt`:
- Line 9: Pin the httpx dependency in requirements.txt to the tested version by
specifying an exact version constraint, ensuring installs remain reproducible
across CI and local environments.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 5d4dd242-0b8a-44d8-bb75-6b999942e52b
📒 Files selected for processing (4)
README.mdmain.pyrequirements.txttests/test_chat_streaming.py
| except Exception as exc: # pragma: no cover - defensive path | ||
| error_payload = json.dumps({"error": str(exc)}) | ||
| yield f"event: error\ndata: {error_payload}\n\n" |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Do not expose raw upstream exceptions in the public SSE contract.
Provider exception strings can disclose internal request or configuration details. Emit a stable public error code/message and log the original exception server-side; update the test to assert that safe contract instead.
main.py#L189-L191: replacestr(exc)in the event payload with a stable, non-sensitive error response.tests/test_chat_streaming.py#L67-L70: assert the stable public error code/message rather than"upstream failed".
🧰 Tools
🪛 ast-grep (0.44.1)
[info] 189-189: use jsonify instead of json.dumps for JSON output
Context: json.dumps({"error": str(exc)})
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
🪛 Ruff (0.15.21)
[warning] 189-189: Do not catch blind exception: Exception
(BLE001)
📍 Affects 2 files
main.py#L189-L191(this comment)tests/test_chat_streaming.py#L67-L70
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@main.py` around lines 189 - 191, In main.py lines 189-191, update the
exception path in the streaming handler to log the original exception
server-side and emit only a stable, non-sensitive public error code/message
instead of str(exc). In tests/test_chat_streaming.py lines 67-70, update the
streaming error assertion to expect that stable public contract rather than
“upstream failed”.
| -d '{"message": "Explain the concept of tawakkul in Islam", "chat_id": "demo"}' | ||
| ``` | ||
|
|
||
| Each `delta` event contains a JSON payload like `{"delta": "..."}` and the terminal `done` event contains the `chat_id` and the completed text. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Document the error event too.
The endpoint emits event: error, but this contract only describes delta and done; clients need the failure payload and terminal semantics to handle upstream failures correctly.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@README.md` at line 83, Update the streaming endpoint event documentation in
README.md to include the error event, describing its failure payload and that it
terminates the stream. Keep the existing delta and done event descriptions
intact.
|
Your Pr should point to dev branch please, fix conflicts so I can merge asap @Myart352 |
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
main.py (3)
847-906: 🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy liftDo not let
/chat/streambypass the safety pipeline.This route sends the prompt directly to Gemini, skipping the input gate and output enforcement used by
/chat. A caller can select the streaming route to bypass policy refusals and output replacement. Apply an equivalent stream-safe policy flow before emitting any model delta; post-processing only after emission is too late.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@main.py` around lines 847 - 906, The event_generator flow around active_chats[chat_id].send_message must apply the same input safety gate and streaming-safe output enforcement as /chat before yielding any model delta. Reject or replace unsafe prompts before invoking Gemini, and ensure each streamed response chunk is policy-checked or withheld/replaced before emission; do not rely on post-processing after chunks are yielded.
862-938: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winEmit named SSE events and include the completed text.
Without
event: delta,event: done, andevent: errorlines, SSE clients receive only the defaultmessageevent, so listeners for the documented event names—and the supplied tests—fail. Also accumulate emitted chunks and include the final text in thedonepayload.Proposed contract fix
+ completed_parts = [] metadata = json.dumps({"type": "metadata", "chat_id": chat_id, "language": effective_language}) - yield f"data: {metadata}\n\n" + yield f"event: metadata\ndata: {metadata}\n\n" for chunk in response_stream: if chunk.text: - content_data = json.dumps({"type": "content", "text": chunk.text}) - yield f"data: {content_data}\n\n" + completed_parts.append(chunk.text) + content_data = json.dumps({"type": "delta", "text": chunk.text}) + yield f"event: delta\ndata: {content_data}\n\n" - done_data = json.dumps({"type": "done", "chat_id": chat_id, "history": [m.model_dump() for m in history]}) - yield f"data: {done_data}\n\n" + done_data = json.dumps({ + "type": "done", + "chat_id": chat_id, + "text": "".join(completed_parts), + "history": [m.model_dump() for m in history], + }) + yield f"event: done\ndata: {done_data}\n\n" - yield f"data: {error_data}\n\n" + yield f"event: error\ndata: {error_data}\n\n"🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@main.py` around lines 862 - 938, Update the streaming generator around the chunk loop and terminal responses to emit named SSE events: add event: delta before each content payload, event: done before the completion payload, and event: error before the error payload. Accumulate each emitted chunk’s text and include the resulting complete response text in the done_data payload while preserving the existing chat history and error details.
867-909: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winUse the async chat API here.
send_message()is synchronous, so calling it inside this async SSE generator blocks the event loop during model generation. Switch this path toget_model()andawait chat_session.send_message_async(..., stream=True), then consume the stream withasync forso it stays non-blocking and matches the async test seam.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@main.py` around lines 867 - 909, Update the streaming response path to use get_model() and retain the returned async chat session in active_chats. Replace the synchronous send_message call with awaited send_message_async(..., stream=True), and consume response_stream using async for so the SSE generator remains non-blocking and matches the async test seam.Source: Path instructions
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@requirements.txt`:
- Around line 7-8: Pin pytest and pytest-asyncio to exact versions in
requirements.txt, and update the CI installation step to rely on those pinned
requirements instead of reinstalling unversioned packages. Preserve the existing
flake8 installation behavior.
---
Outside diff comments:
In `@main.py`:
- Around line 847-906: The event_generator flow around
active_chats[chat_id].send_message must apply the same input safety gate and
streaming-safe output enforcement as /chat before yielding any model delta.
Reject or replace unsafe prompts before invoking Gemini, and ensure each
streamed response chunk is policy-checked or withheld/replaced before emission;
do not rely on post-processing after chunks are yielded.
- Around line 862-938: Update the streaming generator around the chunk loop and
terminal responses to emit named SSE events: add event: delta before each
content payload, event: done before the completion payload, and event: error
before the error payload. Accumulate each emitted chunk’s text and include the
resulting complete response text in the done_data payload while preserving the
existing chat history and error details.
- Around line 867-909: Update the streaming response path to use get_model() and
retain the returned async chat session in active_chats. Replace the synchronous
send_message call with awaited send_message_async(..., stream=True), and consume
response_stream using async for so the SSE generator remains non-blocking and
matches the async test seam.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 548b19b2-5485-48c9-bd08-d2371600a76a
📒 Files selected for processing (3)
README.mdmain.pyrequirements.txt
| pytest | ||
| pytest-asyncio |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
fd -a 'requirements*.txt|pyproject.toml|pytest.ini|tox.ini|.python-version|.tool-versions|Dockerfile' . -E venv
rg -n --hidden -g '!venv/**' \
'pytest(-asyncio)?|setup-python|python-version' \
requirements.txt .github 2>/dev/null || trueRepository: Deen-Bridge/dnb-ai
Length of output: 1573
🏁 Script executed:
sed -n '1,120p' requirements.txt && printf '\n---\n' && sed -n '1,120p' .github/workflows/ci.yml && printf '\n---\n' && sed -n '1,120p' pytest.iniRepository: Deen-Bridge/dnb-ai
Length of output: 3537
Pin pytest and pytest-asyncio in the test environment. requirements.txt should carry exact versions for these new test deps, and the CI job currently installs them again with pip install pytest flake8 pytest-asyncio, which leaves the test setup open to upstream drift.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@requirements.txt` around lines 7 - 8, Pin pytest and pytest-asyncio to exact
versions in requirements.txt, and update the CI installation step to rely on
those pinned requirements instead of reinstalling unversioned packages. Preserve
the existing flake8 installation behavior.
|
@Myart352 Please fix CI |
2 participants
Closes #8
Summary
add a streaming chat endpoint that returns Server-Sent Events
stream incremental text deltas to the client for faster perceived responsiveness
emit a terminal done event with the chat id and completed text
emit a structured error event if upstream generation fails
preserve the existing non-streaming /chat endpoint for current clients
document the new endpoint and provide a working curl example
Testing
added regression tests for normal streaming and upstream error handling
verified the change with pytest
Summary by CodeRabbit
curlexample and the streamed event payload shapes.