如何在Python中让WebSocket解析的text变量全局可用（Asterisk AGI场景）

我通过Asterisk AGI实现语音识别，WebSocket连接接收JSON格式的识别结果。当前脚本能将识别出的单词拼接成短语，但拼接得到的text变量无法用于后续操作（如发送至Telegram及工单系统），核心问题是局部变量无法跨函数访问。

原代码问题分析 #

原代码中text变量定义在process_chunk函数内部，属于局部变量，startAGI函数里的telegram_bot_sendtext(text)完全无法访问该变量。此外，每次处理音频块都会生成新的局部text，没有累积或传递机制。

解决方案 #

通过函数返回值+全局累积变量的方式解决，同时利用返回JSON中现成的text字段（比手动拼接更可靠）。

修改后的完整代码 #

#!/usr/bin/python3
from asterisk.agi import *
import os
from websocket import create_connection
import json
import traceback
import requests

AUDIO_FD = 3
CONTENT_TYPE = 'audio/l16; rate=8000; channels=1'
ACCEPT = 'audio/pcm'

def telegram_bot_sendtext(text):
    bot_token = '6069wxts_nWcA'
    bot_chatID = '-10300'
    bot_message = text
    send_text = f'https://api.telegram.org/bot{bot_token}/sendMessage?chat_id={bot_chatID}&reply_to_message_id=2&parse_mode=Markdown&text={bot_message}'
    response = requests.get(send_text)
    return response.json()

def process_chunk(agi, ws, buf):
    agi.verbose("Processing chunk")
    ws.send_binary(buf)
    res = json.loads(ws.recv())
    agi.verbose(f"Result: {str(res)}")
    current_text = ""
    if 'result' in res:
        # 优先使用返回JSON中的text字段，替代手动拼接
        current_text = res.get('text', " ".join([w['word'] for w in res['result']]))
        os.system(f"espeak -w /tmp/response22.wav \"{current_text.encode('utf-8')}\"")
        os.system("sox /tmp/response22.wav -r 8000 /tmp/response.wav")
        agi.stream_file("/tmp/response")
        os.remove("/tmp/response.wav")
    return current_text

def startAGI():
    agi = AGI()
    agi.verbose("EAGI script started...")
    ani = agi.env['agi_callerid']
    did = agi.env['agi_extension']
    agi.verbose(f"Call answered from: {ani} to {did}")
    ws = create_connection("ws://localhost:2700")
    ws.send('{ "config" : { "sample_rate" : 8000 } }')
    agi.verbose("Connection created")
    full_text = ""  # 累积所有识别结果的变量
    try:
        while True:
            data = os.read(AUDIO_FD, 8000)
            if not data:
                break
            # 接收当前块的识别文本并累积
            chunk_text = process_chunk(agi, ws, data)
            if chunk_text:
                full_text += f"{chunk_text} "
    except Exception as err:
        agi.verbose(''.join(traceback.format_exception(type(err), err, err.__traceback__)).replace('\n', ' '))
    try:
        # 发送整理后的完整文本到Telegram
        if full_text.strip():
            telegram_bot_sendtext(full_text.strip())
        else:
            agi.verbose("No recognized text to send")
    except Exception as exc:
        agi.verbose(f"Post_Auth_Script_Telega : Error : {str(exc)}")
    finally:
        ws.close()

startAGI()

关键修改点 #

1. 变量传递机制：process_chunk函数返回当前音频块的识别文本，startAGI中用full_text变量累积所有结果
2. 优化文本获取：优先使用返回JSON中的text字段，避免手动拼接可能出现的误差
3. 空值处理：增加判断，避免发送空文本到Telegram
4. 代码可读性：使用f-string替代字符串拼接，提升代码简洁性

内容的提问来源于stack exchange，提问作者Norbert Yuhas