diff --git a/lib/modules/agent/controllers/agent_controller.dart b/lib/modules/agent/controllers/agent_controller.dart index b2a387879..5507bcb57 100644 --- a/lib/modules/agent/controllers/agent_controller.dart +++ b/lib/modules/agent/controllers/agent_controller.dart @@ -456,7 +456,7 @@ class AgentController extends GetxController { final start = metaResult["card_navigation"]["start"] as String; final end = metaResult["card_navigation"]["end"] as String; // Logger.i(TAG, 'liwei-------- 打开导航页面: $start $end '); - Future.delayed(Duration(milliseconds: 1500), () { + Future.delayed(Duration(seconds: 2), () { // Logger.i(TAG, 'liwei-------- 打开导航页面: $start $end '); // 2秒后执行的代码 openNavigation(start, end, ""); diff --git a/local_plugins/agent_service/android/src/main/kotlin/com/yunqiinnovation/agent_service/AgentService.kt b/local_plugins/agent_service/android/src/main/kotlin/com/yunqiinnovation/agent_service/AgentService.kt index f73d7c0b9..20b97a7e1 100644 --- a/local_plugins/agent_service/android/src/main/kotlin/com/yunqiinnovation/agent_service/AgentService.kt +++ b/local_plugins/agent_service/android/src/main/kotlin/com/yunqiinnovation/agent_service/AgentService.kt @@ -662,7 +662,7 @@ object AgentService : CoroutineScope { val responseBuilder = StringBuilder() var aiMetadata:String = "" var userMetadata = JSONObject() - + var nobroadcast = false // 如果有图片，添加图片信息到元数据 if (hasImage) { userMetadata.put("hasImage", true) @@ -717,7 +717,7 @@ object AgentService : CoroutineScope { callback = object : OpenAIService.StreamCallback { override fun onToken(token: String) { responseBuilder.append(token) - if (speakResponse) { + if (speakResponse && !nobroadcast) { ttsService?.speakStream(token) } // 发送流式回复token @@ -748,8 +748,6 @@ object AgentService : CoroutineScope { // 保存聊天记录 saveChatMessage(displayText, response,aiMetadata,userMetadata.toString()) - - } // 标记AI流式输出已完成 @@ -792,7 +790,9 @@ object AgentService : CoroutineScope { "result" to functionCallResult.toString(), )) Log.d(TAG, "mcp调用结果: $functionCallResult") - aiMetadata = autoHandleFcunCallResult(functionCallResult); + val (metestr, broadcast)= autoHandleFcunCallResult(functionCallResult); + aiMetadata = metestr; + nobroadcast = broadcast } } ) @@ -1108,8 +1108,9 @@ object AgentService : CoroutineScope { /* * 自动播放音乐 * */ - fun autoHandleFcunCallResult(functionCallResult: JSONObject): String{ + fun autoHandleFcunCallResult(functionCallResult: JSONObject): Pair{ val metaStr = functionCallResult.optString("meta") + var broadcast = false; if (metaStr.isNotEmpty()) { val meta = JSONObject(metaStr) // Log.d(TAG, "mcp调用结果: $metaStr") @@ -1122,6 +1123,7 @@ object AgentService : CoroutineScope { val sgener = cardMusic.optString("sgener", "") val image = cardMusic.optString("image", "") processMusicPlay(mapOf("id" to id,"url" to url, "title" to name, "artist" to sgener,"coverUrl" to image)) + broadcast = true; }else if (meta.has("card_musiclist")) { //音乐列表 val cardMusiclist = meta.getJSONObject("card_musiclist") val musics = cardMusiclist.optJSONArray("musics") @@ -1138,6 +1140,7 @@ object AgentService : CoroutineScope { } if (playlist.isNotEmpty()) { Log.w(TAG, "自动播放音乐列表 ${playlist}") + broadcast = true; processMusicPlayList(playlist) } else { Log.w(TAG, "card_musiclist 中没有有效的音乐条目") @@ -1149,11 +1152,12 @@ object AgentService : CoroutineScope { val cardNavigation = meta.getJSONObject("card_navigation") val start = cardNavigation.optString("start", "") val end = cardNavigation.optString("end", "") + broadcast = true; processNavigation(start,end); } - return metaStr; + return Pair(metaStr, broadcast); } - return ""; + return Pair("", false); } /** diff --git a/local_plugins/open_ai_service/android/src/main/kotlin/com/yunqiinnovation/open_ai_service/MCPClient.kt b/local_plugins/open_ai_service/android/src/main/kotlin/com/yunqiinnovation/open_ai_service/MCPClient.kt index c94ebb3d4..b186b6a64 100644 --- a/local_plugins/open_ai_service/android/src/main/kotlin/com/yunqiinnovation/open_ai_service/MCPClient.kt +++ b/local_plugins/open_ai_service/android/src/main/kotlin/com/yunqiinnovation/open_ai_service/MCPClient.kt @@ -33,6 +33,14 @@ import okhttp3.Interceptor import io.ktor.client.request.* import io.ktor.http.* import com.yunqiinnovation.open_ai_service.mcp.CustomSseClientTransport +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job +import kotlinx.coroutines.delay +import kotlinx.coroutines.isActive +import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock import okio.Buffer /** @@ -68,11 +76,17 @@ class MCPSubClient( // 保存服务器URL private val serverUrl: String = serverUrl - - init { - Log.d(TAG, "MCPSubClient初始化，URL: $serverUrl") - } - + + // 重连配置 + private val maxRetryAttempts = 3 + private val initialReconnectDelay = 1000L // 初始重连延迟1秒 + private val maxReconnectDelay = 30000L // 最大重连延迟30秒 + private var currentReconnectDelay = initialReconnectDelay + private var retryCount = 0 + + // 连接状态锁 + private val connectionLock = Mutex() + // 初始化MCP客户端 private val mcp: Client = Client( clientInfo = Implementation( @@ -95,7 +109,18 @@ class MCPSubClient( // 工具名称到工具的映射，用于快速查找 private val toolMapping = mutableMapOf() - + + // 心跳检测 + private var heartbeatJob: Job? = null + private val heartbeatInterval = 30000L // 30秒 + + // 网络状态监听 + private var networkStateListener: NetworkStateListener? = null + + init { + Log.d(TAG, "MCPSubClient初始化，URL: $serverUrl") + } + /** * 将JsonObject转换为JsonValue */ @@ -110,6 +135,10 @@ class MCPSubClient( * @return 是否连接成功 */ suspend fun connect(): Boolean { + if (isConnected) { + Log.d(TAG, "已经连接到服务器，跳过重复连接") + return true + } try { Log.d(TAG, "开始连接服务器: $serverUrl") @@ -130,6 +159,11 @@ class MCPSubClient( processToolsList(toolsResult) isConnected = true + retryCount = 0 + currentReconnectDelay = initialReconnectDelay + // 启动心跳检测 + startHeartbeat() + Log.d(TAG, "连接成功，可用工具: ${tools.size}个") return true } catch (e: Exception) { @@ -368,9 +402,11 @@ class MCPSubClient( * 调用工具 */ suspend fun callTool(name: String, arguments: Map): JSONObject? { - if (!isConnected) { - Log.e(TAG, "未连接到服务器") - return JSONObject().apply { put("context","未连接到MCP服务器") } + + // 先检查连接状态 + if (!checkConnection()) { + Log.e(TAG, "无法连接到服务器，工具调用失败") + return JSONObject().apply { put("context", "无法连接到MCP服务器") } } if (!containsTool(name)) { @@ -395,32 +431,123 @@ class MCPSubClient( } } catch (e: Exception) { Log.e(TAG, "调用工具失败: ${e.message}") + isConnected = false JSONObject().apply { put("context", "调用工具失败: ${e.message}") } } } - + + /** + * 设置网络状态监听器 + */ + fun setNetworkStateListener(listener: NetworkStateListener) { + this.networkStateListener = listener + } + + /** + * 检查连接状态并自动重连 + */ + suspend fun checkConnection(): Boolean { + if (!isConnected) { + Log.d(TAG, "当前未连接，尝试重新连接...") + return connect() + } + + try { + // 简单ping测试连接状态 + mcp.ping() + return true + } catch (e: Exception) { + Log.e(TAG, "连接检查失败: ${e.message}") + isConnected = false + return false + } + } /** * 检查是否已连接 */ fun isConnected(): Boolean { return isConnected } - + /** + * 处理连接错误并尝试重连 + */ + private suspend fun handleConnectionError(e: Exception) { + isConnected = false + stopHeartbeat() + + if (retryCount < maxRetryAttempts) { + retryCount++ + currentReconnectDelay = minOf(currentReconnectDelay * 2, maxReconnectDelay) + + Log.w(TAG, "连接失败，将在 ${currentReconnectDelay}ms 后尝试重连 (尝试 $retryCount/$maxRetryAttempts)") + + delay(currentReconnectDelay) + connect() + } else { + Log.e(TAG, "已达到最大重试次数($maxRetryAttempts)，停止重连") + // 可以在这里添加通知或回调，告知上层连接彻底失败 + } + } + + /** + * 启动心跳检测 + */ + private fun startHeartbeat() { + heartbeatJob?.cancel() + heartbeatJob = CoroutineScope(Dispatchers.IO).launch { + while (isActive && isConnected) { + try { + delay(heartbeatInterval) + + // 发送心跳请求 + val startTime = System.currentTimeMillis() + val response = mcp.ping() + val latency = System.currentTimeMillis() - startTime + + Log.d(TAG, "心跳检测成功，延迟: ${latency}ms") + } catch (e: Exception) { + Log.e(TAG, "心跳检测失败: ${e.message}") + handleConnectionError(e) + break + } + } + } + } + + /** + * 停止心跳检测 + */ + private fun stopHeartbeat() { + heartbeatJob?.cancel() + heartbeatJob = null + } + /** * 关闭连接 */ override fun close() { runBlocking { - try { - mcp.close() - sseTransport?.close() - isConnected = false - Log.d(TAG, "已关闭MCP服务器连接") - } catch (e: Exception) { - Log.e(TAG, "关闭MCP服务器连接失败: ${e.message}", e) + connectionLock.withLock { + try { + isConnected = false + stopHeartbeat() + mcp.close() + sseTransport?.close() + Log.d(TAG, "已关闭MCP服务器连接") + } catch (e: Exception) { + Log.e(TAG, "关闭MCP服务器连接失败: ${e.message}", e) + } } } } + + /** + * 网络状态监听接口 + */ + interface NetworkStateListener { + fun onNetworkAvailable() + fun onNetworkLost() + } } /** diff --git a/local_plugins/open_ai_service/android/src/main/kotlin/com/yunqiinnovation/open_ai_service/mcp/CustomSseClientTransport.kt b/local_plugins/open_ai_service/android/src/main/kotlin/com/yunqiinnovation/open_ai_service/mcp/CustomSseClientTransport.kt index 6d6a5eb27..b198298a2 100644 --- a/local_plugins/open_ai_service/android/src/main/kotlin/com/yunqiinnovation/open_ai_service/mcp/CustomSseClientTransport.kt +++ b/local_plugins/open_ai_service/android/src/main/kotlin/com/yunqiinnovation/open_ai_service/mcp/CustomSseClientTransport.kt @@ -19,41 +19,8 @@ import kotlin.properties.Delegates import kotlin.time.Duration import java.util.concurrent.atomic.AtomicBoolean import org.json.JSONObject -import kotlinx.serialization.Serializable -import kotlinx.serialization.json.JsonElement -import kotlinx.serialization.json.JsonPrimitive -//@Serializable -//data class JSONRPCRequest( -// val jsonrpc: String = "2.0", -// val method: String, -// val params: JsonElement? = null, -// val id: JsonPrimitive? = null -//) : JSONRPCMessage, Parcelable { -// -// constructor(parcel: Parcel) : this( -// parcel.readString() ?: "2.0", -// parcel.readString() ?: "", -// parcel.readString()?.let { Json.decodeFromString(it) }, -// parcel.readString()?.let { Json.decodeFromString(it) } -// ) -// -// override fun writeToParcel(parcel: Parcel, flags: Int) { -// parcel.writeString(jsonrpc) -// parcel.writeString(method) -// parcel.writeString(params?.let { Json.encodeToString(it) }) -// parcel.writeString(id?.let { Json.encodeToString(it) }) -// } -// -// override fun describeContents(): Int = 0 -// -// companion object CREATOR : Parcelable.Creator { -// override fun createFromParcel(parcel: Parcel): JSONRPCRequest = JSONRPCRequest(parcel) -// override fun newArray(size: Int): Array = arrayOfNulls(size) -// } -//} - /** * 自定义SSE客户端传输层，修复原始SseClientTransport中的URL拼接问题 * 解决URL查询参数与路径拼接错误的问题，确保消息端点URL格式正确 @@ -77,11 +44,6 @@ class CustomSseClientTransport( private var job: Job? = null - - private val heartbeatIntervalMs = 30_000L // 30秒 - private var heartbeatJob: Job? = null - - // 创建JSON解析器，增强灵活性设置 private val json = Json { ignoreUnknownKeys = true // 忽略未知字段 @@ -341,9 +303,6 @@ class CustomSseClientTransport( // 收集SSE事件 collectEvents() - // 启动心跳任务 - startHeartbeat() - // 等待endpoint就绪 endpoint.await() Log.d(TAG, "传输层启动完成，消息端点已就绪") @@ -391,7 +350,6 @@ class CustomSseClientTransport( Log.e(TAG, "关闭失败: 传输层未初始化") error("CustomSseClientTransport is not initialized!") } - heartbeatJob?.cancel() session.cancel() _onClose() job?.cancelAndJoin() @@ -404,29 +362,6 @@ class CustomSseClientTransport( fun isInitialized(): Boolean { return initialized.get() } - - - private fun startHeartbeat() { - heartbeatJob = scope.launch { - while (isActive) { - delay(heartbeatIntervalMs) - try { - sendHeartbeat() - } catch (e: Exception) { - Log.e(TAG, "心跳发送失败: ${e.message}", e) - _onError(e) - } - } - } - } - - private suspend fun sendHeartbeat() { - val pingMessage = JSONRPCRequest( - method = "ping", - jsonrpc = "2.0" - ) - send(pingMessage) - } } \ No newline at end of file