如何利用交互式画布构建实时双子座 2.0 学习助手

Rifx.Online
Programming , Chatbots , Voice Assistants
19 Jan, 2025

Gemini 开发教程 V5

在本教程中，我们将继续使用 Gemini 2.0 及其多模态 Live API 构建迷人的实时聊天应用程序。这次我们将基于绘图画布构建一个具有实时语音和文本交互的学习助手。

您可能已经看过 OpenAI 的旧视频 https://youtu.be/_nSmkyDNulk，演示了 got-4o 如何通过提出引导性问题和提供提示来帮助学生在 Khan Academy 的屏幕上学习数学，从而引导学生找到正确答案。

通过使用我们之前的屏幕共享演示项目，我们可以轻松地通过共享学习网站或文档的屏幕，操作一些编辑工具，让 Gemini 2.0 帮助用户学习内容，从而复制相同的体验。

然而，对于现实世界的应用，仅依赖屏幕共享有几个限制。充其量，这是一种黑客方式，在现实世界中，有许多原因需要避免这种做法，因为它将应用程序与特定环境耦合。例如，即使 web 应用程序是用纯 HTML 和 Javascript 编写的，几乎不可能重用它以转移到移动应用程序。它充满了多线程与屏幕共享、共同编辑和语音通话，这需要跨应用程序操作。因此，当您想要开发一个强大且商业化的产品时，集成和控制用户体验可能会相当困难。

这就是专门围绕画布元素构建的应用内文档编辑器变得至关重要的地方。通过构建我们自己的图像加载和编辑界面，我们可以获得更多控制权，并为未来的功能和用途提供更高的可扩展性和灵活性。

系统架构

现在，让我们来看一下我们项目的设计。

我们将继续使用与之前的屏幕共享演示相同的项目结构，但我们将在客户端与服务器和 Gemini 多模态实时 API 的交互方式上引入一些关键变化。最大的变化是我们用一个交互式画布元素替代了屏幕共享组件，用户可以在其中加载图像并使用绘图工具进行一些基本编辑。客户端还包括音频捕获、播放、文本转录以及与服务器通信的 WebSocket 连接。

在服务器端，代码与我们上一个演示保持一致；也就是说，在接收到消息后，服务器将这两个媒体组件的图像和音频转发给实时 API，图像是从画布元素中捕获的。它还负责接收来自 Gemini API 的流响应并将其转发回客户端。语音流也将被传送到 Gemini 1.5 进行转录生成。

现在，我们将逐步讲解代码。由于行数稍多，我将只解释代码中最重要的部分。如果您想查看完整代码，可以在我的 GitHub 仓库中找到并下载，它不仅包含本教程中的项目，还包括 Gemini 2.0 系列中的之前项目，包括屏幕共享、转录输出和相机交互。

代码演示

现在，让我们开始代码演示的服务器端部分。

服务器设计

安装依赖项，包括两个 Gemini API，生产环境的 google-generativeai 用于语音转文本生成和实验性的多模态实时 API google-genai 用于 Gemini 2.0 实时音频和图像互动，以及用于音频处理的 websockets 和 pydub。

pip install --upgrade google-genai==0.3.0 google-generativeai==0.8.3 websockets pydub

同时，您还应该确保在您的计算机上安装了 FFmpeg 软件。对于 Ubuntu 系统，您可以使用 apt-get install ffmpeg 来安装。

接下来是代码：

import asyncio
import json
import os
import websockets
from google import genai
import base64
import io
from pydub import AudioSegment
import google.generativeai as generative
import wave

## Load API key from environment
os.environ['GOOGLE_API_KEY'] = ''
generative.configure(api_key=os.environ['GOOGLE_API_KEY'])
MODEL = "gemini-2.0-flash-exp"  # use your model ID
TRANSCRIPTION_MODEL = "gemini-1.5-flash-8b"

client = genai.Client(
  http_options={
    'api_version': 'v1alpha',
  }
)

第一部分通过环境变量配置 Google AI Studio 访问 API 的密钥，并指定我们将使用的模型。在这里，我们使用 gemini-2.0-flash-exp 模型进行实时图像和音频交互，并使用 gemini-1.5-flash-8b 模型进行语音转文本转录，因为它速度非常快且足够准确，适合这个简单的任务。

服务器端逻辑的核心是 gemini_session_handler 函数。

每当客户端建立新的 WebSocket 连接时，都会调用此函数。它管理与 Gemini 实时 API 的连接，并作为后台任务调用 send_to_gemini() 和 receive_from_gemini() 函数，这些是数据处理逻辑的主要部分。

async def gemini_session_handler(client_websocket: websockets.WebSocketServerProtocol):
    """处理 websocket 会话中与 Gemini API 的交互。"""
    try:
        config_message = await client_websocket.recv()
        config_data = json.loads(config_message)
        config = config_data.get("setup", {})
        
        config["system_instruction"] = """您是一个学习助手。
        检查学生的问题和答案。
        提出引导性问题并提供提示，引导学生找到正确答案。"""
        

        async with client.aio.live.connect(model=MODEL, config=config) as session:
            print("已连接到 Gemini API")

            async def send_to_gemini():
                """将来自客户端 websocket 的消息发送到 Gemini API。"""
                try:
                  async for message in client_websocket:
                      try:
                          data = json.loads(message)
                          if "realtime_input" in data:
                              for chunk in data["realtime_input"]["media_chunks"]:
                                  if chunk["mime_type"] == "audio/pcm":
                                      await session.send({"mime_type": "audio/pcm", "data": chunk["data"]})
                                      
                                  elif chunk["mime_type"] == "image/jpeg":
                                      await session.send({"mime_type": "image/jpeg", "data": chunk["data"]})
                                      
                      except Exception as e:
                          print(f"发送到 Gemini 时出错: {e}")
                  print("客户端连接关闭（发送）")
                except Exception as e:
                     print(f"发送到 Gemini 时出错: {e}")
                finally:
                   print("send_to_gemini 关闭")



            async def receive_from_gemini():
                """接收来自 Gemini API 的响应并将其转发给客户端，循环直到回合结束。"""
                try:
                    while True:
                        try:
                            print("接收来自 gemini 的数据")
                            async for response in session.receive():
                                if response.server_content is None:
                                    print(f'未处理的服务器消息! - {response}')
                                    continue

                                model_turn = response.server_content.model_turn
                                if model_turn:
                                    for part in model_turn.parts:
                                        if hasattr(part, 'text') and part.text is not None:
                                            await client_websocket.send(json.dumps({"text": part.text}))
                                        elif hasattr(part, 'inline_data') and part.inline_data is not None:
                                            print("音频 mime_type:", part.inline_data.mime_type)
                                            base64_audio = base64.b64encode(part.inline_data.data).decode('utf-8')
                                            
                                            await client_websocket.send(json.dumps({"audio": base64_audio}))
                                            
                                            # 在这里累积音频数据
                                            if not hasattr(session, 'audio_data'):
                                                session.audio_data = b''
                                            session.audio_data += part.inline_data.data
                                            
                                            print("音频已接收")

                                if response.server_content.turn_complete:
                                    print('\n<回合完成>')
                                    # 在这里转录累积的音频
                                    transcribed_text = transcribe_audio(session.audio_data)
                                    if transcribed_text:    
                                        await client_websocket.send(json.dumps({
                                            "text": transcribed_text
                                        }))
                                    # 清除累积的音频数据
                                    session.audio_data = b''
                        except websockets.exceptions.ConnectionClosedOK:
                            print("客户端连接正常关闭（接收）")
                            break  # 如果连接关闭，则退出循环
                        except Exception as e:
                            print(f"接收来自 Gemini 时出错: {e}")
                            break 

                except Exception as e:
                      print(f"接收来自 Gemini 时出错: {e}")
                finally:
                      print("Gemini 连接关闭（接收）")


            # 启动发送循环
            send_task = asyncio.create_task(send_to_gemini())
            # 将接收循环作为后台任务启动
            receive_task = asyncio.create_task(receive_from_gemini())
            await asyncio.gather(send_task, receive_task)


    except Exception as e:
        print(f"Gemini 会话中出错: {e}")
    finally:
        print("Gemini 会话关闭。")

首先，它接收来自客户端的配置数据，包括系统指令和来自客户端的响应方式。

然后，它与 Gemini API 建立连接并开始使用模型的 session。

send_to_gemini() 函数管理从客户端到 Gemini API 的消息流。它接收客户端发送的媒体块，将音频和图像数据打包到 Gemini API 消息格式中并发送。所有这些都是异步完成的，确保流畅的传输。

receive_from_gemini() 函数负责监听 Gemini API 的响应并将数据转发给客户端。我们使用一个 while 循环，直到连接终止或触发错误为止。一旦生成模型响应，我们处理音频数据并将其发送回客户端播放。同时，音频数据将被累积，一旦设置了 turn_complete 标志，音频数据将被发送到 Gemini 1.5 flash 模型进行转录。

def transcribe_audio(audio_data):
    """使用 Gemini 1.5 Flash 转录音频。"""
    try:
        # 确保我们有有效的音频数据
        if not audio_data:
            return "未接收到音频数据。"
            
        # 将 PCM 转换为 MP3
        mp3_audio_base64 = convert_pcm_to_mp3(audio_data)
        if not mp3_audio_base64:
            return "音频转换失败。"
            
        transcription_client = generative.GenerativeModel(model_name=TRANSCRIPTION_MODEL)
        
        prompt = """生成语音的转录文本。 
        请不要在回复中包含任何其他文本。 
        如果您听不到语音，请只说 '<无法识别>'。"""
        
        response = transcription_client.generate_content(
            [
                prompt,
                {
                    "mime_type": "audio/mp3", 
                    "data": base64.b64decode(mp3_audio_base64),
                }
            ]
        )
            
        return response.text

    except Exception as e:
        print(f"转录错误: {e}")
        return "转录失败。", None

def convert_pcm_to_mp3(pcm_data):
    """将 PCM 音频转换为 base64 编码的 MP3。"""
    try:
        # 首先在内存中创建 WAV
        wav_buffer = io.BytesIO()
        with wave.open(wav_buffer, 'wb') as wav_file:
            wav_file.setnchannels(1)  # 单声道
            wav_file.setsampwidth(2)  # 16 位
            wav_file.setframerate(24000)  # 24kHz
            wav_file.writeframes(pcm_data)
        
        wav_buffer.seek(0)
        audio_segment = AudioSegment.from_wav(wav_buffer)

        mp3_buffer = io.BytesIO()
        audio_segment.export(mp3_buffer, format="mp3", codec="libmp3lame")
        
        # 转换为 base64
        mp3_base64 = base64.b64encode(mp3_buffer.getvalue()).decode('utf-8')
        return mp3_base64
        
    except Exception as e:
        print(f"将 PCM 转换为 MP3 时出错: {e}")
        return None

这两个辅助函数 convert_pcm_to_mp3()，顾名思义，用于将模型响应音频转换为 MP3，因为 Gemini 1.5 音频输入不支持 PCM 格式。然后由 transcribe_audio() 函数调用，它将通过与传统的 google-generativeai API 交互生成语音的转录文本。

最后，主函数非常简单，它设置了WebSocket服务器以监听特定端口。对于每个新的连接，调用gemini_session_handler()来建立一个session，然后，由于正在等待asyncio.Future()，服务器将无限期运行。

客户端设计

现在，让我们转到用 HTML 和 JavaScript 编写的客户端代码。完整的代码可以从我的 GitHub 仓库下载。

转到 HTML 部分，我们有一个简单的布局，包括图像上传、绘图控件和一个显示图像的画布。同时还有一个转录文本区域，用于显示转录结果。

async function renderFileOnCanvas(file) {
      if (file.type.startsWith("image/")) {
         const reader = new FileReader();
            reader.onload = (event) => {
                 const img = new Image();
                    img.onload = () => {
                      canvas.width = img.width; // Update Canvas width
                      canvas.height = img.height; // Update Canvas height
                      context.drawImage(img, 0, 0, canvas.width, canvas.height);
                   };
                img.src = event.target.result;
             };
            reader.readAsDataURL(file);
        }

    }

在 JavaScript 部分，renderFileOnCanvas() 函数负责将选定的图像加载到画布上，它处理所有必要的步骤，以确保图像正确显示在画布内，并动态更新画布大小，以避免图像变形。

imageLoader.addEventListener('change', async (e) => {
          const file = e.target.files[0];
          if (file) {

                  await renderFileOnCanvas(file);
          }
     });

这个监听器触发 renderFileOnCanvas 函数，将加载的图像绘制到画布上。它还检查文件参数，如果存在，它将调用 renderFileOnCanvas 函数将选定的图像绘制到画布上。

    function captureImage() {
        const imageData = canvas.toDataURL("image/jpeg").split(",")[1].trim();
        currentFrameB64 = imageData;
    }

    window.addEventListener("load", async () => {
             //await startWebcam();

             setInterval(captureImage, 3000);
             await initializeAudioContext();
            connect();

        });

captureImage() 函数负责将画布的内容转换为 base64，以便可以与音频输入一起发送到服务器，设置为每 3 秒进行一次。

最后，startAudioInput() 和 stopAudioInput() 函数管理麦克风访问、捕获音频并通过 WebSocket 将其发送到后端。我们在之前的教程中已经展示过这些。receiveMessage() 函数管理所有传入消息，通过显示文本输出或播放音频输出来处理，如前面的教程所示。

function receiveMessage(event) {
            const messageData = JSON.parse(event.data);
            const response = new Response(messageData);

            if(response.text){
                displayMessage("GEMINI: " + response.text);
            }
            if(response.audioData){
              injestAudioChuckToPlay(response.audioData);
            }
        }