wx44wx/ai-video-fullstack
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

Merge branch 'main' of https://gitea.xiaowang.eu.org/wx44wx/ai-video-fullstack

Browse Source
This commit is contained in:
Xin Wang
2026-08-05 09:38:03 +08:00
parent 249dbe1fe3 6863567fa9
commit 3a4cb9522e
5 changed files with 124 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/services/pipecat/qwen_audio_realtime.py
+28 -8
View File
@@ -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")
backend/services/pipecat/realtime_tools.py
+14 -1
View File
@@ -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:
backend/services/pipecat/stepfun_realtime.py
+55 -11
View File
@@ -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:
docker-compose.yaml
-2
View File
@@ -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
frontend/src/lib/api.ts
+27 -3
View File
@@ -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";