LLMと電話を組み合わせたパッケージvocodeの中身を勉強する①

LLMによる電話応答ができるパッケージであるvocodeの中身を勉強したいと思ったので、気になるpyファイルの中身がどのようになっているかをchatGPTに解説させました。

https://github.com/vocodedev/vocode-python

chat_gpt_agent.py

from typing import Optional
import openai
from langchain.prompts import (
    ChatPromptTemplate,
    MessagesPlaceholder,
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate,
)
from langchain.chains import ConversationChain
from langchain.chat_models import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from vocode import getenv

from vocode.turn_based.agent.base_agent import BaseAgent


class ChatGPTAgent(BaseAgent):
    def __init__(
        self,
        system_prompt: str,
        api_key: Optional[str] = None,
        initial_message: Optional[str] = None,
        model_name: str = "gpt-3.5-turbo",
        temperature: float = 0.7,
        max_tokens: int = 100,
    ):
        super().__init__(initial_message=initial_message)
        openai.api_key = getenv("OPENAI_API_KEY", api_key)
        if not openai.api_key:
            raise ValueError("OpenAI API key not provided")
        self.prompt = ChatPromptTemplate.from_messages(
            [
                SystemMessagePromptTemplate.from_template(system_prompt),
                MessagesPlaceholder(variable_name="history"),
                HumanMessagePromptTemplate.from_template("{input}"),
            ]
        )
        self.memory = ConversationBufferMemory(return_messages=True)
        if initial_message:
            self.memory.chat_memory.add_ai_message(initial_message)
        self.llm = ChatOpenAI(
            model_name=model_name,
            temperature=temperature,
            max_tokens=max_tokens,
        )
        self.conversation = ConversationChain(
            memory=self.memory, prompt=self.prompt, llm=self.llm
        )

    def respond(self, human_input: str):
        return self.conversation.predict(input=human_input)

このコードは、ChatGPTAgentという名前のクラスを実装しています。このクラスは、GPT-3.5-turboというOpenAIのモデルを使ってチャットボットの対話エージェントを作成します。クラスの構造を段階的に解説します。

必要なライブラリとモジュールをインポートします。これには、typing、openai、langchain（独自のライブラリ）、およびvocodeが含まれます。

ChatGPTAgentクラスはBaseAgentクラスを継承します。これにより、チャットボットに基本的な機能が提供されます。

__init__メソッドでは、いくつかのパラメータを受け取り、それらを使ってクラスのインスタンスを初期化します。主要なパラメータには、システムプロンプト、APIキー、初期メッセージ、モデル名、温度（テキスト生成のランダム性）、最大トークン数（生成されるテキストの長さ）があります。

インスタンス変数として、プロンプト、メモリ、チャットモデル（ここではGPT-3.5-turbo）、会話チェーンが作成されます。プロンプトは、システムメッセージとヒューマンメッセージから生成されます。メモリは、会話の履歴を保持するためのもので、チャットモデルはテキスト生成に使用されます。会話チェーンは、これらの要素を組み合わせて対話を実現します。

respondメソッドは、ユーザーからの入力を受け取り、会話チェーンのpredictメソッドを呼び出して、GPT-3.5-turboモデルからの応答を生成します。

このクラスを使って、APIキーといくつかの設定パラメータを指定してインスタンスを作成することができます。それから、respondメソッドを使って、ユーザーからの入力に対して適切な応答を生成することができます。

gpt4all_agent.py

import asyncio
from concurrent.futures import ThreadPoolExecutor, wait
import logging
import sys
from typing import Optional
from vocode.turn_based.agent.base_agent import BaseAgent

class StopThreadException(Exception):
    pass

class GPT4AllAgent(BaseAgent):
    SENTENCE_ENDINGS = [".", "!", "?"]

    DEFAULT_PROMPT_TEMPLATE = "{history}\nHuman: {human_input}\nAI:"

    def __init__(
        self,
        model_path: str,
        system_prompt: str,
        initial_message: Optional[str] = None,
        logger: logging.Logger = None,
    ):
        from pygpt4all.models.gpt4all_j import GPT4All_J
        
        super().__init__(initial_message)
        self.prompt_template = f"{system_prompt}\n\n{self.DEFAULT_PROMPT_TEMPLATE}"
        self.logger = logger or logging.getLogger(__name__)
        self.memory = [f"AI: {initial_message}"] if initial_message else []
        self.llm = GPT4All_J(model_path)
        self.thread_pool_executor = ThreadPoolExecutor(max_workers=1)

    def create_prompt(self, human_input):
        history = "\n".join(self.memory[-5:])
        return self.prompt_template.format(history=history, human_input=human_input)

    def get_memory_entry(self, human_input, response):
        return f"Human: {human_input}\nAI: {response}"

    def respond(
        self,
        human_input,
    ) -> str:
        self.logger.debug("LLM responding to human input")
        prompt = self.create_prompt(human_input)
        response_buffer = ""
        def new_text_callback(text):
            nonlocal response_buffer
            response_buffer += text
            if len(response_buffer) > len(prompt) and response_buffer.endswith("Human:"):
                response_buffer = response_buffer[:-len("Human:")]
                sys.exit()
        future = self.thread_pool_executor.submit(
            self.llm.generate,
            prompt,
            new_text_callback = new_text_callback,
        )
        wait([future], timeout=10)
        response = response_buffer[(len(prompt) + 1):]
        self.memory.append(self.get_memory_entry(human_input, response))
        self.logger.debug(f"LLM response: {response}")
        return response

    async def respond_async(self, human_input) -> str:
        prompt = self.create_prompt(human_input)
        response_buffer = ""
        def new_text_callback(text):
            nonlocal response_buffer
            response_buffer += text
            if len(response_buffer) > len(prompt) and response_buffer.endswith("Human:"):
                response_buffer = response_buffer[:-len("Human:")]
                raise StopThreadException("Stopping the thread")
        loop = asyncio.get_event_loop()
        try:
            await loop.run_in_executor(self.thread_pool_executor, lambda: self.llm.generate(prompt, new_text_callback = new_text_callback))
        except StopThreadException:
            pass
        response = response_buffer[(len(prompt) + 1):]
        self.memory.append(self.get_memory_entry(human_input, response))
        return response


if __name__ == "__main__":

    async def main():
        chat_responder = GPT4AllAgent(
            system_prompt="The AI is having a pleasant conversation about life.",
            model_path='/Users/ajayraj/Downloads/ggml-gpt4all-j-v1.3-groovy.bin',
        )
        while True:
            response = await chat_responder.respond_async(input("Human: "))
            print(f"AI: {response}")
    
    asyncio.run(main())

このコードは、GPT4AllAgentという名前のクラスを実装しています。このクラスは、GPT-4All_Jという独自の言語モデルを使ってチャットボットの対話エージェントを作成します。クラスの構造を段階的に解説します。

必要なライブラリとモジュールをインポートします。これには、asyncio、concurrent.futures、logging、sys、typing、およびvocodeが含まれます。

GPT4AllAgentクラスはBaseAgentクラスを継承します。これにより、チャットボットに基本的な機能が提供されます。

__init__メソッドでは、いくつかのパラメータを受け取り、それらを使ってクラスのインスタンスを初期化します。主要なパラメータには、モデルのパス、システムプロンプト、初期メッセージ、ロガーがあります。

インスタンス変数として、プロンプトテンプレート、ロガー、メモリ（会話履歴を保持するためのリスト）、言語モデル（GPT4All_J）、およびスレッドプールエグゼキュータ（並行タスク実行用）が作成されます。

create_promptメソッドは、ユーザーからの入力を受け取り、履歴と組み合わせてプロンプトを作成します。

get_memory_entryメソッドは、ユーザー入力と応答を受け取り、メモリエントリ（履歴用）を作成します。

respondメソッドは、ユーザーからの入力を受け取り、言語モデルを使って対話の応答を生成します。このメソッドでは、スレッドプールエグゼキュータを使って、タスクを並行して実行します。

respond_asyncメソッドは、respondメソッドと同様にユーザーからの入力を受け取り、言語モデルを使って対話の応答を生成しますが、非同期タスクとして実行されます。

最後に、if name == "main":ブロックでは、GPT4AllAgentのインスタンスを作成し、respond_asyncメソッドを使ってユーザーからの入力に対して適切な応答を生成します。これは、main()関数をasyncioを使って実行することで実現されます

microphone_input.py

from typing import Optional
import sounddevice as sd
import numpy as np
from pydub import AudioSegment
import io
import wave

from vocode.turn_based.input_device.base_input_device import BaseInputDevice


class MicrophoneInput(BaseInputDevice):
    DEFAULT_SAMPLING_RATE = 44100
    DEFAULT_CHUNK_SIZE = 2048

    def __init__(
        self,
        device_info: dict,
        sampling_rate: int = None,
        chunk_size: int = DEFAULT_CHUNK_SIZE,
    ):
        self.device_info = device_info
        self.sampling_rate = sampling_rate or (
            self.device_info.get("default_samplerate", self.DEFAULT_SAMPLING_RATE)
        )
        self.chunk_size = chunk_size
        self.buffer: Optional[io.BytesIO] = None
        self.wave_writer: Optional[wave.Wave_write] = None
        self.stream = sd.InputStream(
            dtype=np.int16,
            channels=1,
            samplerate=self.sampling_rate,
            blocksize=self.chunk_size,
            device=int(self.device_info["index"]),
            callback=self._stream_callback,
        )
        self.active = False

    @classmethod
    def from_default_device(cls, sampling_rate: int = None):
        return cls(sd.query_devices(kind="input"), sampling_rate)

    def _stream_callback(self, in_data: np.ndarray, *_args):
        if self.active:
            audio_bytes = in_data.tobytes()
            self.wave_writer.writeframes(audio_bytes)

    def create_buffer(self):
        in_memory_wav = io.BytesIO()
        wave_writer = wave.open(in_memory_wav, "wb")
        wave_writer.setnchannels(1)
        wave_writer.setsampwidth(2)
        wave_writer.setframerate(self.sampling_rate)
        return in_memory_wav, wave_writer

    def start_listening(self):
        self.buffer, self.wave_writer = self.create_buffer()
        self.active = True
        self.stream.start()

    def end_listening(self) -> AudioSegment:
        self.stream.stop()
        self.active = False
        self.buffer.seek(0)
        return AudioSegment.from_wav(self.buffer)

このコードは、MicrophoneInputという名前のクラスを実装しています。このクラスは、マイク入力を扱うための基本機能を提供します。具体的には、マイクからのオーディオデータを録音し、録音を終了した後にオーディオデータをAudioSegment形式で返します。クラスの構造を段階的に解説します。

必要なライブラリとモジュールをインポートします。これには、typing、sounddevice、numpy、pydub、io、およびwaveが含まれます。

MicrophoneInputクラスはBaseInputDeviceクラスを継承します。これにより、オーディオ入力デバイスに基本的な機能が提供されます。

__init__メソッドでは、いくつかのパラメータを受け取り、それらを使ってクラスのインスタンスを初期化します。主要なパラメータには、デバイス情報、サンプリングレート、チャンクサイズがあります。

インスタンス変数として、デバイス情報、サンプリングレート、チャンクサイズ、バッファ（オーディオデータを格納するためのもの）、WaveWriter（オーディオデータをWave形式で書き込むためのもの）、および音声入力ストリームが作成されます。音声入力ストリームのコールバック関数として_stream_callbackメソッドが設定されます。

from_default_deviceクラスメソッドは、デフォルトのマイクデバイスからMicrophoneInputインスタンスを作成します。

_stream_callbackメソッドは、音声ストリームから入力データを受け取り、録音がアクティブであれば、WaveWriterを使ってオーディオデータをバッファに書き込みます。

create_bufferメソッドは、オーディオデータを格納するためのバッファ（BytesIOオブジェクト）とWaveWriterオブジェクトを作成します。

start_listeningメソッドは、バッファとWaveWriterを初期化し、録音をアクティブにして音声ストリームを開始します。

end_listeningメソッドは、音声ストリームを停止し、録音を非アクティブにして、バッファからAudioSegment形式のオーディオデータを返します。

このクラスを使って、デフォルトのマイクデバイスからオーディオに接触

speaker_output.py

import sounddevice as sd
import numpy as np
from pydub import AudioSegment

from vocode.turn_based.output_device.base_output_device import BaseOutputDevice


class SpeakerOutput(BaseOutputDevice):
    DEFAULT_SAMPLING_RATE = 44100

    def __init__(
        self,
        device_info: dict,
        sampling_rate: int = None,
    ):
        self.device_info = device_info
        self.sampling_rate = sampling_rate or int(
            self.device_info.get("default_samplerate", self.DEFAULT_SAMPLING_RATE)
        )
        self.stream = sd.OutputStream(
            channels=1,
            samplerate=self.sampling_rate,
            dtype=np.int16,
            device=int(self.device_info["index"]),
        )
        self.stream.start()

    @classmethod
    def from_default_device(cls, sampling_rate: int = None):
        return cls(sd.query_devices(kind="output"), sampling_rate)

    def send_audio(self, audio_segment: AudioSegment):
        raw_data = audio_segment.raw_data
        if audio_segment.frame_rate != self.sampling_rate:
            raw_data = audio_segment.set_frame_rate(self.sampling_rate).raw_data
        self.stream.write(np.frombuffer(raw_data, dtype=np.int16))

    def terminate(self):
        self.stream.close()

このコードは、SpeakerOutputという名前のクラスを実装しています。このクラスは、スピーカー出力を扱うための基本機能を提供します。具体的には、オーディオデータをスピーカーに送信して再生します。クラスの構造を段階的に解説します。

必要なライブラリとモジュールをインポートします。これには、sounddevice、numpy、およびpydubが含まれます。

SpeakerOutputクラスはBaseOutputDeviceクラスを継承します。これにより、オーディオ出力デバイスに基本的な機能が提供されます。

__init__メソッドでは、いくつかのパラメータを受け取り、それらを使ってクラスのインスタンスを初期化します。主要なパラメータには、デバイス情報およびサンプリングレートがあります。

インスタンス変数として、デバイス情報、サンプリングレート、および音声出力ストリームが作成されます。音声出力ストリームは、指定されたデバイス情報とサンプリングレートに基づいて設定されます。

from_default_deviceクラスメソッドは、デフォルトのスピーカーデバイスからSpeakerOutputインスタンスを作成します。

send_audioメソッドは、AudioSegment形式のオーディオデータを受け取り、そのオーディオデータをスピーカーに送信して再生します。オーディオデータのサンプリングレートが現在のサンプリングレートと異なる場合、オーディオデータのサンプリングレートを変更します。

terminateメソッドは、音声出力ストリームを終了します。

このクラスを使って、デフォルトのスピーカーデバイスにオーディオデータを送信して再生することができます。

turn_based_conversation.py

import logging
from vocode.turn_based.agent.base_agent import BaseAgent
from vocode.turn_based.input_device.base_input_device import (
    BaseInputDevice,
)
from vocode.turn_based.output_device.base_output_device import BaseOutputDevice
from vocode.turn_based.synthesizer.base_synthesizer import BaseSynthesizer
from vocode.turn_based.transcriber.base_transcriber import BaseTranscriber


class TurnBasedConversation:
    def __init__(
        self,
        input_device: BaseInputDevice,
        transcriber: BaseTranscriber,
        agent: BaseAgent,
        synthesizer: BaseSynthesizer,
        output_device: BaseOutputDevice,
        logger: logging.Logger = None,
    ):
        self.input_device = input_device
        self.transcriber = transcriber
        self.agent = agent
        self.synthesizer = synthesizer
        self.output_device = output_device
        self.maybe_play_initial_message()
        self.logger = logger or logging.getLogger(__name__)

    def maybe_play_initial_message(self):
        if self.agent.initial_message:
            self.output_device.send_audio(
                self.synthesizer.synthesize(self.agent.initial_message)
            )

    def start_speech(self):
        self.input_device.start_listening()

    def end_speech_and_respond(self):
        human_input = self.transcriber.transcribe(self.input_device.end_listening())
        self.logger.info(f"Transcription: {human_input}")
        agent_response = self.agent.respond(human_input)
        self.logger.info(f"Agent response: {agent_response}")
        self.output_device.send_audio(self.synthesizer.synthesize(agent_response))

このコードは、TurnBasedConversationクラスを実装しています。このクラスは、順番に話すタイプの会話を実現するためのものです。具体的には、音声入力デバイスからの入力を受け取り、それをテキストに変換し、エージェントが応答を生成し、音声に変換して音声出力デバイスに送信します。クラスの構造を段階的に解説します。

必要なモジュールをインポートします。これには、基本のエージェント、入力デバイス、出力デバイス、シンセサイザー、およびトランスクライバーが含まれます。

__init__メソッドでは、いくつかのパラメータを受け取り、それらを使ってクラスのインスタンスを初期化します。主要なパラメータには、入力デバイス、トランスクライバー、エージェント、シンセサイザー、および出力デバイスがあります。

初期メッセージが設定されている場合、それを再生します。

start_speechメソッドは、音声入力デバイスがリスニングを開始するよう指示します。

end_speech_and_respondメソッドは、音声入力デバイスのリスニングを終了し、トランスクライバーを使用して音声をテキストに変換します。その後、エージェントがテキスト入力に対して応答を生成し、シンセサイザーを使用して応答を音声に変換し、音声出力デバイスに送信します。

このクラスを使用すると、音声入力と音声出力デバイスを使って、エージェントとの自然な対話を実現することができます。