当前位置：首页 > news >正文

语音识别技术之科大讯飞在线API

news 2025/10/23 9:34:51

科大讯飞收费标准

控制台-讯飞开放平台

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

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

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

问：实时语音转写（标准版）是什么？答：基于深度全序列卷积神经网络，通过 WebSocket 长连接实时将连续音频流转换为文字.
问：实时语音转写大模型是什么？答：建立在星火大模型预训练框架上，支持多语种与方言免切识别，能智能断句和补全标点.
问：标准版支持哪些音频格式？答：仅支持采样率16 kHz、位深16 bit、pcm_s16le单声道音频.
问：标准版支持哪些语种？答：默认中文普通话，中英混合和英文；其他小语种及方言需在控制台开通后使用.
问：大模型版支持哪些语种和方言？答：免切换支持37种语种和202种中文方言的混合识别.
问：标准版适合哪些应用场景？答：线上会议实时字幕、网络直播实时转写、客服中心对话监控等持续音频流场景.
问：大模型版适合哪些应用场景？答：跨区域多方言视频会议、国际化直播、电商带货等需覆盖多语种和方言的场景.
问：两者核心区别是什么？答：标准版聚焦低延迟与稳定性，仅支持常规语种；大模型版聚焦多语种与方言覆盖、断句标点及高识别精度.

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

套餐	价格(元)	服务时长(小时)	有效期	并发路数
时长套餐一	198	40	1年	50路
时长套餐二	2,000	600	1年	50路
时长套餐三	6,000	2,000	1年	50路
时长套餐四	10,800	6,000	1年	50路
时长套餐五	24,000	15,000	1年	50路
时长套餐六	120,000	100,000	1年	50路
时长套餐七	400,000	500,000	1年	50路

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

套餐	价格(元)	时长(小时)	有效期
免费包（个人）	0	5	1年
免费包（企业）	0	50	1年
套餐一	198	40	1年
套餐二	4,000	1,000	1年
套餐三	17,500	5,000	1年
套餐四	60,000	20,000	1年
套餐五	240,000	100,000	1年
套餐六	600,000	300,000	1年

每分钟单价：大模型各档约为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 = Noneself.access_key_id = Noneself.access_key_secret = Noneself.base_ws_url = "wss://office-api-ast-dx.iflyaisol.com/ast/communicate/v1"self.ws = Noneself.is_connected = Falseself.recv_thread = Noneself.session_id = Noneself._is_loaded = Falseself.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 ASRArgs:path: 配置文件路径（可选）device: 设备类型（忽略，XFYun是云端服务）**kwargs: 其他参数，可包含app_id, access_key_id, access_key_secretReturns: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 = Trueself.xfyun_logger.info("✅ XFYun Modle ASR初始化成功")return Trueexcept Exception as e:self.xfyun_logger.error(f"XFYun Modle ASR初始化失败: {e}")self._is_loaded = Falsereturn Falsedef _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_paramsdef _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 = Trueself.xfyun_logger.debug("WebSocket握手完成，等待服务端就绪...")#time.sleep(1.5)  # 确保服务端完全初始化# 启动接收线程self.recv_thread = threading.Thread(target=self._recv_msg, daemon=True)self.recv_thread.start()return Trueexcept WebSocketException as e:self.xfyun_logger.error(f"WebSocket连接失败：{str(e)}")return Falseexcept Exception as e:self.xfyun_logger.error(f"连接异常：{str(e)}")return Falsedef _recv_msg(self):"""接收服务端消息"""while True:if not self.is_connected or not self.ws:self.xfyun_logger.debug("接收线程：连接已关闭，退出接收循环")breaktry: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'))}")# 更新会话IDif (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()breakexcept OSError as e:self.xfyun_logger.error(f"系统套接字错误：{str(e)}")self._close()breakexcept Exception as e:self.xfyun_logger.error(f"接收异常：{str(e)}")self._close()breakdef _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)}")returnseg_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_textself.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 Nonetry:# 防止和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_idend_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 = 0while wait_time < max_wait_time:with self.result_lock:if self.current_result:result = self.current_resultself.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 resultreturn Noneexcept Exception as e:self.xfyun_logger.error(f"XFYun Modle语音识别失败：{e}")self._close()return Nonedef _close(self):"""安全关闭WebSocket连接"""if self.is_connected and self.ws:self.is_connected = Falsetry: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 = Nonedef is_loaded(self) -> bool:"""检查模型是否已加载"""return self._is_loadeddef close(self):"""清理资源"""self._close()self.audio_buffer = b''self._is_loaded = Falseself.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

查看全文

http://www.dtcms.com/a/516214.html