sune-org/ORP
1
0
Fork 0
mirror of https://github.com/sune-org/ORP.git synced 2026-01-13 16:17:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Feat: Support streaming image generation deltas

Browse Source
This commit is contained in:
multipleof4
2025-11-15 18:24:36 -08:00
parent fd2e44d324
commit 251f4b4e19
1 changed files with 25 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
index.js
View File

@@ -67,6 +67,7 @@ export class MyDurableObject {
this.controller = null; this.controller = null;
this.oaStream = null; this.oaStream = null;
this.pending = ''; this.pending = '';
this.pendingImages = [];
this.flushTimer = null; this.flushTimer = null;
this.lastSavedAt = 0; this.lastSavedAt = 0;
this.lastFlushedAt = 0; this.lastFlushedAt = 0;
@@ -85,7 +86,7 @@ export class MyDurableObject {
getConversationText() { getConversationText() {
const prompt = (this.messages || []).map(m => `## ${m.role}\n\n${this.extractTextFromMessage(m)}`).join('\n\n---\n\n'); const prompt = (this.messages || []).map(m => `## ${m.role}\n\n${this.extractTextFromMessage(m)}`).join('\n\n---\n\n');
const response = this.buffer.map(it => it.text).join(''); const response = this.buffer.map(it => it.text || '').join('');
if (!prompt && !response) return ''; if (!prompt && !response) return '';
return `${prompt}\n\n---\n\n## assistant\n\n${response}`; return `${prompt}\n\n---\n\n## assistant\n\n${response}`;
} }
@@ -125,6 +126,7 @@ export class MyDurableObject {
this.error = snap.error || null; this.error = snap.error || null;
this.messages = Array.isArray(snap.messages) ? snap.messages : []; this.messages = Array.isArray(snap.messages) ? snap.messages : [];
this.pending = ''; this.pending = '';
this.pendingImages = [];
if (this.phase === 'running') { if (this.phase === 'running') {
this.phase = 'evicted'; this.phase = 'evicted';
@@ -142,17 +144,21 @@ export class MyDurableObject {
} }
replay(ws, after) { replay(ws, after) {
this.buffer.forEach(it => { if (it.seq > after) this.send(ws, { type: 'delta', seq: it.seq, text: it.text }); }); this.buffer.forEach(it => { if (it.seq > after) this.send(ws, { type: 'delta', seq: it.seq, text: it.text, images: it.images }); });
if (this.phase === 'done') this.send(ws, { type: 'done' }); if (this.phase === 'done') this.send(ws, { type: 'done' });
else if (['error', 'evicted'].includes(this.phase)) this.send(ws, { type: 'err', message: this.error || 'The run was terminated unexpectedly.' }); else if (['error', 'evicted'].includes(this.phase)) this.send(ws, { type: 'err', message: this.error || 'The run was terminated unexpectedly.' });
} }
flush(force = false) { flush(force = false) {
if (this.flushTimer) { clearTimeout(this.flushTimer); this.flushTimer = null; } if (this.flushTimer) { clearTimeout(this.flushTimer); this.flushTimer = null; }
if (this.pending) { if (this.pending || this.pendingImages.length > 0) {
this.buffer.push({ seq: ++this.seq, text: this.pending }); const payload = { type: 'delta', seq: ++this.seq };
this.bcast({ type: 'delta', seq: this.seq, text: this.pending }); if (this.pending) payload.text = this.pending;
if (this.pendingImages.length > 0) payload.images = this.pendingImages;
this.buffer.push(payload);
this.bcast(payload);
this.pending = ''; this.pending = '';
this.pendingImages = [];
this.lastFlushedAt = Date.now(); this.lastFlushedAt = Date.now();
} }
if (force) this.saveSnapshot(); if (force) this.saveSnapshot();
@@ -165,6 +171,12 @@ export class MyDurableObject {
else if (!this.flushTimer) this.flushTimer = setTimeout(() => this.flush(false), BATCH_MS); else if (!this.flushTimer) this.flushTimer = setTimeout(() => this.flush(false), BATCH_MS);
} }
queueImages(images) {
if (!Array.isArray(images) || images.length === 0) return;
this.pendingImages.push(...images);
if (!this.flushTimer) this.flushTimer = setTimeout(() => this.flush(false), BATCH_MS);
}
async fetch(req) { async fetch(req) {
if (req.method === 'OPTIONS') return new Response(null, { status: 204, headers: CORS_HEADERS }); if (req.method === 'OPTIONS') return new Response(null, { status: 204, headers: CORS_HEADERS });
@@ -179,10 +191,12 @@ export class MyDurableObject {
if (req.method === 'GET') { if (req.method === 'GET') {
await this.autopsy(); await this.autopsy();
const text = this.buffer.map(it => it.text).join('') + this.pending; const text = this.buffer.map(it => it.text || '').join('') + this.pending;
const images = this.buffer.flatMap(it => it.images || []);
if (this.pendingImages.length > 0) images.push(...this.pendingImages);
const isTerminal = ['done', 'error', 'evicted'].includes(this.phase); const isTerminal = ['done', 'error', 'evicted'].includes(this.phase);
const isError = ['error', 'evicted'].includes(this.phase); const isError = ['error', 'evicted'].includes(this.phase);
const payload = { rid: this.rid, seq: this.seq, phase: this.phase, done: isTerminal, error: isError ? (this.error || 'The run was terminated unexpectedly.') : null, text }; const payload = { rid: this.rid, seq: this.seq, phase: this.phase, done: isTerminal, error: isError ? (this.error || 'The run was terminated unexpectedly.') : null, text, images };
return this.corsJSON(payload); return this.corsJSON(payload);
} }
return this.corsJSON({ error: 'not allowed' }, 405); return this.corsJSON({ error: 'not allowed' }, 405);
@@ -342,6 +356,9 @@ export class MyDurableObject {
this.queueDelta(delta.content); this.queueDelta(delta.content);
hasContent = true; hasContent = true;
} }
if (delta?.images) {
this.queueImages(delta.images);
}
} }
} }
@@ -407,7 +424,7 @@ export class MyDurableObject {
mapContentPartToResponses(part) { mapContentPartToResponses(part) {
const type = part?.type || 'text'; const type = part?.type || 'text';
if (['image_url', 'input_image'].includes(type)) return (part?.image_url?.url || part?.image_url) ? { type: 'input_image', image_url: String(part?.image_url?.url || part?.image_url) } : null; if (['image_url', 'input_image'].includes(type)) return (part?.image_url?.url || part?.image_url) ? { type: 'input_image', image_url: String(part?.image_url?.url || part?.image_url) } : null;
if (['text', 'input_text'].includes(type)) return { type: 'input_text', text: String(type === 'text' ? (part.text ?? part.content ?? '') : (part.text ?? '')) }; if (['text', 'input_text'].includes(type)) return { type: 'input_text', text: String(type === 'text' ? (p.text ?? p.content ?? '') : (p.text ?? '')) };
return { type: 'input_text', text: `[${type}:${part?.file?.filename || 'file'}]` }; return { type: 'input_text', text: `[${type}:${part?.file?.filename || 'file'}]` };
} }