vosk-ASR asterisk-ari调用[AI人工智能(五十四)]—东方仙盟

未来之窗软件服务

34人浏览 · 2026-03-16 23:02:54

未来之窗软件服务 · 2026-03-16 23:02:54 发布

目录结构

核心代码

完整代码

#!/usr/bin/python3

import anyio
import asyncari
import logging
import aioudp
import os
import vosk
import array

ast_host = os.getenv("AST_HOST", '127.0.0.1')
ast_port = int(os.getenv("AST_ARI_PORT", 8088))
ast_url = os.getenv("AST_URL", 'http://%s:%d/'%(ast_host,ast_port))
ast_username = os.getenv("AST_USER", 'asterisk')
ast_password = os.getenv("AST_PASS", 'asterisk')
ast_app = os.getenv("AST_APP", 'hello-world')


model = vosk.Model(lang='en-us')
channels = {}

class Channel:

    async def rtp_handler(self, connection):
        async for message in connection:
            data = array.array('h', message[12:])
            data.byteswap()
            if self.rec.AcceptWaveform(data.tobytes()):
                res = self.rec.Result()
            else:
                res = self.rec.PartialResult()
            print(res)

    async def init(self, client, channel):
        self.port = 45000 + len(channels)
        self.rec = vosk.KaldiRecognizer(model, 16000)
        self.udp = aioudp.serve("127.0.0.1", self.port, self.rtp_handler)
        await self.udp.__aenter__()

        bridge = await client.bridges.create(type='mixing')
        media_id = client.generate_id()
        await client.channels.externalMedia(channelId=media_id, app=client._app, external_host='127.0.0.1:' + str(self.port), format='slin16')
        await bridge.addChannel(channel=[media_id, channel.id])

async def statis_handler(objs, ev, client):
    channel = objs['channel']
    channel.answer()
    if 'UnicastRTP' in channel.name:
         return

    local_channel = Channel()
    await local_channel.init(client, channel)
    channels[channel.id] = local_channel

async def main():
     async with asyncari.connect(ast_url, ast_app, ast_username,ast_password) as client:
         async with client.on_channel_event('StasisStart') as listener:
             async for objs, event in listener:
                  await statis_handler(objs, event, client)

if __name__ == "__main__":
    logging.basicConfig(level=logging.DEBUG)
    anyio.run(main)

调用方法

This is an example of channel transcription through ARI/externalMedia

To configure statis make sure modules are loaded:

```
load = res_rtp_multicast.so
load = res_ari.so ; Asterisk RESTful Interface
load = res_ari_applications
load = res_ari_asterisk.so
load = res_ari_bridges
load = res_ari_channels.so
load = res_ari_device_states.so ; RESTful API module - Device state resources
load = res_ari_endpoints.so
load = res_ari_events.so
load = res_ari_model.so ; ARI Model validators
load = res_ari_playbacks.so
load = res_ari_recordings.so
load = res_ari_sounds
load = res_stasis.so ; Stasis application support
load = app_stasis.so ; Stasis dialplan application
load = res_stasis_answer.so
load = res_stasis_device_state.so ; Stasis application device state support
load = res_stasis_playback.so ; Stasis application playback support
load = res_stasis_recording.so ; Stasis application recording support
load = res_stasis_snoop.so ; Stasis application snoop support
```

then check ari.conf:

```
[general]
enabled = yes          
pretty = yes           
allowed_origins = *

[asterisk]
type = user
read_only = no
password = asterisk
```

Then the dialplan

```
exten => 4,1,Answer()
same =  n,Stasis(hello-world)
same =  n,Hangup()
```

Install required python modules


```
pip3 install aioudp asyncari vosk
```

And start the statis app:

```
python3 vosk_ari.py
```

Dial the extension, you should see the transcript on console

示意图

实现原理

一、整体功能（仙盟视角总览）

这段内容本质是搭建一套「仙盟语音传令系统」：

Asterisk（星号服务器）是仙盟总坛，负责接收各路弟子（电话）的语音传令；
ARI（Asterisk RESTful 接口）是仙盟传讯令符，让外部弟子（Python 程序）能和总坛互通消息；
Stasis 是总坛议事堂，专门处理需要特殊处理的传令（比如语音转写）；
Vosk 是仙盟译经师，把弟子的口语传令（语音）转成文字（转写）；
整个流程：弟子拨打总坛分机号→总坛接起→把传令转到议事堂→译经师实时翻译→把文字版传令显示在总坛控制台。

二、逐段拆解（仙盟场景对应解释）

1. 加载 Asterisk 模块（给总坛配备法器）

plaintext

load = res_rtp_multicast.so
load = res_ari.so ; Asterisk RESTful Interface
...
load = res_stasis_snoop.so ; Stasis application snoop support

比喻：这些load = xxx.so是给仙盟总坛配备核心法器 ——
- res_rtp_multicast.so：负责语音信号传输的「传音法器」；
- res_ari.so及相关：激活「传讯令符（ARI）」，让外部程序能和总坛对话；
- res_stasis.so/app_stasis.so：搭建「议事堂（Stasis）」，专门处理特殊传令；
- 其他res_stasis_xxx.so：给议事堂配辅助人员（处理接听、录音、监听等）。
作用：只有加载这些模块，Asterisk 才能支持 ARI 接口和 Stasis 应用，是后续功能的基础。

2. 配置 ari.conf（给传讯令符设使用规则）

plaintext

[general]
enabled = yes          ; 启用传讯令符
pretty = yes           ; 令符返回的消息格式工整（方便人看）
allowed_origins = *    ; 允许所有外部弟子使用令符（无限制）

[asterisk]
type = user            ; 定义使用令符的弟子身份
read_only = no         ; 弟子不仅能看消息，还能发指令
password = asterisk    ; 使用令符的密码（简单密码仅作示例）

比喻：给「传讯令符（ARI）」定规矩 —— 允许使用、谁能用、能做什么。
作用：开启 ARI 接口，设置访问权限和认证信息，让 Python 程序能通过 ARI 控制 Asterisk。

3. 配置拨号计划（给总坛设接令规则）

plaintext

exten => 4,1,Answer()  ; 弟子拨打分机号4时，总坛先接起传令
same =  n,Stasis(hello-world)  ; 把这个传令转到「hello-world议事堂」处理
same =  n,Hangup()     ; 处理完后挂断传令

比喻：仙盟总坛的接令规则 —— 只要有人打分机 4，先接电话，再把这个通话交给名为hello-world的议事堂处理，最后挂电话。
作用：定义电话呼入后的处理逻辑，核心是把通话交给 Stasis 应用，让外部 Python 程序能接管通话的语音处理。

4. 安装 Python 模块（给译经师配工具）

plaintext

pip3 install aioudp asyncari vosk

比喻：给「译经师（Vosk）」配备工作工具 ——
- vosk：译经师的核心本领（语音转文字）；
- asyncari：让译经师能看懂「传讯令符（ARI）」的消息格式；
- aioudp：译经师接收语音数据的「收音法器」。
作用：安装 Python 依赖库，让后续的vosk_ari.py程序能通过 ARI 连接 Asterisk，并调用 Vosk 实现语音转写。

5. 启动 Stasis 应用（让译经师到议事堂上班）

plaintext

python3 vosk_ari.py

比喻：让带着工具的译经师（vosk_ari.py程序）到「hello-world 议事堂」上班，等待处理传令。
作用：运行 Python 程序，该程序会通过 ARI 连接 Asterisk，监听 Stasis 应用的事件，当有通话进入 Stasis 时，实时获取语音流，用 Vosk 转写并输出到控制台。

6. 最终效果（完整流程）

拨号分机 4 → Asterisk 接起通话 → 把通话交给 Stasis → Python 程序通过 ARI 获取通话语音 → Vosk 转写语音为文字 → 控制台显示转写结果。

东方仙盟：拥抱知识开源，共筑数字新生态

在全球化与数字化浪潮中，东方仙盟始终秉持开放协作、知识共享的理念，积极拥抱开源技术与开放标准。我们相信，唯有打破技术壁垒、汇聚全球智慧，才能真正推动行业的可持续发展。

开源赋能中小商户：通过将前端异常检测、跨系统数据互联等核心能力开源化，东方仙盟为全球中小商户提供了低成本、高可靠的技术解决方案，让更多商家能够平等享受数字转型的红利。
共建行业标准：我们积极参与国际技术社区，与全球开发者、合作伙伴共同制定开放协议与技术规范，推动跨境零售、文旅、餐饮等多业态的系统互联互通，构建更加公平、高效的数字生态。
知识普惠，共促发展：通过开源社区、技术文档与培训体系，东方仙盟致力于将前沿技术转化为可落地的行业实践，赋能全球合作伙伴，共同培育创新人才，推动数字经济的普惠式增长

阿雪技术观

在科技发展浪潮中，我们不妨积极投身技术共享。不满足于做受益者，更要主动担当贡献者。无论是分享代码、撰写技术博客，还是参与开源项目维护改进，每一个微小举动都可能蕴含推动技术进步的巨大能量。东方仙盟是汇聚力量的天地，我们携手在此探索硅基生命，为科技进步添砖加瓦。

Hey folks, in this wild tech - driven world, why not dive headfirst into the whole tech - sharing scene? Don't just be the one reaping all the benefits; step up and be a contributor too. Whether you're tossing out your code snippets , hammering out some tech blogs, or getting your hands dirty with maintaining and sprucing up open - source projects, every little thing you do might just end up being a massive force that pushes tech forward. And guess what? The Eastern FairyAlliance is this awesome place where we all come together. We're gonna team up and explore the whole silicon - based life thing, and in the process, we'll be fueling the growth of technology