8. 交互二开使用指南

交互二开现提供两类开发模式：

LinkSoul二次开发
- 智元交互能力保留，由智元提供交互链路上的各个接口，包括 tts、llm、asr 等。用户可根据需求选择调用不同的接口，实现复杂的交互二开功能。
二次开发完全接管
- 关闭智元交互能力，由智元提供包含本体降噪后的机器人麦克风输入，完全由二次开发程序接管语音交互相关内容。

建议在无特殊需求（如：内网环境，特殊交互需求，本地化部署）时，优先选择 LinkSoul二次开发模式。可有效减少开发成本，提高开发效率。

8.1 LinkSoul二次开发

可在灵心平台文档中心获取有关内容。如遇 LinkSoul 二次开发有关问题，可将问题提至工单。渠道如下： https://agirobot.feishu.cn/share/base/form/shrcn89PyePk9S7HvJKOmvYqUUb

使用流程

第一步：登录开发者控制台，创建应用并开通 AgentSDK 能力。

⚠️ 特别注意：创建完成后请立即保存 APP Secret，该信息仅首次创建时可查看，后续无法再次获取。

第二步：创建 SDK 模板

登录灵心平台，在“智能体中心 -> 智能体模板”中选择 SDK 接入创建，填写模板信息，并选择对应的应用（API Key）。

第三步：创建 SDK 智能体

基于 SDK 模板创建智能体，并根据业务需求完成配置。支持配置内容包括：

字段	说明
智能体名称	-
智能体介绍	-
机器人姓名	定义机器人姓名，可作为默认参数输入
设备唤醒词	定义机器人唤醒词，可作为默认参数输入
热词管理	支持强制将特定词或发音识别并映射为指定目标词，用于优化部分关键词的 ASR 识别结果。
设备使用城市	定义设备使用城市，可作为默认参数传入
输入格式	选择输入格式，可选原始音频或 ASR 识别结果
图像输入	选择是否开启交互视频流传输
音色	关闭后，设备将直接播放 SDK 返回的音频内容；打开并选择音色后，设备将调用当前配置的音色对 SDK 返回的文本内容进行播报。
技能调用方式	系统自动回应：技能触发后，由智元内置能力直接处理本次请求，并将相关上下文传递给 Agent，用于后续对话理解与响应。 Agent 调用：由开发者自定义技能触发后处理逻辑，可组合已有表情、动作及返回内容进行编排和返回。
用户信息识别	系统自动回应：根据已配置的用户信息和打招呼方式，平台直接处理并回应。识别信息传入 Agent：将识别到的用户信息通过 SDK 传入自定义 Agent，由 Agent 进行后续处理和回复。
知识库	选择知识库后，可在自定义 Agent 代码中调用知识库检索 API，通过智能体 code 可检索范围将限定为当前智能体所选知识库。API 调用文档

SDK 连接及调试

开发者可参考 SDK 文档完成服务开发，并建立与机器人之间的连接。

第四步：SDK 连通性及调试

SDK 建连成功后，可在智能体编辑页面点击“测试连通性”，验证连接状态。

随后即可进入“对话调试”页面，通过真实语音交互调试 SDK 输入输出效果，快速验证业务逻辑。

8.2 二次开发完全接管

交互二开涉及关闭智元交互能力、麦克风音频获取、扬声器使用、三个部分。以下将从这三个部分介绍如何进行交互二开。

8.2.1 关闭智元交互能力

在此模式下，需要让智元的交互程序退出交互链路，仅输出包含本体降噪后的麦克风音频、人脸识别结果并让出扬声器的控制权，需要进行的操作有两个：

调整 agent 模块为 only_voice / voice_face 模式，可以使用麦克风管理部分 SetAgentPropertiesRequest 接口。

only_voice：仅输出降噪麦克音频 /agent/process_audio_output，后续链路全部断开
voice_face：输出降噪麦克音频 /agent/process_audio_output 和人脸识别结果 /agent/vision/face_id，后续链路全部断开
例如调整为 only_voice 模式所需调用的 rpc 如下：
bash
```
curl -i \
    -H 'content-type:application/json' \
    -X POST 'http://10.42.10.10:59301/rpc/aimdk.protocol.AgentControlService/SetAgentPropertiesRequest' \
    -d '{ "contents": { "properties": { "2": "only_voice" } } }'
```
调用后需要重启机器人才能生效，可以待第二步的修改完成后再一块重启即可。重启后可以通过使用麦克风管理部分 GetAgentPropertiesRequest 接口查询当前交互运行模式，判断模式是否修改成功。（如需恢复，则将SetAgentPropertiesRequest 接口中的 "only_voice / voice_face" 改为 "normal" 即可）

8.2.2 麦克风音频获取

注意：要获取以下音频需要机器人开机时至少联网 2 分钟以上完成音频相关鉴权操作，否则将无原始音频输出，如需离线使用，请首先保证该接口有音频输出后再断网

退出交互链路后，可通过降噪麦克音频 Topic 接口 /agent/process_audio_output 获得本体降噪后的麦克风音频信息。获取本体降噪后的麦克风音频示例程序如下：

python

#!/usr/bin/env python3

import rclpy
from rclpy.node import Node
from rclpy.qos import QoSHistoryPolicy, QoSProfile, QoSReliabilityPolicy
from ros2_plugin_proto.msg import RosMsgWrapper

from aimdk.protocol_pb2 import ProcessedAudioOutput, AudioVADState

import datetime
import os

class AudioSubscriber(Node):
    def __init__(self):
        super().__init__("audio_subscriber")

        # 音频缓冲区，按 stream_id 分别存储
        self.audio_buffers = {}  # {stream_id: bytearray()}
        self.recording_state = {}  # {stream_id: bool} 记录是否正在录音

        # 创建音频文件存储目录
        self.audio_output_dir = "audio_recordings"
        os.makedirs(self.audio_output_dir, exist_ok=True)

        qos_profile = QoSProfile(
            history=QoSHistoryPolicy.KEEP_LAST,
            depth=10,
            reliability=QoSReliabilityPolicy.BEST_EFFORT,
        )

        self.subscription = self.create_subscription(
            RosMsgWrapper,
            "/agent/process_audio_output/pb_3Aaimdk_2Eprotocol_2EProcessedAudioOutput",
            self.audio_callback,
            qos_profile,
        )

        self.get_logger().info("开始订阅降噪音频数据。..")

    def audio_callback(self, msg):
        try:
            # 检查序列化类型是否为 pb
            if msg.serialization_type != "pb":
                self.get_logger().warn(f"不支持的序列化类型：{msg.serialization_type}")
                return

            # 将 data 字段从 list[bytes] 转换为 bytes
            audio_data_bytes = b"".join(msg.data)

            # 使用生成的 protobuf 类解析消息
            processed_audio = ProcessedAudioOutput()
            processed_audio.ParseFromString(audio_data_bytes)

            self.get_logger().info(
                f"收到音频数据：stream_id={processed_audio.stream_id}, "
                f"vad_state={processed_audio.vad_state}, "
                f"audio_size={len(processed_audio.audio_data)} bytes"
            )

            # 根据 VAD 状态处理音频
            self.handle_vad_state(processed_audio)

        except Exception as e:
            self.get_logger().error(f"处理音频消息时出错：{e}")

    def handle_vad_state(self, processed_audio):
        """处理不同的 VAD 状态"""
        vad_state = processed_audio.vad_state
        stream_id = processed_audio.stream_id
        audio_data = processed_audio.audio_data

        # 初始化该 stream_id 的缓冲区（如果不存在）
        if stream_id not in self.audio_buffers:
            self.audio_buffers[stream_id] = bytearray()
            self.recording_state[stream_id] = False

        # VAD 状态名称映射
        vad_state_names = {
            AudioVADState.AUDIO_VAD_STATE_NONE: "无语音",
            AudioVADState.AUDIO_VAD_STATE_BEGIN: "语音开始",
            AudioVADState.AUDIO_VAD_STATE_PROCESSING: "语音处理中",
            AudioVADState.AUDIO_VAD_STATE_END: "语音结束",
        }

        stream_names = {1: "内置麦克风", 2: "外置麦克风"}

        self.get_logger().info(
            f"[{stream_names.get(stream_id, f'未知流{stream_id}')}] "
            f"VAD 状态：{vad_state_names.get(vad_state, f'未知状态{vad_state}')} "
            f"音频数据：{len(audio_data)} bytes"
        )

        # 根据 VAD 状态处理音频数据
        if vad_state == AudioVADState.AUDIO_VAD_STATE_BEGIN:
            self.get_logger().info("🎤 检测到语音开始")
            # 开始新的录音，清空缓冲区
            self.audio_buffers[stream_id].clear()
            self.recording_state[stream_id] = True
            # 添加当前音频数据
            if len(audio_data) > 0:
                self.audio_buffers[stream_id].extend(audio_data)

        elif vad_state == AudioVADState.AUDIO_VAD_STATE_PROCESSING:
            self.get_logger().info("🔄 语音处理中。..")
            # 如果正在录音，继续添加音频数据到缓冲区
            if self.recording_state[stream_id] and len(audio_data) > 0:
                self.audio_buffers[stream_id].extend(audio_data)

        elif vad_state == AudioVADState.AUDIO_VAD_STATE_END:
            self.get_logger().info("✅ 语音结束")
            # 添加最后的音频数据
            if self.recording_state[stream_id] and len(audio_data) > 0:
                self.audio_buffers[stream_id].extend(audio_data)

            # 保存完整的音频段
            if (
                self.recording_state[stream_id]
                and len(self.audio_buffers[stream_id]) > 0
            ):
                self.save_audio_segment(bytes(self.audio_buffers[stream_id]), stream_id)

            # 结束录音
            self.recording_state[stream_id] = False

        elif vad_state == AudioVADState.AUDIO_VAD_STATE_NONE:
            # 无语音状态，不进行录音
            if self.recording_state[stream_id]:
                self.get_logger().info("⏹️ 录音状态重置")
                self.recording_state[stream_id] = False

        # 输出当前缓冲区状态
        if stream_id in self.audio_buffers:
            buffer_size = len(self.audio_buffers[stream_id])
            recording = self.recording_state[stream_id]
            self.get_logger().debug(
                f"[Stream {stream_id}] 缓冲区大小：{buffer_size} bytes, 录音状态：{recording}"
            )

    def save_audio_segment(self, audio_data, stream_id):
        """保存音频段 16kHz, 16 位，单声道 PCM"""
        if len(audio_data) > 0:
            timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S_%f")

            # 按 stream_id 创建子目录
            stream_dir = os.path.join(self.audio_output_dir, f"stream_{stream_id}")
            os.makedirs(stream_dir, exist_ok=True)

            # 生成文件名
            stream_names = {1: "internal_mic", 2: "external_mic"}
            stream_name = stream_names.get(stream_id, f"stream_{stream_id}")
            filename = f"{stream_name}_{timestamp}.pcm"
            filepath = os.path.join(stream_dir, filename)

            try:
                with open(filepath, "wb") as f:
                    f.write(audio_data)
                self.get_logger().info(
                    f"音频段已保存：{filepath} （大小：{len(audio_data)} bytes)"
                )

                # 记录音频文件的时长（假设 16kHz, 16 位，单声道）
                sample_rate = 16000
                bits_per_sample = 16
                channels = 1
                bytes_per_sample = bits_per_sample // 8
                total_samples = len(audio_data) // (bytes_per_sample * channels)
                duration_seconds = total_samples / sample_rate

                self.get_logger().info(
                    f"音频时长：{duration_seconds:.2f} 秒 ({total_samples} 样本）"
                )

            except Exception as e:
                self.get_logger().error(f"保存音频文件失败：{e}")

    def get_buffer_info(self):
        """获取所有缓冲区的信息（用于调试）"""
        info = {}
        for stream_id in self.audio_buffers:
            info[stream_id] = {
                "buffer_size": len(self.audio_buffers[stream_id]),
                "recording": self.recording_state[stream_id],
            }
        return info

def main(args=None):
    rclpy.init(args=args)

    audio_subscriber = AudioSubscriber()

    try:
        audio_subscriber.get_logger().info("正在监听降噪音频数据，按 Ctrl+C 退出。..")
        rclpy.spin(audio_subscriber)
    except KeyboardInterrupt:
        audio_subscriber.get_logger().info("收到退出信号，正在关闭。..")
    finally:
        audio_subscriber.destroy_node()
        rclpy.shutdown()

if __name__ == "__main__":
    main()

以上程序依赖 python 协议包 a3_aimdk 和 ros2 协议包 ros2_plugin_proto，相应的包已经放置在 AimDK 开发包的 prebuilt 目录下，python 包使用 pip install prebuilt/a3_aimdk-3.1.0-py3-none-any.whl 安装，ros2 包需要 source prebuilt/ros2_plugin_proto_aarch64/share/ros2_plugin_proto/local_setup.bash后使用。

请注意，以上程序会接收 ros2 消息，需要设置如下环境变量：

bash

 source /agibot/software/v0/entry/env/env.sh

其中音频数据为 16kHz 采样率，16 位单声道 PCM 数据（小端序），输出的音频为纯净人声，已经过降噪处理，回声消除等，可以直接用于 ASR 识别。

ProcessedAudioOutput消息包含以下字段：

字段名	类型	说明
header	Header	通用消息头，包含时间戳和消息 ID
stream_id	uint32	音频流 ID（1：内置麦克风，2：外置麦克风）
vad_state	AudioVADState	语音活动检测状态
audio_data	bytes	降噪处理后的 PCM 音频数据

AudioVADState 枚举定义：

枚举值	数值	说明
AUDIO_VAD_STATE_NONE	0	无语音
AUDIO_VAD_STATE_BEGIN	1	语音开始
AUDIO_VAD_STATE_PROCESSING	2	语音处理中
AUDIO_VAD_STATE_END	3	语音结束

注意：当前版本外置麦克风的接口输出 vad_state 存在问题，预期一条语音输入的状态为 122222222223，实际输出的状态为 0111111111112，此问题仅在外置麦场景下出现（内置麦正常），并计划在后续版本修复。当前版本建议手动对状态执行 +1 补偿。

8.2.3 扬声器音频播放

要想使用扬声器首先需获取音频焦点：

bash

ros2 service call /audio_5Fmsgs/srv/RequestAudioFocus audio_msgs/srv/RequestAudioFocus "{focus_requester: {pkg_name: audio_examples_sender, priority: 6, priority_weight: 1}}"

pkg_name为ROS包名，意味着音频最后必须以在ROS包工程内的形式进行发送。priority为优先级固定为6即可，priority_weight为音频权重，可设置为1-100任意数。

申请音频焦点后即可将自身TTS返回的pcm数据流通过音频流播放接口进行播放，扬声器播放流式音频具体例程如下，需打包成ros_pkg。

bash

#!/usr/bin/env python3

import rclpy
from rclpy.node import Node
from builtin_interfaces.msg import Time as TimeMsg
from audio_msgs.msg import AudioPlayback,AudioInfo,AudioData  # 需确保 audio msg 可导入

class AudioPlaybackPublisher(Node):
    def __init__(self):
        super().__init__('audio_playback_publisher')
        self.pub = self.create_publisher(AudioPlayback, '/audiohal/audio/playback', 10)
        self.timer = self.create_timer(1.0, self.timer_callback)  # 每秒发送

    def timer_callback(self):
        msg = AudioPlayback()

        # 填写时间戳（字段名 attachments 中为 stamps）
        now = self.get_clock().now().to_msg()
        # 兼容性：优先使用 stamps，否则尝试 stamp
        if hasattr(msg, 'stamps'):
            msg.stamps = now
        elif hasattr(msg, 'stamp'):
            msg.stamp = now

        # 填写 info（AudioInfo）
        info = AudioInfo()
        info.channels = 1          # 单声道
        info.sample_rate = 16000   # 16kHz
        info.sample_format = 'S16LE'
        info.coding_format = 'pcm'
        msg.info = info

        # 填写 data（AudioData），示例给一小段原始 bytes
        sample_bytes = b'\x00\x01\x02\x03\x04\x05\x06\x07'  # 示例数据
        msg.data = AudioData(data=list(sample_bytes))

        msg.pkg_name = 'audio_examples_sender'
        msg.token_id = 'token-0001'

        self.pub.publish(msg)
        self.get_logger().info(
            f'Published AudioPlayback pkg={msg.pkg_name} token={msg.token_id} '
            f'data_len={len(msg.data.data)} sample_rate={msg.info.sample_rate}'
        )


def main(args=None):
    rclpy.init(args=args)
    node = AudioPlaybackPublisher()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == '__main__':
    main()

以上程序在AimDK开发包agent目录下展示，且程序依赖 python 协议包 a3_aimdk 和 ros2 协议包 ros2_plugin_proto，相应的包已经放置在 AimDK 开发包的 prebuilt 目录下，python 包使用 pip install prebuilt/a3_aimdk-3.1.0-py3-none-any.whl安装，ros2 包需要 source prebuilt/audio_msgs_proto_aarch64/share/audio_msgs/local_setup.bash后使用。

8.2.4 唤醒结果上报

若 agent 设置为 only_voice 模式，智元还提供唤醒结果上报能力， Topic 接口为 /agent/wakeup/pb_3Aaimdk_2Eprotocol_2EWakeUpResult

获取唤醒结果上报示例程序如下：

python

#!/usr/bin/env python3

import rclpy
from rclpy.node import Node
from rclpy.qos import (
    QoSDurabilityPolicy,
    QoSHistoryPolicy,
    QoSProfile,
    QoSReliabilityPolicy,
)
from ros2_plugin_proto.msg import RosMsgWrapper

from aimdk.protocol_pb2 import WakeUpResult


TOPIC = "/agent/wakeup/pb_3Aaimdk_2Eprotocol_2EWakeUpResult"


class WakeUpSubscriber(Node):
    def __init__(self):
        super().__init__("wakeup_result_subscriber")

        qos_profile = QoSProfile(
            history=QoSHistoryPolicy.KEEP_LAST,
            depth=10,
            reliability=QoSReliabilityPolicy.RELIABLE,
            durability=QoSDurabilityPolicy.VOLATILE,
        )

        self.subscription = self.create_subscription(
            RosMsgWrapper,
            TOPIC,
            self.wakeup_callback,
            qos_profile,
        )

        self.get_logger().info(f"已开始订阅 WakeUpResult: {TOPIC}")

    def wakeup_callback(self, msg):
        try:
            if msg.serialization_type != "pb":
                self.get_logger().warn(
                    f"收到不支持的序列化类型: {msg.serialization_type}"
                )
                return

            # 拼接 bytes
            raw_bytes = b"".join(msg.data)

            # 解析 protobuf
            wakeup_result = WakeUpResult()
            wakeup_result.ParseFromString(raw_bytes)

            # 日志输出
            import json
            from google.protobuf.json_format import MessageToDict

            self.get_logger().info(
                f"WakeUpResult: {json.dumps(MessageToDict(wakeup_result, preserving_proto_field_name=True), ensure_ascii=False, indent=2)}"
            )

        except Exception as e:
            self.get_logger().error(f"解析 WakeUpResult 数据时出现错误: {e}")


def main(args=None):
    rclpy.init(args=args)
    node = WakeUpSubscriber()

    try:
        node.get_logger().info("正在监听 WakeUpResult，按 Ctrl+C 退出...")
        rclpy.spin(node)
    except KeyboardInterrupt:
        node.get_logger().info("退出中...")
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == "__main__":
    main()

请注意，以上程序会接收 ros2 消息，需要设置如下环境变量：

bash

 source /agibot/software/v0/entry/env/env.sh

8.2.5 人脸识别结果

若 agent 设置为 voice_face 模式，除输出降噪麦克音频 Topic 接口外，同时返回人脸识别结果 Topic 接口 /agent/vision/face_id，only_voice 模式下无此消息

获取人脸识别结果示例程序如下：

python

#!/usr/bin/env python3

import rclpy
from rclpy.node import Node
from rclpy.qos import QoSHistoryPolicy, QoSProfile, QoSReliabilityPolicy
from ros2_plugin_proto.msg import RosMsgWrapper

from aimdk.protocol_pb2 import FaceIdResult


class FaceIdSubscriber(Node):
    def __init__(self):
        super().__init__("face_id_subscriber")

        qos_profile = QoSProfile(
            history=QoSHistoryPolicy.KEEP_LAST,
            depth=10,
            reliability=QoSReliabilityPolicy.BEST_EFFORT,
        )

        self.subscription = self.create_subscription(
            RosMsgWrapper,
            "/agent/vision/face_id/pb_3Aaimdk_2Eprotocol_2EFaceIdResult",
            self.face_id_callback,
            qos_profile,
        )

        self.get_logger().info("已开始订阅 FaceID 数据...")

    def face_id_callback(self, msg):
        try:
            if msg.serialization_type != "pb":
                self.get_logger().warn(
                    f"收到不支持的序列化类型: {msg.serialization_type}"
                )
                return

            # 拼接 bytes
            raw_bytes = b"".join(msg.data)

            # 解析 protobuf
            face_id_result = FaceIdResult()
            face_id_result.ParseFromString(raw_bytes)

            # 日志输出
            import json
            from google.protobuf.json_format import MessageToDict
            self.get_logger().info(
                f"FaceID 结果: {json.dumps(MessageToDict(face_id_result, preserving_proto_field_name=True), ensure_ascii=False, indent=2)}")

        except Exception as e:
            self.get_logger().error(f"解析 FaceID 数据时出现错误: {e}")


def main(args=None):
    rclpy.init(args=args)
    node = FaceIdSubscriber()

    try:
        node.get_logger().info("正在监听 FaceID 数据，按 Ctrl+C 退出...")
        rclpy.spin(node)
    except KeyboardInterrupt:
        node.get_logger().info("退出中...")
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == "__main__":
    main()

以上程序依赖 python 协议包 a3_aimdk 和 ros2 协议包 ros2_plugin_proto，相应的包已经放置在 AimDK 开发包的 prebuilt 目录下，python 包使用 pip install prebuilt/a3_aimdk-3.0.0-py3-none-any.whl 安装，ros2 包需要 source prebuilt/ros2_plugin_proto_aarch64/share/ros2_plugin_proto/local_setup.bash后使用。

请注意，以上程序会接收 ros2 消息，需要设置如下环境变量：

bash

 source /agibot/software/v0/entry/env/env.sh