import 'dart:async'; import 'dart:convert'; import 'dart:math'; import 'dart:typed_data'; import 'package:get/get.dart'; import 'package:web_socket_channel/io.dart'; import '../../core/utils/logger.dart'; import '../models/appconfig.dart'; import '../models/assistant_directive.dart'; /// 对话状态(对应协议的 DialogStateChanged) enum BailianDialogState { idle, listening, thinking, responding } /// 上抛给 UI 的事件类型 enum BailianEventType { started, // 会话建立,拿到 dialog_id stateChanged, // 对话状态变化 speechStarted, // 用户开始说话(ASR 起点) speechContent, // 用户语音识别结果(流式) speechEnded, // 用户说完 requestAccepted, // 服务端接受了 RequestToRespond / RequestToSpeak respondingStarted, respondingContent, // 模型回复文本(流式) respondingEnded, error, closed, } class BailianEvent { const BailianEvent( this.type, { this.text = '', this.state, this.code, this.isFinal = false, this.fatal = false, this.finishReason = '', this.directives = const [], }); final BailianEventType type; final String text; final BailianDialogState? state; final String? code; /// 仅 respondingContent 有意义。实测两种取值决定了这一帧长什么样: /// - `stop`:正常回复,有 text,可能顺带 [directives](定闹钟就是这种); /// - `command_calls`:**纯指令帧,text 和 spoken 都是空字符串**, /// 而且这一轮**不会有 RespondingStarted / RespondingEnded**。 /// UI 照着 text 建气泡的话会留下一个永远填不上的空气泡。 final String finishReason; /// 本帧携带的端侧指令(`extra_info.tool_calls`)。 /// /// 注意别和 `extra_info.tool_infos` 混:那个是服务端插件**已经执行完**的 /// 结果(天气就是),端侧不需要做任何事。 final List directives; /// 对 speechContent / respondingContent 而言表示"这一句到此为止" final bool isFinal; /// 仅对 [BailianEventType.error] 有意义:会话是不是已经废了。 /// /// `task-failed`(header 级)= 整个 task 结束,必须重连; /// `Error`(payload.output 级)= 这一轮出了问题(比如音色不对、超时), /// 连接还在,提示一下就行,不该让整个页面显示"连接失败"。 final bool fatal; @override String toString() => 'BailianEvent($type, text="$text", state=$state, code=$code, final=$isFinal, ' 'fatal=$fatal, finish=$finishReason, directives=${directives.length})'; } /// 阿里云百炼「多模态交互协议」客户端。 /// /// 协议文档:https://help.aliyun.com/zh/model-studio/multimodal-interaction-protocol /// 端点:`wss://dashscope.aliyuncs.com/api-ws/v1/inference` /// /// EMAI 助手的整条链路都在这里:ASR、LLM、TTS 全在百炼服务端完成, /// 客户端只负责推上行音频/文本、收下行文本和音频。 /// (替掉了原先「Azure STT → 火山方舟 doubao LLM → 火山 TTS」的三段拼装。) /// /// 三个配置项,优先读服务端下发的 AppConfig,取不到时用本文件的默认值: /// - `ALIBABA_OPENSPEECH_APP_KEY`:DashScope API Key(与语音翻译、拍照翻译共用),**没有默认值** /// - `ALIBABA_BAILIAN_WORKSPACE_ID`:业务空间 ID,默认 [_defaultWorkspaceId] /// - `ALIBABA_BAILIAN_APP_ID`:多模态应用 ID,默认 [_defaultAppId] /// /// ## 一个应用一个实例 /// /// [appId] / [workspaceId] 传了就用传的,**不传(默认)完全走上面那套全局配置**—— /// 也就是说 EMAI 与设备唤醒会话的行为一个字节都没变。 /// /// 之所以做成实例级而不是改全局:不同 agent 用的是百炼里不同的应用 /// (EMAI 一个、Smartcar 车载助手另一个),改全局值会把所有 agent 一起换掉。 /// /// ⚠️ 每个应用**必须各自 new 一个实例**,不能共用:`start()` 第一件事就是 `stop()`, /// 共用会让两边互相把对方的会话掐掉,下行音频还会串到对方的播放器上。 class BailianMultimodalService extends GetxService { BailianMultimodalService({String? appId, String? workspaceId}) : _appIdOverride = appId, _workspaceIdOverride = workspaceId; /// 实例级覆盖;null/空 = 用全局配置 final String? _appIdOverride; final String? _workspaceIdOverride; static const String _tag = 'BailianMultimodal'; static BailianMultimodalService get to => Get.find(); static const String _endpoint = 'wss://dashscope.aliyuncs.com/api-ws/v1/inference'; /// EMAI 助手对应的百炼多模态应用 static const String _defaultAppId = '2813e297edb149b0956082641f3e7a53'; static const String _defaultWorkspaceId = 'ws-bkqdf9jl7eajs1qy'; /// 上行音频:16k / 16bit / 单声道 / 小端 static const int upstreamSampleRate = 16000; /// 下行音频默认采样率:手机外放走 24k,[PcmStreamPlayer] 按它拼 WAV 头。 /// 恒玄耳机那条链路会在 [start] 里指定 16000——G.722 编码器就是 16k, /// 直接让服务端出 16k 比本地重采样一遍干净。 static const int downstreamSampleRate = 24000; /// 本次会话实际协商的下行采样率([start] 的 downstreamRate 参数) int _downstreamRate = downstreamSampleRate; int get currentDownstreamRate => _downstreamRate; /// 服务端要求任意连续 60s 内必须有消息往来,否则判定异常;文档建议 50s 一次心跳 static const Duration _heartbeatInterval = Duration(seconds: 50); IOWebSocketChannel? _channel; StreamSubscription? _sub; Timer? _heartbeat; String _taskId = ''; String _dialogId = ''; final _events = StreamController.broadcast(); Stream get events => _events.stream; /// 服务端合成的下行音频(PCM,采样率见 [currentDownstreamRate]) final _audioOut = StreamController.broadcast(); Stream get audioOut => _audioOut.stream; final Rx state = BailianDialogState.idle.obs; final RxBool isConnected = false.obs; String get _apiKey => AppConfig.env('ALIBABA_OPENSPEECH_APP_KEY') ?? ''; String get _workspaceId { final o = _workspaceIdOverride?.trim() ?? ''; if (o.isNotEmpty) return o; final v = (AppConfig.env('ALIBABA_BAILIAN_WORKSPACE_ID') ?? '').trim(); return v.isEmpty ? _defaultWorkspaceId : v; } String get _appId { final o = _appIdOverride?.trim() ?? ''; if (o.isNotEmpty) return o; final v = (AppConfig.env('ALIBABA_BAILIAN_APP_ID') ?? '').trim(); return v.isEmpty ? _defaultAppId : v; } /// 只有 API Key 是硬要求:workspace / app id 有兜底默认值。 /// Key 缺失时不要尝试连接——服务端会先回 task-started 再立刻 task-failed, /// 日志上看起来像"连上了又掉",容易误判成网络问题。 bool get isConfigured => _apiKey.trim().isNotEmpty; String get configHint => _apiKey.trim().isEmpty ? '未配置 ALIBABA_OPENSPEECH_APP_KEY' : ''; /// 建立会话。 /// /// [mode] 决定上行怎么切句: /// - `push2talk`:按住说话,由客户端用 [beginSpeech]/[endSpeech] 划定一段 /// - `tap2talk`:点一下开始,服务端 VAD 判断说完 /// - `duplex`:全双工连续对话,服务端持续做 VAD /// /// [voice] 留空就用百炼应用自己配置的音色,一般不要传。 /// /// 纯文字对话也要先 [start]——RequestToRespond 是会话内指令,没有会话发不出去。 /// 本次会话解出来的指令记到哪个来源名下(`emai` / `earphone`)。 /// 服务是单例、两条链路轮流用,所以在 [start] 时定,不由调用方每次传。 String directiveSource = 'emai'; Future start({ String mode = 'push2talk', String? voice, String? userId, bool enableWebSearch = true, int downstreamRate = downstreamSampleRate, String source = 'emai', String? mcpToken, }) async { if (!isConfigured) { Logger.e(_tag, '配置不全,无法启动会话:$configHint'); _events.add(BailianEvent(BailianEventType.error, text: configHint, code: 'NOT_CONFIGURED', fatal: true)); return false; } await stop(); _downstreamRate = downstreamRate; directiveSource = source; _taskId = _uuid(); try { _channel = IOWebSocketChannel.connect( Uri.parse(_endpoint), headers: { 'Authorization': 'Bearer $_apiKey', 'X-DashScope-DataInspection': 'enable', }, pingInterval: const Duration(seconds: 20), ); _sub = _channel!.stream.listen( _onMessage, onError: (e) { Logger.e(_tag, 'WebSocket 错误: $e'); _events.add( BailianEvent(BailianEventType.error, text: '$e', fatal: true)); _cleanup(); }, onDone: () { Logger.i(_tag, 'WebSocket 关闭'); _events.add(const BailianEvent(BailianEventType.closed)); _cleanup(); }, ); _send({ 'header': { 'action': 'run-task', 'task_id': _taskId, 'streaming': 'duplex', }, 'payload': { 'task_group': 'aigc', 'task': 'multimodal-generation', 'function': 'generation', 'model': 'multimodal-dialog', 'input': { 'directive': 'Start', 'workspace_id': _workspaceId, 'app_id': _appId, }, 'parameters': { 'upstream': { 'type': 'AudioOnly', 'mode': mode, 'audio_format': 'pcm', 'sample_rate': upstreamSampleRate, }, 'downstream': { // 不传 voice:音色属于百炼应用的配置,客户端硬编码一个名字 // 只会跟应用实际用的 TTS 模型对不上(实测传 longxiaochun_v2 会被 // 回 "tts voice error , need cosyvoice-v3-flash voice.")。 // 需要临时覆盖时再从外面传。 if (voice != null && voice.trim().isNotEmpty) 'voice': voice, 'audio_format': 'pcm', 'sample_rate': _downstreamRate, // transcript:中间态只回朗读文本,跟 TTS 出声一致, // 免得 UI 上先闪一段 dialog 思考态文本再被覆盖 'intermediate_text': 'transcript', // 显式钉死"每帧回全文":EmaiController 收到 RespondingContent // 是整体替换气泡内容的,改成增量会导致只显示最后一个片段 'incremental_response': false, }, 'client_info': { 'user_id': _safeUserId(userId), }, // biz_params 是可选块,两件事都没有时整块不发—— // 万一应用侧没开这些能力,多带一个字段可能让 Start 直接被拒, // 那样整个功能都起不来,比少一个搜索严重得多。 if (enableWebSearch || _hasMcpToken(mcpToken)) 'biz_params': { 'user_defined_params': { // MCP 会话令牌:百炼把 user_defined_params 透传给 MCP 工具, // 我们的工具据此认出「这是哪个用户」。 // // ⚠️ **必须按工具名嵌套**,平铺写 `'auth_token': xxx` 会被静默丢弃。 // 文档只含糊说「mcp 传递参数依照 mcp 服务本身所需的参数」, // 实测(2026-09-07,四种形状逐个试)才确定是这个形状: // {"<工具名>": {"auth_token": "..."}} // 平铺 / 按 MCP 服务名嵌套 / 放 mcp 子节点,三种都收不到。 // // ⚠️ 这是整条链路上**唯一**能按用户变的通道:客户端连百炼只带 // 百炼的 API Key,百炼调 MCP 的 Authorization 头是控制台里 // 配死的、所有用户共用。不带它,这些工具一律「身份校验失败」。 // // ⚠️ 放的是 user_getmcptoken 签发的**短时效专用令牌**, // 绝不能换成登录 JWT —— 这条路要经过百炼(第三方)。 if (_hasMcpToken(mcpToken)) for (final tool in _mcpAuthTools) tool: {'auth_token': mcpToken}, if (enableWebSearch) 'extra_config': {'enable_web_search': true}, }, }, }, }, }); isConnected.value = true; _heartbeat?.cancel(); _heartbeat = Timer.periodic(_heartbeatInterval, (_) => _sendHeartbeat()); Logger.i( _tag, '会话启动: app_id=$_appId, workspace=$_workspaceId, mode=$mode, ' 'voice=${voice ?? '(应用默认)'}, downstream=${_downstreamRate}Hz'); return true; } catch (e) { Logger.e(_tag, '连接失败: $e'); _events .add(BailianEvent(BailianEventType.error, text: '$e', fatal: true)); _cleanup(); return false; } } /// 结束会话:先发 Stop 让服务端收尾,再 finish-task 结束整个 WS 任务 Future stop() async { if (_channel == null) return; try { _sendDirective('Stop'); _send({ 'header': { 'action': 'finish-task', 'task_id': _taskId, 'streaming': 'duplex', }, 'payload': {'input': {}}, }); } catch (_) {} await _sub?.cancel(); _sub = null; try { await _channel?.sink.close(); } catch (_) {} _cleanup(); } /// push2talk:开始一段说话 void beginSpeech() => _sendDirective('SendSpeech'); /// push2talk:结束这段说话,交给服务端处理 void endSpeech() => _sendDirective('StopSpeech'); /// push2talk:放弃这段说话(比如手指滑出按钮),服务端丢弃已收到的音频 void cancelSpeech() => _sendDirective('CancelSpeech'); /// 推上行音频。PCM 16k/16bit/单声道小端,建议每 ~100ms 一包 /// (字节数 = 采样率 × 2 × 间隔ms / 1000,16k/100ms = 3200 字节)。 void pushAudio(Uint8List pcm) { if (pcm.isEmpty) return; final ch = _channel; if (ch == null) return; try { ch.sink.add(pcm); } catch (e) { Logger.w(_tag, '推送音频失败: $e'); } } /// 文本对话:把文字交给应用的 LLM 处理(type=prompt), /// 或让它直接朗读一段文本(type=transcript)。 void sendText(String text, {bool speakDirectly = false}) { if (text.trim().isEmpty) return; _sendDirective('RequestToRespond', extra: { 'type': speakDirectly ? 'transcript' : 'prompt', 'text': text, }); } /// 打断当前回复并请求发言权 void interrupt() => _sendDirective('RequestToSpeak'); /// 本地开始播放下行音频(服务端据此推进对话状态) void notifyLocalRespondingStarted() => _sendDirective('LocalRespondingStarted'); /// 本地播放完成通知 void notifyLocalRespondingEnded() => _sendDirective('LocalRespondingEnded'); void _sendHeartbeat() { if (_channel == null) return; _sendDirective('HeartBeat'); } void _sendDirective(String directive, {Map? extra}) { if (_channel == null) return; final input = { 'directive': directive, 'workspace_id': _workspaceId, 'app_id': _appId, }; if (_dialogId.isNotEmpty) input['dialog_id'] = _dialogId; if (extra != null) input.addAll(extra); _send({ 'header': { 'action': 'continue-task', 'task_id': _taskId, 'streaming': 'duplex', }, 'payload': { 'task_group': 'aigc', 'task': 'multimodal-generation', 'function': 'generation', 'model': 'multimodal-dialog', 'input': input, }, }); } void _send(Map msg) { final ch = _channel; if (ch == null) return; try { ch.sink.add(json.encode(msg)); } catch (e) { Logger.w(_tag, '发送失败: $e'); } } void _onMessage(dynamic raw) { // 二进制帧就是下行合成音频,直接抛给播放侧 if (raw is List) { _audioOut.add(raw is Uint8List ? raw : Uint8List.fromList(raw)); return; } if (raw is! String) return; Map d; try { d = json.decode(raw) as Map; } catch (e) { Logger.w(_tag, '无法解析的文本帧: $e'); return; } final header = (d['header'] as Map?) ?? const {}; final payload = (d['payload'] as Map?) ?? const {}; final event = (header['event'] ?? '').toString(); switch (event) { case 'task-started': Logger.i(_tag, 'task-started'); break; case 'task-failed': final code = (header['error_code'] ?? '').toString(); final msg = (header['error_message'] ?? '').toString(); Logger.e(_tag, 'task-failed: $code $msg'); _events.add(BailianEvent(BailianEventType.error, text: msg, code: code, fatal: true)); break; case 'task-finished': Logger.i(_tag, 'task-finished'); _events.add(const BailianEvent(BailianEventType.closed)); break; default: _handleDirectiveEvent(event, payload); } } /// 业务事件都在 payload.output.event 里 void _handleDirectiveEvent(String outerEvent, Map payload) { final output = (payload['output'] as Map?) ?? const {}; final name = (output['event'] ?? outerEvent).toString(); final text = (output['text'] ?? '').toString(); final finished = output['finished'] == true; switch (name) { case 'Started': _dialogId = (output['dialog_id'] ?? '').toString(); Logger.i(_tag, 'Started dialog_id=$_dialogId'); _events.add(const BailianEvent(BailianEventType.started)); break; case 'DialogStateChanged': // 协议里是首字母大写(Listening/Thinking/Responding),这里统一小写再映射 final s = (output['state'] ?? '').toString().toLowerCase(); final mapped = switch (s) { 'listening' => BailianDialogState.listening, 'thinking' => BailianDialogState.thinking, 'responding' => BailianDialogState.responding, _ => BailianDialogState.idle, }; state.value = mapped; _events.add(BailianEvent(BailianEventType.stateChanged, state: mapped)); break; case 'SpeechStarted': _events.add(const BailianEvent(BailianEventType.speechStarted)); break; case 'SpeechContent': _events.add(BailianEvent(BailianEventType.speechContent, text: text, isFinal: finished)); break; case 'SpeechEnded': _events.add(BailianEvent(BailianEventType.speechEnded, text: text, isFinal: true)); break; case 'RequestAccepted': _events.add(const BailianEvent(BailianEventType.requestAccepted)); break; case 'RespondingStarted': _events.add(const BailianEvent(BailianEventType.respondingStarted)); break; case 'RespondingContent': final reason = (output['finish_reason'] ?? '').toString(); final parsed = AssistantDirective.parseFromOutput(output, source: directiveSource); // ⚠️ warn 级、且**只在这一帧真的带了 extra_info 或 command_calls 时**打。 // // 「闹钟没定上」这个问题只有两种可能——百炼压根没下发指令,或者下发了 // 但端侧没落地——而 release 包的级别门槛是 warning,用 info/debug 记等于 // 真机上什么都查不到,只能靠猜。这一行就是把两种可能分开的那个证据: // 有 tool_calls 说明模型发了,接下来看 AssistantDirectiveService 的日志; // 一条都没有说明是百炼后台的工具没配上,改客户端没有意义。 // // 频率可控:extra_info 只在整轮收尾那一帧出现,不是每个流式分片都有。 final extra = output['extra_info']; if ((extra is Map && extra.isNotEmpty) || reason == 'command_calls') { Logger.w( _tag, '[指令帧] finish_reason=$reason 解析出=${parsed.length} 条 ' 'extra_info=${_brief(extra)}'); } _events.add(BailianEvent( BailianEventType.respondingContent, text: text, isFinal: finished, finishReason: reason, directives: parsed, )); break; case 'RespondingEnded': _events.add(BailianEvent(BailianEventType.respondingEnded, text: text, isFinal: true)); break; case 'Error': final code = (output['error_code'] ?? '').toString(); final msg = (output['error_message'] ?? '').toString(); final name0 = (output['error_name'] ?? '').toString(); Logger.e(_tag, 'Error: $code/$name0 $msg'); // payload 级 Error 是"这一轮不行",连接通常还在,不标 fatal _events.add(BailianEvent(BailianEventType.error, text: msg.isNotEmpty ? msg : (text.isNotEmpty ? text : name0), code: code)); break; case 'Stopped': _events.add(const BailianEvent(BailianEventType.closed)); break; case 'HeartBeat': break; default: Logger.d(_tag, '未处理事件: $name payload=$payload'); } } /// 日志用:把任意结构截成一行,超长截断。 /// Logger 单行上限 4000,这里再收紧一道,免得一帧把屏幕刷满。 static String _brief(Object? v, {int max = 800}) { if (v == null) return 'null'; final s = v.toString(); return s.length <= max ? s : '${s.substring(0, max)}…(共${s.length}字符)'; } void _cleanup() { _heartbeat?.cancel(); _heartbeat = null; _channel = null; _dialogId = ''; isConnected.value = false; state.value = BailianDialogState.idle; } /// 协议限制 user_id 不超过 36 字符 /// 需要用户身份的 MCP 工具名。令牌要**按工具名逐个挂**,因为百炼是按这个 /// key 去匹配的(见 start() 里 user_defined_params 的说明)。 /// /// ⚠️ 服务端新增需要身份的工具时,这里要同步加一行,否则那个工具永远 /// 拿不到令牌、永远返回「身份校验失败」,而且客户端不会有任何报错。 /// 服务端工具清单见 apps/services/modules/mcp/tool_*.go。 static const List _mcpAuthTools = [ 'search_meeting_notes', 'get_memory_items', 'get_memory_stats', 'get_memory_report', 'get_user_tasks', 'add_user_task', 'cancel_user_task', ]; static bool _hasMcpToken(String? v) => (v ?? '').trim().isNotEmpty; static String _safeUserId(String? raw) { final v = (raw ?? '').trim(); if (v.isEmpty) return 'eaimar-guest'; return v.length <= 36 ? v : v.substring(0, 36); } static String _uuid() { final r = Random.secure(); String hex(int n) => List.generate(n, (_) => r.nextInt(16).toRadixString(16)).join(); return '${hex(8)}-${hex(4)}-4${hex(3)}-a${hex(3)}-${hex(12)}'; } @override void onClose() { stop(); _events.close(); _audioOut.close(); super.onClose(); } }