@@ -74,6 +74,7 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
7474private playhead = 0;
7575private closed = false;
7676private pendingCalls = new Map<string, PendingFunctionCall>();
77+private readonly sources = new Set<AudioBufferSourceNode>();
77787879constructor(
7980private readonly session: RealtimeTalkJsonPcmWebSocketSessionResult,
@@ -93,11 +94,17 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
9394this.inputContext = new AudioContext({ sampleRate: this.session.audio.inputSampleRateHz });
9495this.outputContext = new AudioContext({ sampleRate: this.session.audio.outputSampleRateHz });
9596this.ws = new WebSocket(wsUrl);
97+this.ws.binaryType = "arraybuffer";
9698this.ws.addEventListener("open", () => {
99+if (this.closed) {
100+return;
101+}
97102this.send(this.session.initialMessage ?? { setup: {} });
98103this.startMicrophonePump();
99104});
100-this.ws.addEventListener("message", (event) => this.handleMessage(event.data));
105+this.ws.addEventListener("message", (event) => {
106+void this.handleMessage(event.data);
107+});
101108this.ws.addEventListener("close", () => {
102109if (!this.closed) {
103110this.ctx.callbacks.onStatus?.("error", "Realtime connection closed");
@@ -119,6 +126,7 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
119126this.inputSource = null;
120127this.media?.getTracks().forEach((track) => track.stop());
121128this.media = null;
129+this.stopOutput();
122130void this.inputContext?.close();
123131this.inputContext = null;
124132void this.outputContext?.close();
@@ -128,7 +136,7 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
128136}
129137130138private startMicrophonePump(): void {
131-if (!this.media || !this.inputContext) {
139+if (this.closed || !this.media || !this.inputContext) {
132140return;
133141}
134142this.inputSource = this.inputContext.createMediaStreamSource(this.media);
@@ -152,24 +160,30 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
152160}
153161154162private send(message: unknown): void {
155-if (this.ws?.readyState === WebSocket.OPEN) {
163+if (!this.closed && this.ws?.readyState === WebSocket.OPEN) {
156164this.ws.send(JSON.stringify(message));
157165}
158166}
159167160-private handleMessage(data: unknown): void {
168+private async handleMessage(data: unknown): Promise<void> {
169+if (this.closed) {
170+return;
171+}
161172let message: GoogleLiveMessage;
162173try {
163-message = JSON.parse(String(data)) as GoogleLiveMessage;
174+message = JSON.parse(await decodeGoogleLiveMessageData(data)) as GoogleLiveMessage;
164175} catch {
165176return;
166177}
178+if (this.closed) {
179+return;
180+}
167181if (message.setupComplete) {
168182this.ctx.callbacks.onStatus?.("listening");
169183}
170184const content = message.serverContent;
171185if (content?.interrupted) {
172-this.playhead = this.outputContext?.currentTime ?? 0;
186+this.stopOutput();
173187}
174188if (content?.inputTranscription?.text) {
175189this.ctx.callbacks.onTranscript?.({
@@ -216,13 +230,25 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
216230);
217231buffer.getChannelData(0).set(samples);
218232const source = this.outputContext.createBufferSource();
233+this.sources.add(source);
234+source.addEventListener("ended", () => this.sources.delete(source));
219235source.buffer = buffer;
220236source.connect(this.outputContext.destination);
221237const startAt = Math.max(this.outputContext.currentTime, this.playhead);
222238source.start(startAt);
223239this.playhead = startAt + buffer.duration;
224240}
225241242+private stopOutput(): void {
243+for (const source of this.sources) {
244+try {
245+source.stop();
246+} catch {}
247+}
248+this.sources.clear();
249+this.playhead = this.outputContext?.currentTime ?? 0;
250+}
251+226252private async handleToolCall(call: {
227253id?: string;
228254name?: string;
@@ -238,13 +264,31 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
238264return;
239265}
240266await submitRealtimeTalkConsult({
241-ctx: this.ctx,
267+ctx: this.createActiveContext(),
242268 callId,
243269args: call.args ?? {},
244270submit: (toolCallId, result) => this.submitToolResult(toolCallId, result),
245271});
246272}
247273274+private createActiveContext(): RealtimeTalkTransportContext {
275+return {
276+ ...this.ctx,
277+callbacks: {
278+onStatus: (status, detail) => {
279+if (!this.closed) {
280+this.ctx.callbacks.onStatus?.(status, detail);
281+}
282+},
283+onTranscript: (entry) => {
284+if (!this.closed) {
285+this.ctx.callbacks.onTranscript?.(entry);
286+}
287+},
288+},
289+};
290+}
291+248292private submitToolResult(callId: string, result: unknown): void {
249293const pending = this.pendingCalls.get(callId);
250294if (!pending) {
@@ -268,3 +312,25 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
268312});
269313}
270314}
315+316+async function decodeGoogleLiveMessageData(data: unknown): Promise<string> {
317+if (typeof data === "string") {
318+return data;
319+}
320+if (typeof Blob !== "undefined" && data instanceof Blob) {
321+data = await data.arrayBuffer();
322+}
323+if (isArrayBufferLike(data)) {
324+return new TextDecoder().decode(new Uint8Array(data));
325+}
326+if (ArrayBuffer.isView(data)) {
327+return new TextDecoder().decode(new Uint8Array(data.buffer, data.byteOffset, data.byteLength));
328+}
329+return String(data);
330+}
331+332+function isArrayBufferLike(data: unknown): data is ArrayBuffer {
333+return (
334+data instanceof ArrayBuffer || Object.prototype.toString.call(data) === "[object ArrayBuffer]"
335+);
336+}
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。