WebRTC方式控制Go2机器人的异步调用异常问题排查

1. asyncio.run(connect())执行后连接自动关闭 #

asyncio.run()的设计逻辑是创建全新事件循环，运行传入协程后立即关闭循环。而WebRTC连接完全依赖事件循环持续运行，循环关闭后所有异步IO操作（信令、媒体流）终止，直接导致连接断开。

2. 添加nest_asyncio后机器人无响应 #

nest_asyncio的循环调度逻辑和原生asyncio存在差异，而你提取的WebRTC模块是基于原生asyncio开发的。嵌套循环会导致客户端的IO任务无法被正确调度，指令数据包根本没发送到机器人端，所以机器人无响应。

3. 改为async后仅fun2生效 #

主要有两个可能：

• 协程执行顺序问题：如果fun1和fun2是同步await调用，可能fun1指令未等机器人确认就执行fun2，导致fun1指令被覆盖；或者fun1内部异步操作未正确await，指令没发送完成。
• WebRTC客户端非协程安全：多协程直接操作同一客户端实例，会导致内部状态混乱（比如发送队列被覆盖、信令冲突），只有最后执行的fun2指令能正常发送。

4. 仅单协程/loop.create_task能正常执行多指令 #

LangChain Agent调用函数时，通常处于同步上下文或独立协程上下文，而你的WebRTC客户端绑定在特定事件循环上。直接在Agent上下文执行异步控制函数，要么没关联到正确循环，要么多协程调用导致客户端状态冲突。单协程内是串行操作无冲突，loop.create_task则是把任务提交到客户端绑定的循环，能被正确调度。

1. 维持全局事件循环长期运行 #

放弃asyncio.run()单次启动，改为创建并维持全局事件循环：

import asyncio

# 全局事件循环
loop = asyncio.get_event_loop()

async def main():
    await connect()  # 初始化WebRTC连接
    await asyncio.Future()  # 用空Future让循环持续运行

if __name__ == "__main__":
    loop.run_until_complete(main())

2. 封装协程安全的WebRTC客户端 #

把客户端封装成单例，用锁保证多协程调用时的状态安全：

import asyncio
from functools import wraps

class Go2WebRTCClient:
    _instance = None
    _lock = asyncio.Lock()

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    async def connect(self):
        # 原connect逻辑
        pass

    def coro_safe(func):
        @wraps(func)
        async def wrapper(self, *args, **kwargs):
            async with self._lock:
                return await func(self, *args, **kwargs)
        return wrapper

    @coro_safe
    async def fun1(self):
        # 原fun1控制逻辑
        pass

    @coro_safe
    async def fun2(self):
        # 原fun2控制逻辑
        pass

3. 为LangChain Agent提供同步调用入口 #

用asyncio.run_coroutine_threadsafe把异步函数提交到全局循环，Agent可直接同步调用：

# 全局客户端实例
client = Go2WebRTCClient()

# Agent调用的同步入口
def call_fun1():
    future = asyncio.run_coroutine_threadsafe(client.fun1(), loop)
    return future.result()  # 等待协程执行完成

def call_fun2():
    future = asyncio.run_coroutine_threadsafe(client.fun2(), loop)
    return future.result()

4. 确保指令执行的完整性 #

在控制函数内添加等待机器人确认的逻辑，避免指令被覆盖：

@coro_safe
async def fun1(self):
    await self.send_command("move_forward")
    await self.wait_for_ack()  # 新增等待机器人响应确认的逻辑

内容的提问来源于stack exchange，提问作者Jimmy Hanns