LLMと電話を組み合わせたパッケージvocodeの中身を勉強する①
LLMによる電話応答ができるパッケージであるvocodeの中身を勉強したいと思ったので、気になるpyファイルの中身がどのようになっているかをchatGPTに解説させました。
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メソッドは、音声入力デバイスのリスニングを終了し、トランスクライバーを使用して音声をテキストに変換します。その後、エージェントがテキスト入力に対して応答を生成し、シンセサイザーを使用して応答を音声に変換し、音声出力デバイスに送信します。
このクラスを使用すると、音声入力と音声出力デバイスを使って、エージェントとの自然な対話を実現することができます。
Discussion