Skip to content

Stream AI reasoning as ephemeral thinking chunks #1729

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gugu merged 1 commit into from
Apr 23, 2026
Merged

Stream AI reasoning as ephemeral thinking chunks #1729

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 21 additions & 15 deletions backend/src/entities/ai/use-cases/request-info-from-table-with-ai-v7.use.case.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -171,9 +171,8 @@ export class RequestInfoFromTableWithAIUseCaseV7

for await (const chunk of stream) {
if (chunk.type === 'text' && chunk.content) {
response.write(chunk.content);
accumulatedContent += chunk.content;
totalAccumulatedResponse += chunk.content;
this.writeChunk(response, { type: 'thinking', content: chunk.content });
}

if (chunk.type === 'tool_call' && chunk.toolCall) {
Expand All @@ -191,13 +190,15 @@ export class RequestInfoFromTableWithAIUseCaseV7
);

if (pendingToolCalls.length === 0) {
this.writeChunk(response, { type: 'thinking_commit' });
totalAccumulatedResponse += accumulatedContent;
break;
}

this.writeChunk(response, { type: 'thinking_reset' });

for (const toolCall of pendingToolCalls) {
this.logger.log(
`Tool call: ${toolCall.name}, arguments=${JSON.stringify(toolCall.arguments)}`,
);
this.logger.log(`Tool call: ${toolCall.name}, arguments=${JSON.stringify(toolCall.arguments)}`);
}

const toolResults = await this.executeToolCalls(
Expand All @@ -209,9 +210,7 @@ export class RequestInfoFromTableWithAIUseCaseV7
);

for (const toolResult of toolResults) {
this.logger.log(
`Tool result for ${toolResult.toolCallId}: resultLength=${toolResult.result.length}`,
);
this.logger.log(`Tool result for ${toolResult.toolCallId}: resultLength=${toolResult.result.length}`);
}

if (this.aiProvider === AIProviderType.OPENAI && lastResponseId) {
Expand All @@ -227,9 +226,7 @@ export class RequestInfoFromTableWithAIUseCaseV7

depth++;
} catch (loopError) {
this.logger.error(
`Error in tool loop at depth ${depth + 1}: ${loopError.message}`,
);
this.logger.error(`Error in tool loop at depth ${depth + 1}: ${loopError.message}`);
throw loopError;
}
}
Expand All @@ -238,13 +235,24 @@ export class RequestInfoFromTableWithAIUseCaseV7
this.logger.warn(`Tool loop reached max depth (${this.maxDepth})`);
const maxDepthMessage =
'\n\nYour question is too complex to process at this time. Please try simplifying it or breaking it down into smaller parts.';
response.write(maxDepthMessage);
this.writeChunk(response, { type: 'text', content: maxDepthMessage });
totalAccumulatedResponse += maxDepthMessage;
}

return { lastResponseId, accumulatedResponse: totalAccumulatedResponse };
}

private writeChunk(
response: Response,
chunk:
| { type: 'thinking'; content: string }
| { type: 'thinking_reset' }
| { type: 'thinking_commit' }
| { type: 'text'; content: string },
): void {
response.write(JSON.stringify(chunk) + '\n');
Copy link

Copilot AI Apr 23, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The stream is now NDJSON-framed (JSON.stringify(chunk) + "\n"), but the response headers are still configured as text/event-stream in setupResponseHeaders(). This content-type mismatch can confuse intermediaries/clients and is not a valid SSE payload format. Please update the headers to reflect NDJSON (e.g., application/x-ndjson; charset=utf-8) or change the payload to proper SSE framing if SSE is intended.

Suggested change
response.write(JSON.stringify(chunk) + '\n');
response.write(`data: ${JSON.stringify(chunk)}\n\n`);

Copilot uses AI. Check for mistakes.
}

private async executeToolCalls(
toolCalls: AIToolCall[],
dataAccessObject: IDataAccessObject | IDataAccessObjectAgent,
Expand Down Expand Up @@ -306,9 +314,7 @@ export class RequestInfoFromTableWithAIUseCaseV7
result = encodeError({ error: `Unknown tool: ${toolCall.name}` });
}
} catch (error) {
this.logger.error(
`Tool call ${toolCall.name} (${toolCall.id}) failed: ${error.message}`,
);
this.logger.error(`Tool call ${toolCall.name} (${toolCall.id}) failed: ${error.message}`);
result = encodeError({ error: error.message });
}

Expand Down
15 changes: 15 additions & 0 deletions .../app/components/dashboard/db-table-view/db-table-ai-panel/db-table-ai-panel.component.css
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -298,6 +298,21 @@
}
}

.ai-thinking {
font-size: 12px;
line-height: 1.5;
color: rgba(0, 0, 0, 0.5);
font-style: italic;
white-space: pre-wrap;
word-break: break-word;
}

@media (prefers-color-scheme: dark) {
.ai-thinking {
color: rgba(255, 255, 255, 0.5);
}
}

.ai-error-message {
border-radius: 8px;
padding: 8px 8px 0;
Expand Down
13 changes: 8 additions & 5 deletions ...app/components/dashboard/db-table-view/db-table-ai-panel/db-table-ai-panel.component.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,10 +9,10 @@ <h2 class="mat-heading-2 ai-panel-sidebar__title">
<span *ngIf="!isExpanded">AI insights</span>
</h2>
<div class="ai-panel-sidebar__actions">
<button mat-icon-button (click)="toggleExpand()">
<button type="button" mat-icon-button (click)="toggleExpand()">
<mat-icon>{{isExpanded ? 'close_fullscreen' : 'open_in_full'}}</mat-icon>
</button>
<button mat-icon-button (click)="handleClose()">
<button type="button" mat-icon-button (click)="handleClose()">
<mat-icon>close</mat-icon>
</button>
</div>
Expand All @@ -23,7 +23,10 @@ <h2 class="mat-heading-2 ai-panel-sidebar__title">
<div class="ai-message-chain">
<div *ngFor="let message of messagesChain" class="{{message.type}}-message">
<span style="white-space: pre-wrap" *ngIf="message.type == 'user'">{{message.text}}</span>
<markdown *ngIf="message.type == 'ai'" mermaid [data]="message.text"></markdown>
<ng-container *ngIf="message.type == 'ai'">
<div *ngIf="message.thinking && !message.text" class="ai-thinking">{{ message.thinking }}</div>
<markdown *ngIf="message.text" mermaid [data]="message.text"></markdown>
</ng-container>
</div>
</div>
</div>
Expand Down Expand Up @@ -54,7 +57,7 @@ <h2 class="mat-heading-2 ai-panel-sidebar__title">

<div class="ai-completions" *ngIf="showCompletions && activeCompletions.length"
(mouseleave)="onCompletionMouseLeave()">
<button *ngFor="let completion of activeCompletions"
<button type="button" *ngFor="let completion of activeCompletions"
class="ai-completion-item"
(mouseenter)="onCompletionHover(completion)"
(mousedown)="selectCompletion(completion)">
Expand All @@ -67,7 +70,7 @@ <h2 class="mat-heading-2 ai-panel-sidebar__title">
<div class="ai-category" *ngFor="let category of suggestionCategories">
<p class="ai-category__title">{{category.title}}</p>
<div class="ai-category__chips">
<button *ngFor="let suggestion of category.suggestions"
<button type="button" *ngFor="let suggestion of category.suggestions"
class="suggestion-chip"
[class.suggestion-chip_active]="activeSuggestion?.title === suggestion.title"
(click)="onSuggestionChipClick(suggestion)">
Expand Down
27 changes: 21 additions & 6 deletions ...c/app/components/dashboard/db-table-view/db-table-ai-panel/db-table-ai-panel.component.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import { Angulartics2, Angulartics2Module } from 'angulartics2';
import { MarkdownModule } from 'ngx-markdown';
import posthog from 'posthog-js';
import { AiService } from 'src/app/services/ai.service';
import { AiStreamChunk } from 'src/app/services/api.service';
import { ConnectionsService } from 'src/app/services/connections.service';
import { TableStateService } from 'src/app/services/table-state.service';
import { TablesService } from 'src/app/services/tables.service';
Expand Down Expand Up @@ -53,6 +54,7 @@ export class DbTableAiPanelComponent implements OnInit, AfterViewInit, OnDestroy
public messagesChain: {
type: string;
text: string;
thinking?: string;
}[] = [];
public aiSuggestions: { title: string; prompt: string; completions: string[] }[] = [];
public suggestionCategories: {
Expand All @@ -69,7 +71,7 @@ export class DbTableAiPanelComponent implements OnInit, AfterViewInit, OnDestroy

private _aiService = inject(AiService);
private _abortController: AbortController | null = null;
private _loadingStepsInterval: any = null;
private _loadingStepsInterval: ReturnType<typeof setInterval> | null = null;
private _loadingSteps: string[] = [
'Connecting to database',
'Analyzing table structure',
Expand Down Expand Up @@ -109,7 +111,7 @@ export class DbTableAiPanelComponent implements OnInit, AfterViewInit, OnDestroy

async ngAfterViewInit() {
const mermaid = await import('mermaid');
const mermaidAPI: any = mermaid.default ?? mermaid;
const mermaidAPI = (mermaid.default ?? mermaid) as { initialize: (config: Record<string, unknown>) => void };
const isDark = window.matchMedia('(prefers-color-scheme: dark)').matches;
mermaidAPI.initialize({
startOnLoad: false,
Expand Down Expand Up @@ -216,7 +218,7 @@ export class DbTableAiPanelComponent implements OnInit, AfterViewInit, OnDestroy

if (response) {
this.threadID = response.threadId;
const aiMessage = { type: 'ai', text: '' };
const aiMessage = { type: 'ai', text: '', thinking: '' };
this.messagesChain.push(aiMessage);
this.stopLoadingSteps();

Expand Down Expand Up @@ -269,7 +271,7 @@ export class DbTableAiPanelComponent implements OnInit, AfterViewInit, OnDestroy
);

if (stream) {
const aiMessage = { type: 'ai', text: '' };
const aiMessage = { type: 'ai', text: '', thinking: '' };
this.messagesChain.push(aiMessage);
this.stopLoadingSteps();

Expand All @@ -296,9 +298,22 @@ export class DbTableAiPanelComponent implements OnInit, AfterViewInit, OnDestroy
}
}

private async _consumeStream(stream: AsyncGenerator<string>, message: { type: string; text: string }) {
private async _consumeStream(
stream: AsyncGenerator<AiStreamChunk>,
message: { type: string; text: string; thinking?: string },
) {
for await (const chunk of stream) {
message.text += chunk;
if (chunk.type === 'thinking') {
message.thinking = (message.thinking ?? '') + chunk.content;
} else if (chunk.type === 'thinking_reset') {
message.thinking = '';
} else if (chunk.type === 'thinking_commit') {
message.text += message.thinking ?? '';
message.thinking = '';
} else if (chunk.type === 'text') {
message.text += chunk.content;
message.thinking = '';
}
}
}

Expand Down
6 changes: 3 additions & 3 deletions frontend/src/app/services/ai.service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,9 @@
import { Injectable, inject } from '@angular/core';
import { ApiService } from './api.service';
import { AiStreamChunk, ApiService } from './api.service';

export interface AiStreamResponse {
threadId: string;
stream: AsyncGenerator<string>;
stream: AsyncGenerator<AiStreamChunk>;
}

@Injectable({
Expand Down Expand Up @@ -38,7 +38,7 @@ export class AiService {
threadId: string,
message: string,
signal?: AbortSignal,
): Promise<AsyncGenerator<string> | null> {
): Promise<AsyncGenerator<AiStreamChunk> | null> {
const result = await this._api.postStream(
`/ai/v4/request/${connectionId}`,
{ user_message: message },
Expand Down
50 changes: 42 additions & 8 deletions frontend/src/app/services/api.service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,24 @@ export interface ApiResponse<T> {
status: number;
}

export type AiStreamChunk =
| { type: 'thinking'; content: string }
| { type: 'thinking_reset' }
| { type: 'thinking_commit' }
| { type: 'text'; content: string };

function tryParseChunk(line: string): AiStreamChunk | null {
try {
const parsed = JSON.parse(line);
if (parsed && typeof parsed.type === 'string') {
return parsed as AiStreamChunk;
}
Comment on lines +30 to +33
Copy link

Copilot AI Apr 23, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

tryParseChunk currently accepts any JSON object with a string type and then casts it to AiStreamChunk. This can yield invalid shapes at runtime (e.g., {type:"thinking"} without content, or unknown type values), which will later be treated as valid and can append undefined into the UI. Consider validating type against the allowed set and, for thinking/text, also validating typeof content === 'string' before returning.

Suggested change
const parsed = JSON.parse(line);
if (parsed && typeof parsed.type === 'string') {
return parsed as AiStreamChunk;
}
const parsed: unknown = JSON.parse(line);
if (!parsed || typeof parsed !== 'object') {
return null;
}
const chunk = parsed as { type?: unknown; content?: unknown };
switch (chunk.type) {
case 'thinking':
return typeof chunk.content === 'string'
? { type: 'thinking', content: chunk.content }
: null;
case 'thinking_reset':
return { type: 'thinking_reset' };
case 'thinking_commit':
return { type: 'thinking_commit' };
case 'text':
return typeof chunk.content === 'string'
? { type: 'text', content: chunk.content }
: null;
default:
return null;
}

Copilot uses AI. Check for mistakes.
} catch {
// ignore malformed line
}
return null;
}

@Injectable({
providedIn: 'root',
})
Expand Down Expand Up @@ -51,7 +69,7 @@ export class ApiService {
return result?.data ?? null;
}

async postResponse<T>(url: string, body?: unknown, options?: ApiRequestOptions): Promise<ApiResponse<T> | null> {
postResponse<T>(url: string, body?: unknown, options?: ApiRequestOptions): Promise<ApiResponse<T> | null> {
return this._fetchResponse<T>('POST', url, body, options);
}

Expand All @@ -69,7 +87,7 @@ export class ApiService {
url: string,
body?: unknown,
options?: ApiRequestOptions,
): Promise<{ headers: Headers; stream: AsyncGenerator<string> } | null> {
): Promise<{ headers: Headers; stream: AsyncGenerator<AiStreamChunk> } | null> {
try {
const fullUrl = this._buildUrl(url, options?.params);
const headers = this._buildHeaders(options?.headers);
Expand All @@ -96,19 +114,35 @@ export class ApiService {
const reader = response.body.getReader();
const decoder = new TextDecoder();

async function* textStream(): AsyncGenerator<string> {
async function* chunkStream(): AsyncGenerator<AiStreamChunk> {
let buffer = '';
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
yield decoder.decode(value, { stream: true });
if (done) {
const tail = buffer.trim();
if (tail) {
const parsed = tryParseChunk(tail);
if (parsed) yield parsed;
}
break;
}
Comment on lines 121 to +129
Copy link

Copilot AI Apr 23, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In chunkStream(), when done is true you parse the remaining buffer, but you never flush the TextDecoder state. With { stream: true }, multi-byte characters can be buffered internally and will be lost unless you call decoder.decode() once at the end to flush pending bytes before processing the tail.

Copilot uses AI. Check for mistakes.
buffer += decoder.decode(value, { stream: true });
let newlineIdx: number;
while ((newlineIdx = buffer.indexOf('\n')) >= 0) {
const line = buffer.slice(0, newlineIdx).trim();
buffer = buffer.slice(newlineIdx + 1);
if (!line) continue;
const parsed = tryParseChunk(line);
if (parsed) yield parsed;
}
}
} finally {
reader.releaseLock();
}
}

return { headers: response.headers, stream: textStream() };
return { headers: response.headers, stream: chunkStream() };
} catch (err: unknown) {
if ((err as Error).name === 'AbortError') return null;
this._handleError(err, options);
Expand Down Expand Up @@ -188,15 +222,15 @@ export class ApiService {

const gclidMatch = document.cookie.match(/(?:^|;\s*)autoadmin_gclid=([^;]*)/);
if (gclidMatch?.[1]) {
headers['GCLID'] = gclidMatch[1];
headers.GCLID = gclidMatch[1];
}

const pathSegments = location.pathname.split('/');
const connectionId = pathSegments.length >= 3 ? pathSegments[2] : null;
if (connectionId) {
const masterKey = localStorage.getItem(`${connectionId}__masterKey`);
if (masterKey) {
headers['masterpwd'] = masterKey;
headers.masterpwd = masterKey;
}
}

Expand Down
Loading