agent/agent-framework
2
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Python: Fix DevUI streaming memory growth and add cross-platform regression coverage (#5221)

Browse Source 
* fix for memory leak in devui

* update async sleep

* remove old func
This commit is contained in:
Eduard van Valkenburg
2026-04-14 11:27:52 +02:00
committed by GitHub
Unverified
parent 7bb0feca59
commit 98e17764a4
6 changed files with 1211 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/devui/frontend/src/services/api.ts
+74 -38
View File
@@ -16,9 +16,10 @@ import type {
import type { AgentFrameworkRequest } from "@/types/agent-framework";
import type { ExtendedResponseStreamEvent } from "@/types/openai";
import {
applyStreamingEventToState,
createStreamingState,
loadStreamingState,
updateStreamingState,
markStreamingCompleted,
saveStreamingState,
clearStreamingState,
} from "./streaming-state";
import { isAbortError } from "@/hooks";
@@ -72,6 +73,7 @@ const DEFAULT_API_BASE_URL =
// Retry configuration for streaming
const RETRY_INTERVAL_MS = 1000; // Base retry interval (will use exponential backoff)
const MAX_RETRY_ATTEMPTS = 10; // Max 10 retries (~30 seconds with exponential backoff)
const STREAMING_STATE_SAVE_INTERVAL_MS = 250;
// Get backend URL from localStorage or default
function getBackendUrl(): string {
@@ -223,7 +225,7 @@ class ApiClient {
chat_client_type: entity.chat_client_type,
context_provider: entity.context_provider,
middleware: entity.middleware,
};
} as AgentInfo;
} else {
// Workflow - prefer executors field, fall back to tools for backward compatibility
const executorList = entity.executors || entity.tools || [];
@@ -263,7 +265,7 @@ class ApiClient {
input_type_name: entity.input_type_name || "Input",
start_executor_id: startExecutorId,
tools: [],
};
} as WorkflowInfo;
}
});
@@ -484,31 +486,65 @@ class ApiClient {
let hasYieldedAnyEvent = false;
let currentResponseId: string | undefined = resumeResponseId;
let lastMessageId: string | undefined = undefined;
let lastStreamingStateSaveAt = 0;
let storedState = conversationId ? loadStreamingState(conversationId) : null;
let streamingState = storedState ? { ...storedState } : null;
const persistStreamingState = (force: boolean = false): void => {
if (!conversationId || !streamingState) {
return;
}
const now = Date.now();
if (!force && now - lastStreamingStateSaveAt < STREAMING_STATE_SAVE_INTERVAL_MS) {
return;
}
lastStreamingStateSaveAt = now;
saveStreamingState({
...streamingState,
timestamp: now,
});
};
const recordStreamingEvent = (event: ExtendedResponseStreamEvent): void => {
if (!conversationId || !currentResponseId) {
return;
}
streamingState = applyStreamingEventToState(
streamingState ?? createStreamingState({
conversationId,
responseId: currentResponseId,
lastMessageId,
lastSequenceNumber,
accumulatedText: storedState?.accumulatedText,
}),
event,
currentResponseId,
lastMessageId
);
const isTextDelta =
event.type === "response.output_text.delta" &&
"delta" in event &&
typeof event.delta === "string" &&
event.delta.length > 0;
persistStreamingState(!isTextDelta);
};
// Try to resume from stored state if conversation ID is provided
if (conversationId) {
const storedState = loadStreamingState(conversationId);
if (storedState) {
// Use stored response ID if no explicit one provided
if (!resumeResponseId) {
currentResponseId = storedState.responseId;
}
lastSequenceNumber = storedState.lastSequenceNumber;
lastMessageId = storedState.lastMessageId;
// Replay stored events only if we're not explicitly resuming
// (explicit resume means the caller already has the events)
if (!resumeResponseId) {
for (const event of storedState.events) {
hasYieldedAnyEvent = true;
yield event;
}
} else {
// Mark that we've already seen events up to this sequence number
hasYieldedAnyEvent = storedState.events.length > 0;
}
if (storedState) {
// Use stored response ID if no explicit one provided
if (!resumeResponseId) {
currentResponseId = storedState.responseId;
}
lastSequenceNumber = storedState.lastSequenceNumber;
lastMessageId = storedState.lastMessageId;
hasYieldedAnyEvent =
storedState.lastSequenceNumber >= 0 ||
Boolean(storedState.accumulatedText);
}
while (retryCount <= MAX_RETRY_ATTEMPTS) {
@@ -621,7 +657,8 @@ class ApiClient {
if (done) {
// Stream completed successfully
if (conversationId) {
markStreamingCompleted(conversationId);
clearStreamingState(conversationId);
streamingState = null;
}
return;
}
@@ -640,7 +677,8 @@ class ApiClient {
// Handle [DONE] signal
if (dataStr === "[DONE]") {
if (conversationId) {
markStreamingCompleted(conversationId);
clearStreamingState(conversationId);
streamingState = null;
}
return;
}
@@ -676,6 +714,9 @@ class ApiClient {
if (conversationId) {
clearStreamingState(conversationId);
}
storedState = null;
streamingState = null;
lastStreamingStateSaveAt = 0;
yield {
type: "error",
message: "Connection lost - previous response failed. Starting new response.",
@@ -684,9 +725,7 @@ class ApiClient {
hasYieldedAnyEvent = true;
// Save new event to storage
if (conversationId && currentResponseId) {
updateStreamingState(conversationId, openAIEvent, currentResponseId, lastMessageId);
}
recordStreamingEvent(openAIEvent);
yield openAIEvent;
}
@@ -698,9 +737,7 @@ class ApiClient {
hasYieldedAnyEvent = true;
// Save event to storage before yielding
if (conversationId && currentResponseId) {
updateStreamingState(conversationId, openAIEvent, currentResponseId, lastMessageId);
}
recordStreamingEvent(openAIEvent);
yield openAIEvent;
}
@@ -709,9 +746,7 @@ class ApiClient {
hasYieldedAnyEvent = true;
// Still save to storage if we have conversation context
if (conversationId && currentResponseId) {
updateStreamingState(conversationId, openAIEvent, currentResponseId, lastMessageId);
}
recordStreamingEvent(openAIEvent);
yield openAIEvent;
}
@@ -730,7 +765,8 @@ class ApiClient {
// Don't retry on abort
if (isAbortError(error)) {
if (conversationId) {
markStreamingCompleted(conversationId); // Clean up state
clearStreamingState(conversationId);
streamingState = null;
}
throw error; // Re-throw abort error without retrying
}
python/packages/devui/frontend/src/services/streaming-state.ts
+83 -71
View File
@@ -3,7 +3,6 @@
*
* Manages browser storage of streaming response state to enable:
* - Resume interrupted streams after page refresh
* - Replay cached events before fetching new ones
* - Graceful recovery from network disconnections
*/
@@ -14,7 +13,6 @@ export interface StreamingState {
responseId: string;
lastMessageId?: string;
lastSequenceNumber: number;
events: ExtendedResponseStreamEvent[];
timestamp: number; // When this state was last updated
completed: boolean; // Whether the stream completed successfully
accumulatedText?: string; // Accumulated text content for quick restoration
@@ -23,6 +21,14 @@ export interface StreamingState {
const STORAGE_KEY_PREFIX = "devui_streaming_state_";
const STATE_EXPIRY_MS = 24 * 60 * 60 * 1000; // 24 hours
interface CreateStreamingStateOptions {
conversationId: string;
responseId: string;
lastMessageId?: string;
lastSequenceNumber?: number;
accumulatedText?: string;
}
/**
* Storage key for a specific conversation
*/
@@ -31,16 +37,81 @@ function getStorageKey(conversationId: string): string {
}
/**
* Extract accumulated text from events (for quick restoration)
* Read raw streaming state from storage, including completed entries.
*/
function extractAccumulatedText(events: ExtendedResponseStreamEvent[]): string {
let text = "";
for (const event of events) {
if (event.type === "response.output_text.delta" && "delta" in event) {
text += event.delta;
}
function readStreamingState(conversationId: string): StreamingState | null {
const key = getStorageKey(conversationId);
const data = localStorage.getItem(key);
if (!data) {
return null;
}
return text;
const state: StreamingState = JSON.parse(data);
// Check if state has expired
const age = Date.now() - state.timestamp;
if (age > STATE_EXPIRY_MS) {
clearStreamingState(conversationId);
return null;
}
return state;
}
/**
* Create an initial streaming state snapshot.
*/
export function createStreamingState({
conversationId,
responseId,
lastMessageId,
lastSequenceNumber = -1,
accumulatedText,
}: CreateStreamingStateOptions): StreamingState {
return {
conversationId,
responseId,
lastMessageId,
lastSequenceNumber,
timestamp: Date.now(),
completed: false,
accumulatedText,
};
}
/**
* Apply an incoming stream event to an in-memory streaming state snapshot.
*/
export function applyStreamingEventToState(
state: StreamingState,
event: ExtendedResponseStreamEvent,
responseId: string,
lastMessageId?: string
): StreamingState {
const sequenceNumber = "sequence_number" in event ? event.sequence_number : undefined;
const nextState: StreamingState = {
...state,
responseId,
lastMessageId,
timestamp: Date.now(),
completed: event.type === "response.completed" || event.type === "response.failed",
};
if (sequenceNumber !== undefined) {
nextState.lastSequenceNumber = sequenceNumber;
}
if (
event.type === "response.output_text.delta" &&
"delta" in event &&
typeof event.delta === "string" &&
event.delta.length > 0
) {
nextState.accumulatedText = `${state.accumulatedText ?? ""}${event.delta}`;
}
return nextState;
}
/**
@@ -71,19 +142,8 @@ export function saveStreamingState(state: StreamingState): void {
*/
export function loadStreamingState(conversationId: string): StreamingState | null {
try {
const key = getStorageKey(conversationId);
const data = localStorage.getItem(key);
if (!data) {
return null;
}
const state: StreamingState = JSON.parse(data);
// Check if state has expired
const age = Date.now() - state.timestamp;
if (age > STATE_EXPIRY_MS) {
clearStreamingState(conversationId);
const state = readStreamingState(conversationId);
if (!state) {
return null;
}
@@ -99,54 +159,6 @@ export function loadStreamingState(conversationId: string): StreamingState | nul
}
}
/**
* Update streaming state with a new event
*/
export function updateStreamingState(
conversationId: string,
event: ExtendedResponseStreamEvent,
responseId: string,
lastMessageId?: string
): void {
try {
const existing = loadStreamingState(conversationId);
const sequenceNumber = "sequence_number" in event ? event.sequence_number : undefined;
const newEvents = existing ? [...existing.events, event] : [event];
const state: StreamingState = {
conversationId,
responseId,
lastMessageId,
lastSequenceNumber: sequenceNumber ?? (existing?.lastSequenceNumber ?? -1),
events: newEvents,
timestamp: Date.now(),
completed: event.type === "response.completed" || event.type === "response.failed",
accumulatedText: extractAccumulatedText(newEvents),
};
saveStreamingState(state);
} catch (error) {
console.error("Failed to update streaming state:", error);
}
}
/**
* Mark streaming state as completed
*/
export function markStreamingCompleted(conversationId: string): void {
try {
const existing = loadStreamingState(conversationId);
if (existing) {
existing.completed = true;
existing.timestamp = Date.now();
saveStreamingState(existing);
}
} catch (error) {
console.error("Failed to mark streaming as completed:", error);
}
}
/**
* Clear streaming state for a conversation
*/