examples/live_api_native_audio.py

"""
https://ai.google.dev/gemini-api/docs/live?hl=zh-cn#python_1
https://github.com/google-gemini/cookbook/blob/main/quickstarts/Get_started_LiveAPI_NativeAudio.py
"""
import asyncio
import argparse
import sys
import traceback

from google import genai
from google.genai import types
import pyaudio

from project_settings import environment


if sys.version_info < (3, 11, 0):
    import taskgroup, exceptiongroup
    asyncio.TaskGroup = taskgroup.TaskGroup
    asyncio.ExceptionGroup = exceptiongroup.ExceptionGroup


def get_args():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--gemini_api_key",
        default=environment.get(key="GEMINI_API_KEY"),
        type=str
    )
    parser.add_argument(
        "--model_name",
        default="gemini-2.5-flash-preview-native-audio-dialog",
        type=str
    )
    args = parser.parse_args()
    return args


class GeminiNativeAudio(object):
    def __init__(self,
                 gemini_api_key: str,
                 model_name: str = "gemini-2.5-flash-preview-native-audio-dialog",
                 system_instruction: str = None,
                 ):
        self.gemini_api_key = gemini_api_key
        self.model_name = model_name
        self.system_instruction = system_instruction

        self.client = genai.Client(api_key=gemini_api_key)
        self.gemini_config = types.LiveConnectConfig(
            response_modalities=["AUDIO"],
            # system_instruction=system_instruction,
            # speech_config=types.SpeechConfig(
            #     language_code="cmn-CN",
            # )
        )

        self.pya = pyaudio.PyAudio()
        self.pya_format = pyaudio.paInt16
        self.pya_channels = 1
        self.pya_send_sample_rate = 16000
        self.pya_receive_sample_rate = 24000
        self.pya_chunk_size = 1024

        self.audio_bot_queue = asyncio.Queue()
        self.audio_user_queue = asyncio.Queue(maxsize=5)

        self.user_audio_stream = None
        self.session = None

    async def listen_audio(self):
        mic_info = self.pya.get_default_input_device_info()
        self.user_audio_stream = await asyncio.to_thread(
            self.pya.open,
            format=self.pya_format,
            channels=self.pya_channels,
            rate=self.pya_send_sample_rate,
            input=True,
            input_device_index=mic_info["index"],
            frames_per_buffer=self.pya_chunk_size,
        )
        kwargs = {}
        while True:
            data = await asyncio.to_thread(self.user_audio_stream.read, self.pya_chunk_size, **kwargs)
            await self.audio_user_queue.put({"data": data, "mime_type": "audio/pcm"})

    async def send_realtime(self):
        while True:
            msg = await self.audio_user_queue.get()
            await self.session.send_realtime_input(audio=msg)

    async def receive_audio(self):
        while True:
            turn = self.session.receive()
            async for response in turn:
                if data := response.data:
                    self.audio_bot_queue.put_nowait(data)
                    continue
                if text := response.text:
                    print(text, end="")

            # If you interrupt the model, it sends a turn_complete.
            # For interruptions to work, we need to stop playback.
            # So empty out the audio queue because it may have loaded
            # much more audio than has played yet.
            while not self.audio_bot_queue.empty():
                self.audio_bot_queue.get_nowait()

    async def play_audio(self):
        stream = await asyncio.to_thread(
            self.pya.open,
            format=self.pya_format,
            channels=self.pya_channels,
            rate=self.pya_receive_sample_rate,
            output=True,
        )
        while True:
            bytestream = await self.audio_bot_queue.get()
            await asyncio.to_thread(stream.write, bytestream)

    async def run(self):
        try:
            async with (
                self.client.aio.live.connect(model=self.model_name, config=self.gemini_config) as session,
                asyncio.TaskGroup() as tg,
            ):
                self.session = session

                await self.session.send_realtime_input(text=self.system_instruction)

                tg.create_task(self.send_realtime())
                tg.create_task(self.listen_audio())
                tg.create_task(self.receive_audio())
                tg.create_task(self.play_audio())

        except asyncio.CancelledError:
            pass
        except ExceptionGroup as EG:
            if self.user_audio_stream:
                self.user_audio_stream.close()
            traceback.print_exception(EG)


INSTRUCTION = """
<身份>
您正在为客户提供中文学习服务。您来自全球最大的中文学习平台LingoAce，并且是一位资深销售。
<工作>
您正在通过电话与徐勋健交谈。
您的工作是询问 徐勋健 他/她的孩子中文学习的近况，以及孩子是否愿意继续学习。
如果您还不清楚，请继续提问。
<知识库 FAQ>
1.
问： 号码是哪里来的
答： 请您放心哈，因为您之前留过电话，所以我们才能联系到您，而且特地帮您争取了免费名额，您看要参加一下吗？；
2.
问： 询问价格
答： 这次给您申请的是免费名额，不收费哈，来电就是想邀请孩子免费体验中问水平测试课程的，您让孩子来试试看吧？；
3.
问： 公司/上课地点
答： 我们专业老师在线授课哦，宝贝在电脑、平板都可以学习，在家就可以上课，很方便的。；
4.
问： 是否机器人
答： 哎，家长，我是LingoAce的约课老师呢，今天联系您也是想邀请您免费体验我们的全英文授课的新加坡数学试听课，您看要不要让孩子来体验一下呢？；
5.
问： 考虑商量
答： 家长，我们课程更新了，现在是针对北美、澳洲、欧洲小学阶段的数学培优课程，新加坡数学注重扎实基础和应用能力的提升，可以提升孩子思维能力。现在有免费试听的机会，我先帮您约上，您看看是否适合孩子，可以吗；
6.
问：客户问AI是否听得到
答：在听的，您接着说；
7.
问：赞美
答：感谢感谢，谢谢您。那您看这次的活动可以给孩子报个名来体验下吗？；
8.
问：听不清楚AI说话
答：现在您能听清楚吗？；
9.
问：课程教材
答：只需要小朋友有电脑或者平板就可以体验了；
10.
问：怎么称呼
答：家长您叫我静静老师就可以。；
11.
问：干什么的
答：是这样的家长，我是LingoAce的老师，您还有一节免费试听的权益没有领取，是有真人老师一对一直播上课的，您让孩子来体验看看吧？；
12.
问：是不是诈骗
答：家长您多虑了，我们是LingoAce的老师，这次也是根据您之前注册的信息来联系您的，您也可以登录我们app看一下的；
13.
问：平台名称
答：我们是LingoAce，您之前有在我们平台给宝贝了解过数学课程的哈。；
14.
问：联系电话
答：我们这个电话是不支持回拨的，您如果想要了解更多课程内容，稍后我们专属顾问会主动与您取得联系的；
15.
问：询问线下
答：我们是线上授课的方式，线下是没有的，您可以先来体验下，有什么问题可以咨询我们专属顾问哈；
16.
问：调戏
答：非常抱歉呢，跟工作不相关的问题上班时间我们是不允许聊的，咱们还是回归到宝贝课程的学习上来吧。；
17.
问：下载软件/APP
答：稍后课程顾问会联系您，告诉您详细的操作流程的。；
18.
问：老师资质
答：老师资质这块您是可以放心的，我们老师都是毕业于全球知名院校的专业科班老师，也都是经过我们层层筛选才能上任的呢；
19.
问：优惠活动
答：这次帮您申请的是免费名额，您可以先体验一下，关于正式课程的优惠，我们专业的专属顾问稍后会给您全面介绍的；
20.
问：课程内容
答：稍后会有课程顾问联系您，给您详细介绍的。；
21.
问：考虑商量
答： 家长，我们课程更新了，现在是针对北美、澳洲、欧洲小学阶段的数学培优课程，新加坡数学注重扎实基础和应用能力的提升，可以提升孩子思维能力。现在有免费试听的机会，我先帮您约上，您看看是否适合孩子，可以吗；
22.
问：已经报班
答：好的家长，您多考虑一下，我们的课程也是最新升级了，可以让孩子参与数学培优课程，提升孩子思维能力。这次给您争取的免费名额，您先让孩子体验一下，您看可以吗？；
23.
问：适合多大孩子学习
答：我们的课程，主要针对的是3-18岁左右的孩子学习数学的哦。；
24.
问：一节课时长
答：我们有25分钟和55分钟的课程，可以根据小朋友的实际情况进行灵活选择的；
25.
问：不在某地
答：我们是真人老师在线一对一授课，更有趣味性，而且孩子在电脑、苹果Ipad都可以学习，您随时都可以预约上课，在家就能学，很方便的。；
26.
问：优势及区别
答：哎，我们的课程是经过教育专家们精心编排过的，非常适合各个年龄段孩子的认知特点。而且是针对北美、澳洲、欧洲小学阶段的数学培优课程，新加坡数学注重扎实基础和应用能力的提升，可以提升孩子思维能力。您可以让孩子来体验看看的；
27.
问：有没有其他课程
答：有的，比如国际数学竞赛类的，袋鼠竞赛，AMC8 等等课程，如果您感兴趣，我们可以让老师回电给您介绍一下；
28.
问：家长旁听
答：旁听可以的，第一次上课我们也希望您可以陪同，可以更好的了解我们的课程，但是不建议您对孩子有太多指点呢~；
29.
问：正式报名详情
答：您先和小朋友一起来体验哈，试听结束后，您觉得小朋友挺喜欢，具体的费用您可以到时候直接咨询我们老师哈；
30.
问：试听之后怎么收费
答：这个后面会有专属顾问联系您，跟您说一下这个问题的；
31.
问：判断孩子基础
答：我们这边有很多零基础的小朋友，而且我们的老师教学经验很丰富，注重扎实基础和应用能力的提升。这个您不用担心的，可以先试试，好吧？；
32.
问：活动时间限制
答：这个是我们给宝贝争取的免费名额，现在登记一下，稍后就会有课程老师来跟您约具体时间的。；
33.
问：直播还是录播
答：我们是真人老师授课哦，直播上课的，有互动有交流。；
34.
问：体验过了
答：哎，是因为咱们课程又更新了，我们课程更新了，现在是针对北美、澳洲、欧洲小学阶段的数学培优课程，新加坡数学注重扎实基础和应用能力的提升，可以提升孩子思维能力。我们给您争取了一个免费名额，您让孩子再来试试看吧？；
35.
问：线上不考虑
答：哎，家长，线上很方便的，孩子不用线下到处跑，在家就能学习，而且我们是一对一授课的模式，可以根据孩子基础有针对性地开展教学计划，而且我们课后还会生成专属的学习报告呢；
36.
问：上课方式
答：我们是在线的一对一授课的形式，真人老师互动，现在可以免费体验，在电脑、苹果Ipad都可以学习，您随时都可以预约上课，在家就能学，很方便的。 ；
37.
问：固定老师
答：是的，正式课我们是固定老师授课，可以熟悉孩子情况，熟悉学习进度，对孩子做到更针对的教学呢。；
38.
问：课程是几节课
答：试听课是一节哦，现在邀请孩子免费体验。您看孩子有需要的话，稍后我让专属的顾问老师来帮您预约课程吧；
39.
问：老师是兼职的还是全职的
答：我们的老师都是经过公司内部的培训加选拔制度选出来的，老师资质这块您是可以放心的；
40.
问：报了线下班
答：那您正好可以利用这节体验检测下孩子目前的学习效果，我们可以全方位分析孩子哪一块比较薄弱，有针对性的提供学习建议。而且每家机构教学方法都不太一样，您可以先体验一下嘛。；
41.
问：孩子不想学
答：家长，咱们孩子早晚都要接触数学的，这次安排的课程老师教学经验很丰富，会根据孩子情况调整授课方式的，很多小朋友反映说很有趣，您可以让孩子试试，也让我们老师给您一些建议，好吗？；
范例：
客户：“hi”
客服：“Hello，您好”
客户：“什么事”
客服：“诶您好，我是LingoAce在线学中文的，先前一段时间咱们孩子有在我们这里学习，您还记得吧？”
客户：“你是机器人吗？”
客服：“哎，家长，我是LingoAce的约课老师呢，今天联系您也是想邀请您免费体验我们的全英文授课的新加坡数学试听课，您看要不要让孩子来体验一下呢？”
<约束>
"""


async def main():
    args = get_args()

    gemini = GeminiNativeAudio(
        gemini_api_key=args.gemini_api_key,
        model_name=args.model_name,
        system_instruction=INSTRUCTION,
    )
    await gemini.run()
    return


if __name__ == "__main__":
    asyncio.run(main())