Stream reasoning

[jmsevin]

This pull request introduces significant improvements to the chat API, especially around streaming agent responses, event formatting, and database handling. It adds a new /chat/agent_stream endpoint for streaming agent responses over Server-Sent Events (SSE), standardizes SSE formatting and headers, improves type safety and serialization, and enhances the handling of chat history and language detection. Additionally, it introduces an option to exclude vectors from search results and modernizes model serialization.

Streaming and SSE Enhancements

• Added a new /chat/agent_stream endpoint to stream agent responses, sending incremental updates and a final payload using Server-Sent Events (SSE). This includes helper functions for formatting SSE events, tracking stream state, and serializing payloads. [1] [2]
• Standardized SSE headers and formatting for all streaming endpoints, ensuring consistent client behavior and improved compatibility. [1] [2] [3]

Database and Type Handling

• Improved database connection handling by using psycopg.AsyncConnection[DictRow] and a dedicated ASYNC_DICT_ROW_FACTORY for type safety and clarity in async row processing. [1] [2]
• Updated agent response registration to only use the returned message_id, simplifying data collection logic.

Model and Serialization Updates

• Enhanced the AgentResponse model to include new fields: status and step, supporting richer streaming updates.
• Updated payload normalization to use model_dump() instead of the deprecated dict() method, aligning with modern Pydantic usage.

Search Improvements

• Added a without_vectors option to the search handler, allowing results to be returned without vector data for efficiency; updated agent resource search to use this option. [1] [2] [3]

Language Detection and Robustness

• Improved language and past message reference detection logic for robustness, ensuring validated and correctly typed responses from LLMs. [1] [2]
• Refactored shared abstractions and typing for agent input state and streaming chunk handling. [1] [2] [3]

Other minor changes include dependency updates and small code cleanups.

[Copilot]

Pull request overview

Note

Copilot was unable to run its full agentic suite in this review.

Adds an SSE streaming endpoint for agent responses and normalizes streaming chunk handling across the infra/service and API layers.

Changes:

• Introduces agent streaming (/chat/agent_stream) with SSE helpers, state aggregation, and final “stop” payload.
• Adds infra support for streaming agent chunks (get_agent_chunks / _extract_agent_chunk) and tighter JSON parsing validation.
• Adds tests for new streaming utilities and agent chunk extraction.

Reviewed changes

Copilot reviewed 10 out of 11 changed files in this pull request and generated 8 comments.

Show a summary per file

File	Description
src/app/shared/infra/abst_chat.py	Adds agent chunk streaming helpers and strengthens LLM JSON parsing/type validation.
src/app/api/api_v1/endpoints/chat.py	Adds SSE helpers and a new /chat/agent_stream endpoint; wraps existing streams as SSE.
src/app/models/chat.py	Extends AgentResponse schema with status and step.
src/app/search/services/search.py	Adds without_vectors option to omit vectors from search results.
src/app/services/agent.py	Uses without_vectors=True when fetching resources.
src/app/services/helpers.py	Switches payload normalization to Pydantic v2 model_dump().
src/app/tests/services/test_abst_chat_utils.py	Adds tests for AbstractChat streaming/chunk extraction and JSON parsing fallback.
src/app/tests/api/api_v1/test_chat_utils.py	Adds unit tests for new chat endpoint helper functions.
src/app/tests/api/api_v1/test_chat.py	Adds an integration-ish test for the new agent SSE streaming endpoint.
pyproject.toml	Updates dependencies (currently with unresolved merge conflict).

Comments suppressed due to low confidence (1)

src/app/shared/infra/abst_chat.py:238

• Same issue here: raise e loses traceback. Also, the broad except Exception: + fallback can obscure the true root cause (e.g., non-iterable stream types). Preserve tracebacks with bare raise, and consider narrowing the exception you treat as a signal to attempt the sync fallback.

        try:
            async for chunk in stream:
                for part in self._extract_stream_chunk(chunk):
                    yield part
        except Exception:
            try:
                for chunk in stream:
                    for part in self._extract_stream_chunk(chunk):
                        yield part
            except Exception as e:
                logger.error("get_stream_chunks api_error=%s", e)
                raise e

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.