From 55381c5162014d45810afce3729fb0d9a69d9d91 Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Wed, 5 Aug 2026 09:28:21 +0800 Subject: [PATCH 1/2] fix(realtime): unblock fixed speech after tool calls --- .../services/pipecat/qwen_audio_realtime.py | 36 +++++++--- backend/services/pipecat/realtime_tools.py | 15 ++++- backend/services/pipecat/stepfun_realtime.py | 66 +++++++++++++++---- 3 files changed, 97 insertions(+), 20 deletions(-) diff --git a/backend/services/pipecat/qwen_audio_realtime.py b/backend/services/pipecat/qwen_audio_realtime.py index 2dfa759..465cf81 100644 --- a/backend/services/pipecat/qwen_audio_realtime.py +++ b/backend/services/pipecat/qwen_audio_realtime.py @@ -109,6 +109,8 @@ class QwenAudioRealtimeService(AIService): self._websocket = None self._receive_task: asyncio.Task | None = None self._session_ready = asyncio.Event() + self._response_done = asyncio.Event() + self._response_done.set() self._pending_events: list[dict[str, Any]] = [] self._warned_input_sample_rate = False @@ -123,7 +125,10 @@ class QwenAudioRealtimeService(AIService): self._user_transcript_timestamp = "" self._deferred_assistant_messages: list[dict[str, Any]] = [] self._tools: list[RealtimeTool] = [] - self._tool_session = RealtimeToolSession(self._send_tool_event) + self._tool_session = RealtimeToolSession( + self._send_tool_event, + wait_for_response_boundary=self._response_done.wait, + ) self._fixed_speech_completion: asyncio.Future[None] | None = None self._suppress_response_transcript = False self._speech_started_handler: SpeechStartedHandler | None = None @@ -373,6 +378,7 @@ class QwenAudioRealtimeService(AIService): websocket = self._websocket self._websocket = None self._session_ready.clear() + self._response_done.set() self._pending_events.clear() self._response_active = False self._user_transcript_pending = False @@ -404,6 +410,9 @@ class QwenAudioRealtimeService(AIService): if self._websocket is websocket: self._websocket = None self._session_ready.clear() + self._response_done.set() + self._greeting_request_item_id = None + self._resolve_fixed_speech() if self._receive_task is asyncio.current_task(): self._receive_task = None @@ -424,6 +433,7 @@ class QwenAudioRealtimeService(AIService): await self._send_event(payload, wait_until_ready=False) elif event_type == "response.created": self._response_active = True + self._response_done.clear() self._audio_suppressed = False elif event_type == "response.audio.delta": audio = event.get("delta") @@ -473,13 +483,17 @@ class QwenAudioRealtimeService(AIService): await self._finish_assistant_text(interrupted=interrupted) await self._delete_greeting_request() self._resolve_fixed_speech() + self._response_done.set() elif event_type == "response.output_item.added": self._remember_function_call(event) elif event_type in { "response.function_call_arguments.done", "response.output_item.done", }: - await self._handle_function_call_event(event) + self.create_task( + self._handle_function_call_event(dict(event)), + name="qwen_audio_realtime_tool_call", + ) elif event_type == "error": error = event.get("error") message = error.get("message") if isinstance(error, dict) else str(error) @@ -527,12 +541,18 @@ class QwenAudioRealtimeService(AIService): ) if not name: return - await self._tool_session.handle_call( - name=name, - call_id=call_id, - arguments=source.get("arguments", event.get("arguments")), - ) - self._function_names.pop(call_id, None) + try: + await self._tool_session.handle_call( + name=name, + call_id=call_id, + arguments=source.get("arguments", event.get("arguments")), + ) + except asyncio.CancelledError: + raise + except Exception as exc: # noqa: BLE001 - background provider task + logger.exception(f"Qwen-Audio Realtime 工具调用处理失败:{exc}") + finally: + self._function_names.pop(call_id, None) def _remember_function_call(self, event: dict[str, Any]) -> None: item = event.get("item") diff --git a/backend/services/pipecat/realtime_tools.py b/backend/services/pipecat/realtime_tools.py index 1c1437c..dcbee19 100644 --- a/backend/services/pipecat/realtime_tools.py +++ b/backend/services/pipecat/realtime_tools.py @@ -46,13 +46,20 @@ RealtimeToolDispatcher = Callable[ [str, dict[str, Any], str], Awaitable[RealtimeToolResult] ] SendProviderEvent = Callable[[dict[str, Any]], Awaitable[None]] +WaitForResponseBoundary = Callable[[], Awaitable[Any]] class RealtimeToolSession: """Serialize provider calls and answer every call id at most once.""" - def __init__(self, send_event: SendProviderEvent) -> None: + def __init__( + self, + send_event: SendProviderEvent, + *, + wait_for_response_boundary: WaitForResponseBoundary | None = None, + ) -> None: self._send_event = send_event + self._wait_for_response_boundary = wait_for_response_boundary self._dispatcher: RealtimeToolDispatcher | None = None self._handled_call_ids: set[str] = set() self._lock = asyncio.Lock() @@ -105,6 +112,12 @@ class RealtimeToolSession: }, } ) + # Function-call argument events arrive before the provider's + # response.done boundary. Starting another response sooner is + # rejected by both supported realtime providers. The provider + # receive loop stays free while this background task waits. + if self._wait_for_response_boundary is not None: + await self._wait_for_response_boundary() if result.after_output is not None: await result.after_output() if result.continue_response: diff --git a/backend/services/pipecat/stepfun_realtime.py b/backend/services/pipecat/stepfun_realtime.py index 6b8819c..22791bb 100644 --- a/backend/services/pipecat/stepfun_realtime.py +++ b/backend/services/pipecat/stepfun_realtime.py @@ -72,13 +72,19 @@ class StepFunRealtimeService(AIService): self._websocket = None self._receive_task: asyncio.Task | None = None self._session_ready = asyncio.Event() + self._response_done = asyncio.Event() + self._response_done.set() self._pending_events: list[dict[str, Any]] = [] self._assistant_turn_id: str | None = None self._assistant_text = "" self._assistant_timestamp = "" self._tools: list[RealtimeTool] = [] - self._tool_session = RealtimeToolSession(self._send_tool_event) + self._tool_session = RealtimeToolSession( + self._send_tool_event, + wait_for_response_boundary=self._response_done.wait, + ) self._fixed_speech_completion: asyncio.Future[None] | None = None + self._fixed_speech_request_item_id: str | None = None self._suppress_response_transcript = False self._speech_started_handler: SpeechStartedHandler | None = None self._function_names: dict[str, str] = {} @@ -184,14 +190,25 @@ class StepFunRealtimeService(AIService): self._resolve_fixed_speech() self._fixed_speech_completion = completion self._suppress_response_transcript = suppress_transcript + item_id = f"item_{uuid4().hex}" + self._fixed_speech_request_item_id = item_id await self._send_event( { - "type": "response.create", - "session": { - "instructions": f"请原样无修改地输出下面的话：\n{text}", + "type": "conversation.item.create", + "item": { + "id": item_id, + "type": "message", + "role": "user", + "content": [ + { + "type": "input_text", + "text": f"请原样无修改地朗读下面的话：\n{text}", + } + ], }, } ) + await self._send_event({"type": "response.create"}) return completion async def _connect(self) -> None: @@ -225,8 +242,10 @@ class StepFunRealtimeService(AIService): websocket = self._websocket self._websocket = None self._session_ready.clear() + self._response_done.set() self._tool_session.clear() self._function_names.clear() + self._fixed_speech_request_item_id = None self._resolve_fixed_speech() if websocket and websocket.state is State.OPEN: try: @@ -259,6 +278,9 @@ class StepFunRealtimeService(AIService): if self._websocket is websocket: self._websocket = None self._session_ready.clear() + self._response_done.set() + self._fixed_speech_request_item_id = None + self._resolve_fixed_speech() if self._receive_task is asyncio.current_task(): self._receive_task = None @@ -271,6 +293,8 @@ class StepFunRealtimeService(AIService): pending, self._pending_events = self._pending_events, [] for payload in pending: await self._send_event(payload, wait_until_ready=False) + elif event_type == "response.created": + self._response_done.clear() elif event_type == "response.audio.delta": audio = event.get("delta") if audio: @@ -308,14 +332,19 @@ class StepFunRealtimeService(AIService): "interrupted", } await self._finish_assistant_text(interrupted=interrupted) + await self._delete_fixed_speech_request() self._resolve_fixed_speech() + self._response_done.set() elif event_type == "response.output_item.added": self._remember_function_call(event) elif event_type in { "response.function_call_arguments.done", "response.output_item.done", }: - await self._handle_function_call_event(event) + self.create_task( + self._handle_function_call_event(dict(event)), + name="stepfun_realtime_tool_call", + ) elif event_type == "error": error = event.get("error") message = error.get("message") if isinstance(error, dict) else str(error) @@ -407,12 +436,18 @@ class StepFunRealtimeService(AIService): ) if not name: return - await self._tool_session.handle_call( - name=name, - call_id=call_id, - arguments=source.get("arguments", event.get("arguments")), - ) - self._function_names.pop(call_id, None) + try: + await self._tool_session.handle_call( + name=name, + call_id=call_id, + arguments=source.get("arguments", event.get("arguments")), + ) + except asyncio.CancelledError: + raise + except Exception as exc: # noqa: BLE001 - background provider task + logger.exception(f"StepFun Realtime 工具调用处理失败:{exc}") + finally: + self._function_names.pop(call_id, None) def _remember_function_call(self, event: dict[str, Any]) -> None: item = event.get("item") @@ -430,6 +465,15 @@ class StepFunRealtimeService(AIService): if completion is not None and not completion.done(): completion.set_result(None) + async def _delete_fixed_speech_request(self) -> None: + item_id = self._fixed_speech_request_item_id + self._fixed_speech_request_item_id = None + if item_id: + await self._send_event( + {"type": "conversation.item.delete", "item_id": item_id}, + wait_until_ready=False, + ) + async def _send_event( self, payload: dict[str, Any], *, wait_until_ready: bool = True ) -> None: From 69736f3c5114f927d2e54cb7d6fd2d1c66510e03 Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Wed, 5 Aug 2026 09:28:40 +0800 Subject: [PATCH 2/2] fix(auth): align local API host with browser --- docker-compose.yaml | 2 -- frontend/src/lib/api.ts | 30 +++++++++++++++++++++++++++--- 2 files changed, 27 insertions(+), 5 deletions(-) diff --git a/docker-compose.yaml b/docker-compose.yaml index a941085..cbcf703 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -74,8 +74,6 @@ services: image: node:22-slim working_dir: /app command: sh -c "npm install && npm run dev" - environment: - NEXT_PUBLIC_API_BASE_URL: "http://localhost:8000" volumes: - ./frontend:/app - /app/node_modules diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 0e03707..64f1cb7 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -1,5 +1,5 @@ /** - * 后端 API 客户端。基址走 NEXT_PUBLIC_API_BASE,缺省指向本地后端 :8000。 + * 后端 API 客户端。显式配置优先；本地开发缺省沿用页面主机名并连接 :8000。 * * JSON 契约与后端 schemas.py 对齐(camelCase),所以返回体可直接喂给页面 state。 * 注意:api_key 读取时后端永远打码,写回打码占位符表示"不改 key"(写时哨兵)。 @@ -7,8 +7,32 @@ import { loginPathWithReturnTo } from "@/lib/auth-redirect"; -export const API_BASE = - process.env.NEXT_PUBLIC_API_BASE_URL ?? "http://localhost:8000"; +const LOOPBACK_HOSTS = new Set(["localhost", "127.0.0.1", "[::1]"]); + +function resolveApiBase(): string { + const configured = process.env.NEXT_PUBLIC_API_BASE_URL?.replace(/\/$/, ""); + if (typeof window === "undefined") { + return configured || "http://localhost:8000"; + } + if (!configured) { + return `${window.location.protocol}//${window.location.hostname}:8000`; + } + + // localhost and 127.0.0.1 reach the same machine but are different cookie + // sites. Keep an explicitly configured scheme/port while aligning equivalent + // loopback hostnames with the address actually used in the browser. + const configuredUrl = new URL(configured); + if ( + LOOPBACK_HOSTS.has(configuredUrl.hostname) && + LOOPBACK_HOSTS.has(window.location.hostname) + ) { + configuredUrl.hostname = window.location.hostname; + return configuredUrl.toString().replace(/\/$/, ""); + } + return configured; +} + +export const API_BASE = resolveApiBase(); export type ModelType = "LLM" | "ASR" | "TTS" | "Realtime" | "Embedding" | "Agent";