こちらはMOSHアドベントカレンダーの7日目です！
はじめに
こんにちは、MOSHの村井です。私たちは個人のサービスクリエイターの業務効率化を支援するべく、決済や予約といったワークフローのオートメーションやDXを可能にするプロダクトを日々開発しています。
2024年は（も？）AIが盛り上がった、というか一周回って落ち着いた年ということもあり、改めてAI技術を活用したサービスの進化に取り組んでいます。
今回は、今年にリリースした動画要約機能の実装プロセスや、その結果得られた学びについてお話ししたいと思います。
動画要約機能のプロジェクトの背景
昨今、個人のクリエイターの領域においても、デジタルコンテンツ、特に動画コンテンツを用いたサービス提供が急速に広がっています。
Zoomなどを用いたオンラインでのサービス提供が増えたことで、そのアーカイブ視聴や切り抜きの拡散といったニーズが広がったことが背景にあると思います。
無論動画コンテンツは優れたメディアですが、一方で視聴者にとっては膨大な情報量が障害となる場合もあります。
サービスのアーカイブ動画などは2時間ほどの規模になることもしょっちゅうであり、これをすべて通して見なくては効果が得られないというのは、加速している視聴者のライフサイクルにフィットしません。
結果として、興味のある情報だけを的確に把握したい、また、短時間で必要なコンテンツにアクセスしたい、というニーズは多くの視聴者から寄せられることとなりました。
このような課題に立ち向かうために、AIを活用した動画要約機能の開発を進めました。
このプロジェクトは、クリエイターにとっては、従来の編集作業にかかる時間を削減し、視聴者が動画を簡単に理解できるような新しい体験を提供することを目指しています。
AIを用いて的を得た動画の要約を実現し、視聴者が求める情報を瞬時に提供することで、視聴者は長時間の動画を視聴することなく、必要な情報に迅速にアクセスできるようにしたいと思っています。
また、こうした視聴者の体験の向上は、クリエイターのサービスの質の向上につながるため、結果的に視聴者の購入継続にもつながってくると考えています。
アーキテクチャ全体の説明
さっそく、動画要約機能の技術的な詳細に触れていきましょう。
システムアーキテクチャ
このプロジェクトのアーキテクチャは、以下の主要なコンポーネントで構成されています。
- 動画アップロード：ユーザーがMOSHのファイルストレージ（s3）に動画データをアップロードします。
- 音声認識：アップロードされた音声データはOpenAIのWhisperを使ってテキストに変換されます。
- 要約生成：テキスト化された内容はDifyを介してOpenAI等のLLMに送付され、自動要約されます。
- 結果の返却：生成された要約結果がユーザーに返却されます
シーケンス図
このシーケンス図は、動画データのアップロードからAIを用いた処理、最終的な生成結果の通知に至るまでの一連の流れを示しています。全体のプロセスはServerlessアーキテクチャに基づいており、イベント駆動型の設計になっている点が特徴です。
プロセスはユーザーが動画データをアップロードすることから始まります。
ユーザーが動画をアップロードすると、Elemental MediaConvertがその動画をエンコードします。
この際に、音声データを抽出しておきます。
さまざまなフォーマットの動画が入力されることを想定すると、特定のフォーマットにコンバートしたのちに、要約のプロセスを処理した方が入力データのバリエーションによる処理の分岐を考慮せずに済みます。
エンコードが完了すると、完了通知がEncoded Event Lambdaに送信されます。この通知をトリガーにして、次のステップ、すなわちAI処理を実行するGenerate Process Lambdaが呼び出されます。このように、各プロセスは特定のイベントに対して起動されるます。
次に、Generate Process LambdaはS3から事前に抽出した音声データをロードします。この時、音声データを適切に分割するために、pydubライブラリを利用しています。
音声データが準備できた後、AI音声認識モデルであるWhisperに対して音声の書き起こしを行うためのリクエストがOpenAIに送信されます。
OpenAIからの応答を受け取ることで、その書き起こし結果がテキストとして得られます。
続いて、得られたテキストデータを基に、Difyに要約をリクエストします。
Difyから得られた要約結果はGenerate Process Lambdaによってデータベースに保存されます。
ここまでが生成処理の役割になります。
その後、データベースへの保存をトリガーとして、生成完了の通知が送信され、最終的にユーザーに対して生成完了メッセージが送られます。
一連の流れはおおむね疎結合で、各プロセスが特定のイベントを契機に実行されるため、スケーラブルで柔軟なシステムになっています。
（と言えば聞こえはいいですが、トレーサビリティが低く、トラブルシューティングの難易度が高いなどの課題も多いです...）
実装の詳細
私たちが開発した動画要約機能には、AI技術が主要な役割を果たしています。
OpenAIのWhisperを利用して音声データをテキスト化し、その後の要約処理を行っているため、重要な要素技術と言えるでしょう。
Whisperの利用
Whisperは高精度の音声認識を提供し、使いやすさも兼ね備えています。
以下は、音声データをテキストに変換するための実装例です。（動作検証はしてません）
import io
from pydub import AudioSegment
from openai import OpenAI
class OpenaiService:
def __init__(self, openai_api_key: str):
self.__openai_api_key = openai_api_key
def transcribe(self, file: bytes) -> str:
client = OpenAI(api_key=self.__openai_api_key)
# 音声ファイルを分割し、各チャンクを文字起こししてSRTファイルとトランスクリプトを生成する
from pydub import AudioSegment
from pydub.utils import which
AudioSegment.converter = which("ffmpeg")
AudioSegment.ffprobe = which("ffprobe")
audio = AudioSegment.from_file(io.BytesIO(file), format="mp3")
length = len(audio)
chunk_length = 1000000 # 1,000,000ミリ秒ごとに分割
overlap_length = 10000 # 10,000ミリ秒のオーバーラップ
chunks = []
# チャンクを生成する際にオーバーラップを考慮
for i in range(0, length, chunk_length - overlap_length):
end = min(i + chunk_length, length)
chunk = audio[i:end]
chunks.append((chunk, i, end)) # チャンクと実際の開始/終了時間を保存
all_segments = []
transcript_text = ""
for i, (chunk, s, _e) in enumerate(chunks, start=1):
file_path = "temp_file_path"
chunk.export(file_path, format="mp3")
with open(file_path, "rb") as audio_file:
transcript = client.audio.transcriptions.create(
file=audio_file,
model="whisper-1",
response_format="verbose_json",
language="ja",
)
for segment in transcript.segments:
segment["start"] += s / 1000 # チャンクの実際の開始時間を加算
segment["end"] += s / 1000 # チャンクの実際の終了時間を加算
all_segments.append(segment)
transcript_text += transcript.text + "\n"
このプロセスでは、音声データが入力として受け取られ、Whisperモデルに渡されます。
そして、得られたテキストが要約処理に活用されます。
書き起こしの精度が、要約の精度に直結する実装になっている点は注意が必要です。
Difyの利用
次に、Difyを用いたテキストの要約についても見てみましょう。以下はその実装例です。
import json
import requests
class DifyService:
API_TIMEOUT_SECONDS = 300
def __init__(self, api_key: str):
self.__api_key = api_key
self.__base_url = "https://api.dify.ai/v1"
def complete_messages(self, inputs: str) -> Optional[str]:
response = self.__http.post(
url=f"{self.__base_url}/completion-messages",
headers=HTTPHeaders(
authorization=f"Bearer {self.__api_key}", content_type=MIMEType.APPLICATION_JSON
),
data=json.dumps(
{"inputs": {"query": inputs}, "response_mode": "blocking", "user": self.__api_user}
),
timeout=self.API_TIMEOUT_SECONDS,
)
if not response or isinstance(response, list):
return None
answer = response.get("answer") if response else None
if not answer:
return None
return answer
def run_workflows(self, inputs: str) -> Optional[Dict]:
response = requests.post(
url=f"{self.__base_url}/workflows/run",
headers={
"Authorization": f"Bearer {self.__api_key}",
"Content-Type": "application/json",
},
json={
"inputs": {"query": inputs},
"response_mode": "streaming",
"user": self.__api_user,
},
timeout=self.API_TIMEOUT_SECONDS,
stream=True,
)
if not response:
return None
final_response = None
for line in response.iter_lines():
if not line:
continue
decoded_line = line.decode()
if not decoded_line.startswith("data:"):
continue
event_data = json.loads(decoded_line[5:])
if event_data.get("event") == "node_finished":
final_response = event_data["data"]
if not final_response:
return None
outputs = final_response.get("outputs")
if not outputs:
return None
return outputs
Difyの仕様上、複数の処理を一つのワークフローにまとめる場合、run_workflowsを呼び出すようになっています。
なんてことのないAPI呼び出しですが、Difyを用いるメリットは大きく、LLMのプロンプトの変更がデプロイなしで可能です。
このようにすることで、AIエンジニアとWebエンジニアの役割を分け、AIエンジニアの作業にWebエンジニアの介在が不要となります。
Protocol buffersの利用
今回のプロジェクトは検証的側面が強く、正式に提供価値として管理運用していくか、不透明な部分もありました。
また、AI関連技術は進化が早く、移り変わりも激しいため、なるべく利用する箇所を広げたくありませんでした。
そこで、実装するコードはExperimentalModuleというモジュールに分離することにしました。
モジュールへのアクセスはインターフェースをProtocol buffersで定義しました。
こうすることで、例えばOpenAIがGeminiに変わったとしても、利用側では修正が不要な設計にしています。
以下はインタフェースの定義例です。
syntax = "proto3";
package ai;
import "google/protobuf/struct.proto";
message CreateVideoSummaryRequest {
string input = 10;
}
message CreateVideoSummaryResponse {
google.protobuf.Struct result = 10;
}
service AiClient {
CreateVideoSummary(CreateVideoSummaryRequest) returns (CreateVideoSummaryResponse) {};
}
Protocol buffersを利用することによって、データのやり取りが効率化され、新機能の追加や既存機能の修正が容易になります。
まとめ
今回の取り組みによって、AIを活用した動画要約機能は無事に実装されました。
この新機能はクリエイターさんからも好評をいただいており、この機能を利用したいがゆえにMOSHを導入いただいているケースもあるほどです。
まだまだ精度向上の余地もあるため、ユーザーさんからのフィードバックをもとに、要約の精度や使いやすさを改善し続け、さらなる機能向上に努めていきます。
MOSHは、「個人の才能をもっと世の中に発信したい」という思いから始まった事業です。
最初は文字、今は写真や動画、そしてAIと、個人が扱える手段は増えていくものの、その活用の方法は確立していません。
こうしたAIの利用方法についても、どんどん提案していきたいと思います。
さて、明日は村山さんです。乞うご期待！
Discussion