Feat : Add SSE streaming endpoint

[Clyde0909]

• Reduce perceived latency by increasing state transparency for long-running responses

Summary

Some VDB queries in the RAG pipeline can take a long time, and users had no visibility into whether their requests were still being processed.
This PR introduces SSE-based streaming to make the request state visible during long-running operations.

Changes

• Add SSE streaming endpoint
• Stream initial responses and status updates for long-running RAG requests

Impact

• Improves state transparency for slow VDB queries
• Reduces perceived latency and user confusion

[Copilot]

Pull request overview

This pull request adds Server-Sent Events (SSE) streaming functionality to the RAG pipeline to improve transparency and reduce perceived latency during long-running vector database queries. It introduces a new streaming endpoint that sends real-time progress updates as queries are processed.

Changes:

• Added new SSE streaming endpoint /answerQuestion/stream using GET method
• Created async generator async_gen_process_sql_category to support streaming responses with intermediate progress updates
• Implemented structured SSE message format with status indicators for vector search completion, VQL generation, and final results

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 13 comments.

File	Description
api/utils/sdk_answer_question.py	Added async generator function to process SQL category questions with streaming support, yielding intermediate results during VQL generation and execution
api/endpoints/answerQuestion.py	Added new GET endpoint for SSE streaming and generator function to orchestrate the streaming response with progress updates

---

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

[Copilot]

The async generator function lacks a docstring explaining its purpose, parameters, yields, and behavior. This is especially important for generator functions as they have different execution patterns. Add a comprehensive docstring documenting the function's purpose (streaming SQL category processing), its parameters, what it yields (dict with 'type' and 'data' keys), and any important behavioral notes about the streaming protocol.

Suggested change

	async def async_gen_process_sql_category(request, vector_search_tables, sql_gen_llm, chat_llm, category_response, auth, timings, session_id = None, sample_data = None):
	async def async_gen_process_sql_category(request, vector_search_tables, sql_gen_llm, chat_llm, category_response, auth, timings, session_id = None, sample_data = None):
	"""
	Stream the processing of a SQL (VQL) category question and its LLM-based enrichment.
	This async generator mirrors the behavior of ``process_sql_category`` but exposes
	its work as a sequence of streaming events. It incrementally performs VQL query
	generation, optional query fixing, execution, and response enrichment, yielding
	structured messages that callers can consume as they arrive instead of waiting
	for the full result.
	Parameters
	----------
	request :
	An object representing the incoming question and options. It is expected to
	provide fields such as ``question``, ``custom_instructions``,
	``vector_search_sample_data_k``, and flags that influence response shape
	(for example, verbosity or plotting options).
	vector_search_tables :
	Metadata or configuration describing the tables, views, or indices that can
	be used for vector / semantic search when generating and executing VQL.
	sql_gen_llm :
	LLM client used for translating the natural language question into a VQL
	query and for any follow‑up SQL/VQL related reasoning.
	chat_llm :
	LLM client used primarily for conversational or explanatory responses such
	as natural language answers, explanations, and related questions.
	category_response :
	Parsed category information and filters derived from earlier classification
	steps, used to constrain VQL generation (for example, selected schema,
	metrics, or dimensions).
	auth :
	Authentication or authorization context associated with the request. Passed
	through to lower‑level helpers that may require user or tenant information.
	timings : dict
	Mutable dictionary used to accumulate timing information for different
	stages of processing. This generator updates the ``"llm_time"`` entry and
	any other timing keys used by helper functions.
	session_id : Optional[str], default None
	Identifier for the current session / conversation, forwarded to LLM
	backends and logging or tracing utilities.
	sample_data :
	Optional sample data or configuration used to augment vector search or
	improve query generation quality.
	Yields
	------
	dict
	A streaming event represented as a dictionary with at least:
	* ``"type"``: A string describing the kind of event (for example,
	``"result"`` for the final structured response).
	* ``"data"``: The associated payload for that event. For a final
	``"result"`` event this is typically the complete response object
	including query information, answer text, related questions,
	tokens, timings, and any generated graph data.
	Notes
	-----
	* The generator may yield multiple events over the lifetime of a single
	request. Consumers should iterate until completion rather than assuming
	a single value.
	* When no valid VQL can be generated, the generator yields a final
	``"result"`` event whose ``"data"`` includes an explanatory message in
	the ``"answer"`` field.
	* Timing information in ``timings`` is updated as processing progresses;
	the final ``"result"`` reflects the accumulated timings at the end of
	the pipeline.
	"""

[Copilot]

Inconsistent spacing in function parameter list. The parameters 'session_id = None' and 'sample_data = None' have spaces around the equals sign, but PEP 8 style guide recommends no spaces around equals signs for default parameter values. This should be 'session_id=None' and 'sample_data=None' to follow Python conventions and be consistent with standard formatting.

[Copilot]

Commented-out print statement should be removed. Debug code like '# print(c_data)' should not be committed to the codebase. Either remove it entirely or replace it with proper logging if the information is needed for debugging.

Suggested change

# print(c_data)

[Copilot]

The early exit path when no VQL query is generated does not return after yielding the result. This causes the code to continue executing and call query_fixer with a None/empty vql_query, which will likely cause an error. Add a return statement after the yield on line 405 to properly exit the function.

Suggested change

	return

[Copilot]

The function uses GET method for a streaming endpoint that processes questions. GET requests typically don't support request bodies, but the code expects answerQuestionRequest from Query parameters. Complex objects like answerQuestionRequest are not well-suited for query parameters. Consider using POST method instead, which is more appropriate for sending structured data and processing operations.

[Copilot]

There is significant code duplication between async_gen_process_sql_category and the existing process_sql_category function. Both functions share almost identical logic for VQL generation, query fixing, execution, and response preparation. Consider refactoring to extract the shared logic into helper functions or making the streaming behavior an optional parameter to avoid maintaining two separate implementations of the same logic.

[Copilot]

Commented-out code should be removed. The comment '# return response' on line 508 is leftover code that serves no purpose and should be deleted. Comments should explain why, not repeat what the code does or contain unused code.

Suggested change

# return response

[Copilot]

Missing error handling for exceptions that could occur after the initial try-except block. Operations like sql_category, async_gen_process_sql_category, process_metadata_category, and process_unknown_category could raise exceptions, but there's no try-except to catch them and yield an appropriate error message to the SSE client. In a streaming context, unhandled exceptions will abruptly terminate the connection. Consider wrapping the main processing logic in a try-except block and yielding error messages in SSE format when exceptions occur.

Suggested change

	with timing_context("llm_time", timings):
	category, category_response, category_related_questions, sql_category_tokens = await sdk_ai_tools.sql_category(
	query=request_data.question,
	vector_search_tables=vector_search_tables,
	llm=llm,
	mode=request_data.mode,
	custom_instructions=request_data.custom_instructions,
	session_id=session_id
	)
	if category == "SQL":
	async for chunk in sdk_answer_question.async_gen_process_sql_category(
	request=request_data,
	vector_search_tables=vector_search_tables,
	category_response=category_response,
	auth=auth,
	timings=timings,
	session_id=session_id,
	sample_data=sample_data,
	chat_llm=llm,
	sql_gen_llm=llm
	):
	event = chunk.get("type")
	c_data = chunk.get("data")
	if event == "query_gen":
	# print(c_data)
	vql_progress_data = {
	"status": "vql_generation_completed",
	"result": {
	"answer": c_data,
	"message": "VQL generation completed"
	}
	}
	yield f"data: {json.dumps(vql_progress_data)}\n\n"
	elif event == "result":
	response = c_data
	response['tokens'] = add_tokens(response['tokens'], sql_category_tokens)
	elif category == "METADATA":
	response = sdk_answer_question.process_metadata_category(
	category_response=category_response,
	category_related_questions=category_related_questions,
	vector_search_tables=vector_search_tables,
	timings=timings,
	tokens=sql_category_tokens,
	disclaimer=request_data.disclaimer
	)
	else:
	response = sdk_answer_question.process_unknown_category(timings=timings)
	response['llm_provider'] = request_data.llm_provider
	response['llm_model'] = request_data.llm_model
	# Send final completion message
	final_data = {
	"status": "completed",
	"result": jsonable_encoder(response)
	}
	yield f"data: {json.dumps(final_data)}\n\n"
	try:
	with timing_context("llm_time", timings):
	category, category_response, category_related_questions, sql_category_tokens = await sdk_ai_tools.sql_category(
	query=request_data.question,
	vector_search_tables=vector_search_tables,
	llm=llm,
	mode=request_data.mode,
	custom_instructions=request_data.custom_instructions,
	session_id=session_id
	)
	if category == "SQL":
	async for chunk in sdk_answer_question.async_gen_process_sql_category(
	request=request_data,
	vector_search_tables=vector_search_tables,
	category_response=category_response,
	auth=auth,
	timings=timings,
	session_id=session_id,
	sample_data=sample_data,
	chat_llm=llm,
	sql_gen_llm=llm
	):
	event = chunk.get("type")
	c_data = chunk.get("data")
	if event == "query_gen":
	# print(c_data)
	vql_progress_data = {
	"status": "vql_generation_completed",
	"result": {
	"answer": c_data,
	"message": "VQL generation completed"
	}
	}
	yield f"data: {json.dumps(vql_progress_data)}\n\n"
	elif event == "result":
	response = c_data
	response['tokens'] = add_tokens(response['tokens'], sql_category_tokens)
	elif category == "METADATA":
	response = sdk_answer_question.process_metadata_category(
	category_response=category_response,
	category_related_questions=category_related_questions,
	vector_search_tables=vector_search_tables,
	timings=timings,
	tokens=sql_category_tokens,
	disclaimer=request_data.disclaimer
	)
	else:
	response = sdk_answer_question.process_unknown_category(timings=timings)
	response['llm_provider'] = request_data.llm_provider
	response['llm_model'] = request_data.llm_model
	# Send final completion message
	final_data = {
	"status": "completed",
	"result": jsonable_encoder(response)
	}
	yield f"data: {json.dumps(final_data)}\n\n"
	except Exception as e:
	# Log the exception and send an SSE-formatted error message
	logging.exception("Error while processing SSE question")
	error_data = {
	"status": "error",
	"error": {
	"message": str(e),
	"type": e.__class__.__name__,
	},
	}
	yield f"data: {json.dumps(error_data)}\n\n"

[Copilot]

When vector_search_tables is empty, the function yields an error message but doesn't include a status field in the JSON. This is inconsistent with other SSE messages which include a "status" field (e.g., "vector_search_completed", "completed"). Consider adding a status field like "error" for consistency and easier client-side parsing.

[Copilot]

The variable 'response' is used before being assigned when category is "METADATA" or unknown. The response variable is only set inside the if statement for category == "SQL", but is then accessed unconditionally at line 404 and later lines. This will cause an UnboundLocalError if the category is not "SQL". Initialize the response variable before the conditional logic or handle all category cases properly.