1. はじめに
Alexaって便利ですよね。
もし、Alexaのように話しかけて動くPCがあればかっこいいと思いませんか。
（そういうサービスがすでにあるかどうかは知りません。）
少なくともWindowsにはSiriのような機能は無いと思っています。
…無いなら、作ればいいですかね。
2. 何ができるの
① PC起動と同時に作成したAssistantBotが裏で起動
② F9キーを押すとマイクが起動して録音開始
③ 話しかけた内容に対して、返答が返ってくる
○○を検索して：検索ウィンドウが開く
上記以外：ChatGPTの回答がスピーカーから出力
④ botは常駐/タスクトレイから終了させることも可能
3. 全体図
以下が最終的な全体の流れです
① ローカルPCからマイクで入力された音声ファイルをS3に保存
② S3の保存をトリガーにLambdaが音声ファイルを取得
③ 音声ファイルをTranscribeでテキスト化
④ テキスト化された文章をOpenAIに投げて回答を作成
⑤ 回答結果をPollyで音声ファイルに変換
⑥ 回答の音声ファイルをS3に保存
⑦ SQSに処理が完了したことを通知
⑧ ローカルPCがS3から回答結果の音声ファイルを取得してスピーカーで出力
4. 著者の環境
OS：Windows11
Python：3.11.9
5. 使用したAWSのサービス（スキップ可能？）
AWSで使用したサービスをざっくりと説明します。（ざっくり過ぎて説明ではない）
詳しくは公式サイトを参考にしてください。
5-1. Amazon S3
Amazon S3（Simple Storage Service）は、AWSが提供するオブジェクトストレージサービス。安い。使いやすい。
5-2. AWS Lambda
AWS Lambdaは、サーバーレスコンピューティングサービス。
コードさえ書ければ動かせちゃう。便利。
5-3. Amazon Transcribe
Amazon Transcribeは、機械学習を利用した音声認識サービス。
音声をテキストに変換してくれる。
5-4. Amazon Polly
Amazon Polly は、文章をリアルな音声に変換するサービス。
テキストを音声に変換してくれる。
5-5. Amazon SQS
Amazon SQS（Simple Queue Service）は、システムやプロセス間でデータを非同期でやり取りするメッセージキューサービス
6. 具体的なコード内容
具体的なコードの中身というよりは、各ファイルが何をしているのかが重要だと思ったので、コードは気になった方だけどうぞ。
長くなるので、特にここでは具体的なコードの説明はしない。
（本当はしたい。どっかで別で出そうかな…）
6-1. ローカルPC側
PCのマイクとスピーカーを使用する仕様上、録音、出力はPC側で処理する必要があり、クラウドと混合する形になりました。（使用と仕様は狙いました。）
ローカルのフォルダ構成
voice-assistant/
├── .env
├── icon.png # トレイアイコン用画像
├── pipeline_runner.py
├── tray_app.py
├── dist/
│ ├── tray_app.exe
│ └── （その他ビルド生成物）
└── build/
.env
AWSの認証情報やS3/SQSの設定値を記載した環境設定ファイル
コード
AWS_ACCESS_KEY_ID=
AWS_SECRET_ACCESS_KEY=
AWS_REGION=
S3_BUCKET=
S3_INPUT_KEY=
S3_OUTPUT_KEY=
SQS_QUEUE_URL=
OPENAI_API_KEY =
pipline_runner.py
F9キー録音〜SQS受信・再生までの実装
コード
import keyboard
import sounddevice as sd
import soundfile as sf
import time
import os
import boto3
import pygame
import json
import webbrowser
from dotenv import load_dotenv
load_dotenv()
s3 = boto3.client(
's3',
region_name = os.getenv("AWS_REGION"),
aws_access_key_id = os.getenv("AWS_ACCESS_KEY_ID"),
aws_secret_access_key = os.getenv("AWS_SECRET_ACCESS_KEY")
)
sqs = boto3.client(
'sqs',
region_name=os.getenv("AWS_REGION"),
aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY")
)
DURATION = 4
SAMPLE_RATE = 44100
# PCマイクから音声を録音し、WAVファイルとして保存
def record_audio(filename="input.wav"):
recording = sd.rec(int(DURATION * SAMPLE_RATE), samplerate=SAMPLE_RATE, channels=1)
sd.wait()
sf.write(filename, recording, SAMPLE_RATE)
# 録音したWAVファイルをAWS S3にアップロード
def upload_to_s3():
s3.upload_file("input.wav", os.getenv("S3_BUCKET"), os.getenv("S3_INPUT_KEY"))
# クラウド処理の完了をSQS通知で待機・取得
def wait_for_sqs_message(queue_url, total_timeout=60):
start = time.time()
while time.time() - start < total_timeout:
response = sqs.receive_message(
QueueUrl=queue_url,
WaitTimeSeconds=20,
MaxNumberOfMessages=1
)
messages = response.get('Messages', [])
if messages:
body = json.loads(messages[0]['Body'])
receipt_handle = messages[0]['ReceiptHandle']
sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=receipt_handle)
return body
return None
# 取得した音声ファイル（MP3）をPCスピーカーで再生
def play_audio(file_path="output.mp3"):
try:
pygame.mixer.init()
pygame.mixer.music.load(file_path)
pygame.mixer.music.play()
while pygame.mixer.music.get_busy():
time.sleep(0.1)
pygame.mixer.music.stop()
pygame.mixer.quit()
except Exception as e:
print("再生エラー")
# 使用済み音声ファイルを削除
def cleanup():
for f in ["input.wav", "output.mp3"]:
if os.path.exists(f):
os.remove(f)
def run_pipeline():
while not keyboard.is_pressed("F9"):
time.sleep(0.1)
while keyboard.is_pressed("F9"):
time.sleep(0.1)
record_audio()
upload_to_s3()
sqs_message = wait_for_sqs_message(os.getenv("SQS_QUEUE_URL"))
print(sqs_message)
if sqs_message:
message_type = sqs_message.get("type")
if message_type == "search": # 検索してくださいの処理
keyword = sqs_message.get("keyword", "").strip()
print(keyword)
webbrowser.open(f"https://www.google.com/search?q={keyword}")
elif message_type == "audio": # 上記以外の処理
bucket = sqs_message["bucket"]
key = sqs_message["key"]
s3.download_file(bucket, key, "output.mp3")
play_audio()
cleanup()
time.sleep(1)
tray_app.py
トレイアイコン実装＆バックグラウンド常駐起動
コード
import threading
import time
from pystray import Icon, Menu, MenuItem
from PIL import Image
from pipeline_runner import run_pipeline
running = True
processing = False
def main_loop():
global processing
while running:
if not processing:
processing = True
run_pipeline()
processing = False
time.sleep(1)
def on_quit(icon, item):
global running
running = False
icon.stop()
def setup_tray():
icon_path = "icon.png"
image = Image.open(icon_path)
menu = Menu(MenuItem("終了", on_quit))
icon = Icon("VoiceAssistant", image, "アシスタント起動中", menu)
thread = threading.Thread(target=main_loop, daemon=True)
thread.start()
icon.run()
if __name__ == "__main__":
setup_tray()
dist/tray_app.exe
tray_app.py を PyInstaller でビルドした実行形式ファイル。
Windows環境でダブルクリックまたはスタートアップ自動起動に設定して常駐運用できる。
build/
PyInstaller がビルド時に使用する一時ファイル群
6-2. AWS Lambda側
① S3から録音されたファイルを取得
② Transcribeで音声ファイルを文字起こし
③ 録音の文字起こし結果をChatGPTのAPIに投げ、回答生成
④ 生成結果を取得してPollyで音声ファイルに変換
⑤ S3に保存してSQSに保存完了を通知
コード
import boto3
import os
import time
import json
import urllib.request
import openai
s3 = boto3.client("s3")
transcribe = boto3.client("transcribe")
polly = boto3.client("polly")
sqs = boto3.client("sqs")
openai.api_key = os.environ["OPENAI_API_KEY"]
# ChatGPTから返答を取得
def get_chatgpt_response(prompt: str) -> str:
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "あなたは親しみやすいアシスタントです。"},
{"role": "user", "content": prompt}
],
max_tokens=10,
temperature=0.7
)
return response.choices[0].message['content'].strip()
# SQS通知送信
def notify_via_sqs(bucket, message: dict):
sqs.send_message(
QueueUrl=os.environ["SQS_QUEUE_URL"],
MessageBody=json.dumps(message)
)
def handler(event, context):
bucket = event['Records'][0]['s3']['bucket']['name']
key = event['Records'][0]['s3']['object']['key']
input_uri = f"s3://{bucket}/{key}"
job_name = f"transcribe-{int(time.time())}"
# 音声認識（文字起こし）
transcribe.start_transcription_job(
TranscriptionJobName = job_name,
Media={'MediaFileUri': input_uri},
MediaFormat='wav',
LanguageCode = "ja-JP",
)
while True:
result = transcribe.get_transcription_job(TranscriptionJobName=job_name)
status = result['TranscriptionJob']['TranscriptionJobStatus']
if status in ['COMPLETED', 'FAILED']:
break
time.sleep(2)
transcript_url = result['TranscriptionJob']['Transcript']['TranscriptFileUri']
response = urllib.request.urlopen(transcript_url)
transcript_text = json.loads(response.read())['results']['transcripts'][0]['transcript']
# 分岐処理（検索 or ChatGPT返答）
if "を検索して" in transcript_text:
keyword = transcript_text.split("を検索して")[0].strip()
if keyword:
notify_via_sqs(bucket, {"type": "search", "keyword": keyword})
else:
return {'statusCode': 200, 'body': json.dumps('キーワードなし')}
else:
answer = get_chatgpt_response(transcript_text)
polly_response = polly.synthesize_speech(
Text=answer,
OutputFormat="mp3",
VoiceId="Mizuki",
LanguageCode="ja-JP"
)
audio_key = "output/output.mp3"
s3.put_object(
Body=polly_response['AudioStream'].read(),
Bucket=bucket,
Key=audio_key
)
notify_via_sqs(bucket, {"type": "audio", "bucket": bucket, "key": audio_key})
return {'statusCode': 200, 'body': json.dumps('完了')}
7. つまずいたところ
AWS LambdaでOpenAIのライブラリを使う
openaiのライブラリを使用するためにレイヤーを追加して動かしたらエラーが発生。
エラー内容：No module named 'jiter.jiter'
いろいろ漁りましたが、結局解決したのはバージョンしかなかった。。。
直接解決に使用した参考サイトを添付しておきます
非同期処理による弊害
ローカルPCから回答結果を出力する関係上、クラウドの処理の実行完了を待たなくてはならない。
回答結果がS3に保存される前にローカルがS3から回答を持ってこようとするため、更新前の回答結果（ひとつ前の回答結果）を出力してしまう。
ChatGPTにどうするかを聞いてみた結果
改善案：S3オブジェクトのタイムスタンプ確認
んー、さすがにめんどくさいなあ。他にないですか
改善案：Lambda完了通知をSQSやSNSで受け取る
お！ええやん！
ホットワード起動
本当はSiriやAlexaのように何か特定のワード（Alexa：「Alexa!」のような）を言って動かそうと思ったが、Windowsでそれをやるのがめんどくさそうだったので一旦断念。。。
いつかやりたい
8. おわりに
ちょうどAWSの資格勉強中で、多くのサービスを触ることができたことが勉強になった。