科大讯飞收费标准

控制台-讯飞开放平台

可基于控制台注册免费试用5小时（直接点击套餐有免费额度套餐）

总结一下个人认为的效果，标准版已经out，模型版太贵，不如本地部署。有钱无所谓

以下为针对“实时语音转写（标准版）”与“实时语音转写大模型”两者区别及应用场景：

1. 问：实时语音转写（标准版）是什么？ 答：基于深度全序列卷积神经网络，通过 WebSocket 长连接实时将连续音频流转换为文字.

2. 问：实时语音转写大模型是什么？ 答：建立在星火大模型预训练框架上，支持多语种与方言免切识别，能智能断句和补全标点.

3. 问：标准版支持哪些音频格式？ 答：仅支持采样率16 kHz、位深16 bit、pcm_s16le单声道音频.

4. 问：标准版支持哪些语种？ 答：默认中文普通话，中英混合和英文；其他小语种及方言需在控制台开通后使用.

5. 问：大模型版支持哪些语种和方言？ 答：免切换支持37种语种和202种中文方言的混合识别.

6. 问：标准版适合哪些应用场景？ 答：线上会议实时字幕、网络直播实时转写、客服中心对话监控等持续音频流场景.

7. 问：大模型版适合哪些应用场景？ 答：跨区域多方言视频会议、国际化直播、电商带货等需覆盖多语种和方言的场景.

8. 问：两者核心区别是什么？ 答：标准版聚焦低延迟与稳定性，仅支持常规语种；大模型版聚焦多语种与方言覆盖、断句标点及高识别精度.

标准版购买页显示各时长包均标注“并发路数：50路”，其余为价格、时长与有效期信息。单独方言购买优惠期5k/年，无优惠2w/年。

大模型为新版能力，产品文档提示其在方言与多语种上较标准版更强，以下价格来自网站。

每分钟单价：大模型各档约为0.0333–0.0825元/分钟，标准版各档约为0.0133–0.0825元/分钟

每小时单价：大模型各档约为4.95-2.00 元/小时，标准版各档约为4.95 –0.80 元/小时

使用教程

个人认为标准版本已被淘汰，主流使用大语言模型版本。

基于我的代码开发，你需要做的是替换API和外部接口加入。

self.app_id = kwargs.get('app_id', '')
self.access_key_id = kwargs.get('access_key_id', '')
self.access_key_secret = kwargs.get('access_key_secret', '')

外部接口方法

#单例模式初始化
 self.xfyunmodle_asr = create_asr(asr_type="xfyun_modle")
 if not self.xfyunmodle_asr.init():
     logger.error("error:XFYunModle ASR组件初始化失败")
 logger.info("✅ XFYunModle ASR组件初始化成功")


#转录工作
self.xfyunmodle_asr.generate(
                  audio=combined_audio,
                  is_final=False
                ) 
#类内对象 self.current_result 接收websocket 转录的文本。

对象代码

class XFYunModle(ASRAUTO):
    """
    基于官方demo扩展 讯飞云实时语音转写API的ASR实现
    使用WebSocket进行实时语音识别，严格控制发送节奏
    """
    
    # 全局配置：与服务端确认的固定参数
    FIXED_PARAMS = {
        "audio_encode": "pcm_s16le",
        "lang": "autodialect",
        "samplerate": "16000"  # 固定16k采样率，对应每40ms发送1280字节
    }
    AUDIO_FRAME_SIZE = 1280  # 每帧音频字节数（16k采样率、16bit位深、40ms）
    FRAME_INTERVAL_MS = 40    # 每帧发送间隔（毫秒）
    
    def __init__(self):
        """初始化XFYun Modle ASR实例"""
        self.app_id = None
        self.access_key_id = None
        self.access_key_secret = None
        self.base_ws_url = "wss://office-api-ast-dx.iflyaisol.com/ast/communicate/v1"
        self.ws = None
        self.is_connected = False
        self.recv_thread = None
        self.session_id = None
        self._is_loaded = False
        self.xfyun_logger = get_logger(__name__ + ".xfyun_modle")
        
        # 识别结果管理
        self.current_result = ""
        self.result_lock = threading.Lock()
        self.result_ready = threading.Event()
        
        # 音频缓冲区
        self.audio_buffer = b''
    
    def init(self, path: str = "", device: str = "cpu", **kwargs) -> bool:
        """
        初始化XFYun Modle ASR
        
        Args:
            path: 配置文件路径（可选）
            device: 设备类型（忽略，XFYun是云端服务）
            **kwargs: 其他参数，可包含app_id, access_key_id, access_key_secret
            
        Returns:
            bool: 初始化是否成功
        """
        try:
            # 从kwargs获取配置
            self.app_id = kwargs.get('app_id', '')
            self.access_key_id = kwargs.get('access_key_id', '')
            self.access_key_secret = kwargs.get('access_key_secret', '')
            
            self._is_loaded = True
            self.xfyun_logger.info("✅ XFYun Modle ASR初始化成功")
            return True
            
        except Exception as e:
            self.xfyun_logger.error(f"XFYun Modle ASR初始化失败: {e}")
            self._is_loaded = False
            return False
    
    def _get_utc_time(self):
        """生成服务端要求的UTC时间格式：yyyy-MM-dd'T'HH:mm:ss+0800"""
        beijing_tz = datetime.timezone(datetime.timedelta(hours=8))
        now = datetime.datetime.now(beijing_tz)
        return now.strftime("%Y-%m-%dT%H:%M:%S%z")
    
    def _generate_auth_params(self):
        """生成鉴权参数（严格按字典序排序，匹配Java TreeMap）"""
        auth_params = {
            "accessKeyId": self.access_key_id,
            "appId": self.app_id,
            "uuid": uuid.uuid4().hex,
            "utc": self._get_utc_time(),
            **self.FIXED_PARAMS
        }
        
        # 计算签名：过滤空值 → 字典序排序 → URL编码 → 拼接基础字符串
        sorted_params = dict(sorted([
            (k, v) for k, v in auth_params.items()
            if v is not None and str(v).strip() != ""
        ]))
        base_str = "&".join([
            f"{urllib.parse.quote(k, safe='')}={urllib.parse.quote(v, safe='')}"
            for k, v in sorted_params.items()
        ])
        
        # HMAC-SHA1 加密 + Base64编码
        signature = hmac.new(
            self.access_key_secret.encode("utf-8"),
            base_str.encode("utf-8"),
            hashlib.sha1
        ).digest()
        auth_params["signature"] = base64.b64encode(signature).decode("utf-8")
        return auth_params
    
    def _connect(self):
        """建立WebSocket连接"""
        try:
            auth_params = self._generate_auth_params()
            params_str = urllib.parse.urlencode(auth_params)
            full_ws_url = f"{self.base_ws_url}?{params_str}"
            self.xfyun_logger.debug(f"连接URL：{full_ws_url}")
            
            # 初始化WebSocket连接
            self.ws = create_connection(
                full_ws_url,
                timeout=15,
                enable_multithread=True
            )
            self.is_connected = True
            self.xfyun_logger.debug("WebSocket握手完成，等待服务端就绪...")
            #time.sleep(1.5)  # 确保服务端完全初始化
            
            # 启动接收线程
            self.recv_thread = threading.Thread(target=self._recv_msg, daemon=True)
            self.recv_thread.start()
            return True
            
        except WebSocketException as e:
            self.xfyun_logger.error(f"WebSocket连接失败：{str(e)}")
            return False
        except Exception as e:
            self.xfyun_logger.error(f"连接异常：{str(e)}")
            return False
    
    def _recv_msg(self):
        """接收服务端消息"""
        while True:
            if not self.is_connected or not self.ws:
                self.xfyun_logger.debug("接收线程：连接已关闭，退出接收循环")
                break
            
            try:
                msg = self.ws.recv()
                if not msg:
                    self.xfyun_logger.debug("服务端关闭连接")
                    self._close()
                    break
                
                # 仅处理文本消息
                if isinstance(msg, str):
                    try:
                        msg_json = json.loads(msg)
                        #https://www.xfyun.cn/doc/spark/asr_llm/rtasr_llm.html#_1-%E6%A6%82%E8%BF%B0                        
                        msg_type = msg_json.get('msg_type', '')
                        #self.xfyun_logger.debug(f"接收消息 - msg_type: {msg_type}, data类型: {type(msg_json.get('data'))}")
                        
                        # 更新会话ID
                        if (msg_type == 'action' 
                            and 'sessionId' in msg_json.get('data', {})):
                            self.session_id = msg_json['data']['sessionId']
#                            self.xfyun_logger.debug(f"更新session_id: {self.session_id}")
                        
                        # 处理识别结果
                        if msg_type == 'result':
                            self._process_recognition_result(msg_json)
                            
                    except json.JSONDecodeError:
                        self.xfyun_logger.warning(f"非JSON文本消息：{msg[:50]}...")
                else:
                    self.xfyun_logger.debug(f"收到二进制消息（长度：{len(msg)}字节），忽略")
            
            except WebSocketException as e:
                self.xfyun_logger.error(f"连接中断：{str(e)}")
                self._close()
                break
            except OSError as e:
                self.xfyun_logger.error(f"系统套接字错误：{str(e)}")
                self._close()
                break
            except Exception as e:
                self.xfyun_logger.error(f"接收异常：{str(e)}")
                self._close()
                break
    
    def _process_recognition_result(self, result_dict):
        """处理识别结果"""
        try:
            # 获取data字段，可能是字符串或已解析的字典
            data = result_dict.get("data", {})
            
            # 如果data是字符串，需要解析
            if isinstance(data, str):
                data = json.loads(data)
            
            # 如果data不是字典，无法处理
            if not isinstance(data, dict):
                self.xfyun_logger.warning(f"data字段类型错误: {type(data)}")
                return
            
            seg_id = data.get("seg_id", "")

            if "cn" in data and "st" in data["cn"]:
                st = data["cn"]["st"]
                type_ = st.get("type")

                if type_ == "0":  # 完整转写内容
                    text_parts = []
                    if "rt" in st:
                        for rt_item in st["rt"]:
                            if "ws" in rt_item:
                                for ws_item in rt_item["ws"]:
                                    if "cw" in ws_item:
                                        for cw_item in ws_item["cw"]:
                                            w = cw_item.get("w", "")
                                            text_parts.append(w)

                    if text_parts:
                        result_text = ''.join(text_parts)
                        with self.result_lock:
                            self.current_result = result_text
                            self.result_ready.set()
                        self.xfyun_logger.debug(f"XFYunModle识别结果: {result_text}")

        except Exception as e:
            self.xfyun_logger.error(f"XFYunModle ASR结果处理错误: {e}")


    def generate(self,
                audio: np.ndarray,
                cache: Optional[Dict] = None,
                is_final: bool = False,
                **kwargs) -> Optional[str]:
        """
        执行语音识别（流式缓冲发送模式，类似XFYunASR）
        
        Args:
            audio: 音频数据 (numpy array, 16kHz, 16bit)
            cache: 缓存（未使用）
            is_final: 是否为最终识别
            **kwargs: 其他参数
            
        Returns:
            Optional[str]: 识别结果文本
        """
        if not self._is_loaded:
            self.xfyun_logger.warning("XFYun Modle ASR未初始化")
            return None
        
        try:
            # 防止和para冲突，关闭最终帧
            is_final = False
            
            # 确保WebSocket连接存在
            if not self.ws or not self.is_connected:
                if not self._connect():
                    self.xfyun_logger.error("无法建立XFYun Modle WebSocket连接")
                    return None
            
            # 转换音频数据格式
            if audio.dtype != np.int16:
                audio = audio.astype(np.int16)
            audio_bytes = audio.tobytes()
            
            # 将音频数据添加到缓冲区
            self.audio_buffer += audio_bytes
            
            # 当缓冲区达到1280字节(40ms)时发送
            while len(self.audio_buffer) >= self.AUDIO_FRAME_SIZE:
                chunk_to_send = self.audio_buffer[:self.AUDIO_FRAME_SIZE]
                self.audio_buffer = self.audio_buffer[self.AUDIO_FRAME_SIZE:]
                
                # 发送音频数据
                self.ws.send_binary(chunk_to_send)
                #time.sleep(self.FRAME_INTERVAL_MS / 1000)  # 40ms间隔
            
            # 如果是最终帧，发送结束标记并等待最终结果
            if is_final:
                # 发送剩余的音频数据
                if len(self.audio_buffer) > 0:
                    self.ws.send_binary(self.audio_buffer)
                    self.audio_buffer = b''
                  #  time.sleep(self.FRAME_INTERVAL_MS / 1000)
                
                # 发送结束标记
                end_msg = {"end": True}
                if self.session_id:
                    end_msg["sessionId"] = self.session_id
                end_msg_str = json.dumps(end_msg, ensure_ascii=False)
                self.ws.send(end_msg_str)
                self.xfyun_logger.debug("已发送结束标记")
                
                # 等待最终识别结果
                max_wait_time = 5.0  # 最大等待5秒
                wait_time = 0
                while wait_time < max_wait_time:
                    with self.result_lock:
                        if self.current_result:
                            result = self.current_result
                            self.xfyun_logger.debug(f"XFYun Modle ASR结果: {result}")
                            self.current_result = ""  # 清空结果
                            self._close()
                            return result
                  #  time.sleep(0.1)
                    wait_time += 0.1
                
                # self._close()
                return None
            
            # 非最终帧，检查是否有中间结果
            with self.result_lock:
                if self.current_result:
                    result = self.current_result
                    #self.xfyun_logger.debug(f"XFYun Modle ASR中间结果: {result}")
                    self.current_result = ""  # 清空结果
                    return result
            
            return None
            
        except Exception as e:
            self.xfyun_logger.error(f"XFYun Modle语音识别失败：{e}")
            self._close()
            return None
    
    def _close(self):
        """安全关闭WebSocket连接"""
        if self.is_connected and self.ws:
            self.is_connected = False
            try:
                if self.ws.connected:
                    self.ws.close(status=1000, reason="客户端正常关闭")
                self.xfyun_logger.debug("WebSocket已安全关闭")
            except Exception as e:
                self.xfyun_logger.error(f"关闭时出错：{str(e)}")
        self.ws = None
    
    def is_loaded(self) -> bool:
        """检查模型是否已加载"""
        return self._is_loaded
    
    def close(self):
        """清理资源"""
        self._close()
        self.audio_buffer = b''
        self._is_loaded = False
        self.xfyun_logger.info("XFYun Modle ASR资源已清理")


def create_asr(asr_type: str = "paraformer") -> ASRAUTO:
    """
    创建ASR实例的工厂函数

    Args:
        asr_type: ASR类型 ("paraformer", "xfyun", "xfyun_modle" 或 "dummy")

    Returns:
        ASRAUTO: ASR实例
    """
    if asr_type.lower() == "paraformer":
        return ParaformerASR()
    elif asr_type.lower() == "xfyun":
        return XFYunASR()
    elif asr_type.lower() == "xfyun_modle":
        return XFYunModle()
    elif asr_type.lower() == "dummy":
        return DummyASR()
    else:
        logger.warning(f"未知的ASR类型: {asr_type}，使用虚拟ASR")
        return DummyASR()

学习社区
https://github.com/0voice