voicebot/agi-scripts/voicebot.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import uuid
import requests
from asterisk.agi import AGI


TMP_DIR = "/tmp/voicebot"

STT_URL = "http://127.0.0.1:5001/transcribe"
TTS_URL = "http://127.0.0.1:5002/synthesize"

OLLAMA_URL = "http://127.0.0.1:11434/api/generate"
OLLAMA_MODEL = "qwen3:14b"



def log(agi, msg):
    agi.verbose(
        "VOICEBOT: " + str(msg),
        1
    )



def tts_generate(agi, text, name):

    try:

        log(
            agi,
            "TTS request: " + text
        )


        r = requests.post(
            TTS_URL,
            json={
                "text": text
            },
            timeout=60
        )

        r.raise_for_status()


        path = TMP_DIR + "/" + name + ".wav"


        with open(path, "wb") as f:
            f.write(r.content)


        log(
            agi,
            "TTS created " + path
        )


        return path


    except Exception as e:

        log(
            agi,
            "TTS ERROR " + str(e)
        )

        return None



def play_file(agi, path):

    if not path:
        return


    try:

        file = path.replace(
            ".wav",
            ""
        )


        agi.appexec(
            "Playback",
            file
        )


    except Exception as e:

        log(
            agi,
            "PLAY ERROR " + str(e)
        )




def record_audio(agi, callid):


    filename = TMP_DIR + "/" + callid + "_record"


    log(
        agi,
        "Recording user voice"
    )


    try:


        result = agi.send_command(

            f'RECORD FILE "{filename}" wav "#" 10000 0 s=5'

        )


        log(
            agi,
            "Record result: " + str(result)
        )


    except Exception as e:


        log(
            agi,
            "Record ERROR " + str(e)
        )

        return None



    wav = filename + ".wav"


    if os.path.exists(wav):


        size=os.path.getsize(wav)


        log(
            agi,
            f"Recording saved {wav} size={size}"
        )


        return wav



    log(
        agi,
        "Recording file missing"
    )


    return None





def stt_process(agi,wav):


    try:


        log(
            agi,
            "Sending audio to STT"
        )


        with open(wav,"rb") as f:


            r=requests.post(

                STT_URL,

                files={
                    "audio":f
                },

                timeout=120

            )


        data=r.json()


        text=data.get(
            "text",
            ""
        )


        log(
            agi,
            "STT text: " + text
        )


        return text



    except Exception as e:


        log(
            agi,
            "STT ERROR " + str(e)
        )


        return ""





def ask_ai(agi,text):


    try:


        log(
            agi,
            "AI request: " + text
        )


        prompt = """
تو دستیار تلفنی هستی.
کوتاه و طبیعی جواب بده.
حداکثر دو جمله.

کاربر:
""" + text



        r=requests.post(

            OLLAMA_URL,

            json={

                "model":OLLAMA_MODEL,

                "prompt":prompt,

                "stream":False,

                "options":{
                    "num_predict":80,
                    "temperature":0.3
                }

            },

            timeout=180

        )


        data=r.json()


        answer=data.get(
            "response",
            ""
        )


        log(
            agi,
            "AI response: " + answer
        )


        return answer



    except Exception as e:


        log(
            agi,
            "AI ERROR " + str(e)
        )


        return "ارتباط با هوش مصنوعی برقرار نشد."





def main():


    os.makedirs(
        TMP_DIR,
        exist_ok=True
    )


    agi=AGI()


    callid=str(uuid.uuid4())[:8]


    log(
        agi,
        "Call started " + callid
    )


    try:


        agi.answer()



        welcome=tts_generate(

            agi,

            "سلام، من دستیار صوتی هوشمند هستم. بفرمایید.",

            callid+"_welcome"

        )


        play_file(
            agi,
            welcome
        )



        wav=record_audio(
            agi,
            callid
        )


        if not wav:

            log(
                agi,
                "No recording"
            )

            return



        text=stt_process(
            agi,
            wav
        )



        if text:

            answer=ask_ai(
                agi,
                text
            )

        else:

            answer="صدای شما را دریافت نکردم."



        reply=tts_generate(

            agi,

            answer,

            callid+"_reply"

        )


        play_file(
            agi,
            reply
        )


        log(
            agi,
            "Call completed"
        )



    except Exception as e:


        log(
            agi,
            "MAIN ERROR " + str(e)
        )


    finally:

        agi.hangup()





if __name__ == "__main__":

    main()