概要
ChatGPTで基本的な会話をするとともに、DALL-Eで絵を描き、LINEに連携する
ベースにするプロジェクト
django
create a app
console
python manage.py startapp line_qa_with_gpt_and_dalle
.env.example
.env.example
SECRET_KEY=
DEBUG=True
DB_ENGINE=django.db.backends.mysql
DB_NAME=gptplayground_db
DB_USER=python
DB_PASSWORD=
OPENAI_API_KEY=
GEMINI_API_KEY=
settings.py
config/settings.py
:
+ import os
from pathlib import Path
- import environ
+ from dotenv import load_dotenv
:
- # read at .env
- env = environ.Env(DEBUG=(bool, False))
- environ.Env.read_env(Path(BASE_DIR, ".env"))
- DEBUG = env("DEBUG") # read DEBUG at .env
- SECRET_KEY = env("SECRET_KEY") # read SECRET_KEY at .env
+ # .env ファイルを読み込む
+ load_dotenv()
+
+ DEBUG = os.getenv("DEBUG") # read DEBUG at .env
+ SECRET_KEY = os.getenv("SECRET_KEY") # read SECRET_KEY at .env
:
INSTALLED_APPS = [
:
"retrieval_qa_with_source",
+ "line_qa_with_gpt_and_dalle",
]
:
DATABASES = {
"default": {
- "ENGINE": env("DB_ENGINE"),
- "NAME": env("DB_NAME"),
- "USER": env("DB_USER"),
- "PASSWORD": env("DB_PASSWORD"),
+ "ENGINE": os.getenv("DB_ENGINE"),
+ "NAME": os.getenv("DB_NAME"),
+ "USER": os.getenv("DB_USER"),
+ "PASSWORD": os.getenv("DB_PASSWORD"),
"HOST": "localhost",
"PORT": "3306",
}
:
-LANGUAGE_CODE = 'en-us'
+LANGUAGE_CODE = 'ja'
-TIME_ZONE = 'UTC'
+TIME_ZONE = 'Asia/Tokyo'
:
STATIC_ROOT = BASE_DIR / "static"
STATIC_URL = "static/"
+MEDIA_ROOT = BASE_DIR / "media"
+MEDIA_URL = "media/"
config/urls.py
config/urls.py
+from django.conf.urls.static import static
+from config import settings
:
urlpatterns = [
path("retrieval_qa_with_source/", include("retrieval_qa_with_source.urls")),
+ path("line_qa_with_gpt_and_dalle/", include("line_qa_with_gpt_and_dalle.urls")),
path("admin/", admin.site.urls),
] + static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT) ←これも追加
app/urls.py
line_qa_with_gpt_and_dalle/urls.py(new)
from django.urls import path
from . import views
app_name = "line_qa_with_gpt"
urlpatterns = [
path("", views.HomeView.as_view(), name="home"),
]
models.py
line_qa_with_gpt_and_dalle/models.py
from django.contrib.auth.models import User
from django.db import models
class ChatLogsWithLine(models.Model):
user = models.ForeignKey(User, on_delete=models.CASCADE)
role = models.CharField(max_length=255)
content = models.TextField()
file_path = models.CharField(max_length=255, null=True)
invisible = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
forms.py
line_qa_with_gpt_and_dalle/forms.py(new)
from django import forms
class UserTextForm(forms.Form):
question = forms.CharField(widget=forms.Textarea)
def __init__(self, *args, **kwargs):
for field in self.base_fields.values():
field.widget.attrs["class"] = "form-control"
field.widget.attrs["rows"] = 3
super().__init__(*args, **kwargs)
templatetags
テンプレートで独自に関数を使うならテンプレートフィルタ
line_qa_with_gpt_and_dalle/templatetags/__init__.py(new)
line_qa_with_gpt_and_dalle/templatetags/split_ext.py(new)
from os.path import splitext
from django import template
register = template.Library()
@register.filter
def split_ext(path):
_, ext = splitext(path)
return ext[1:] # Remove the leading dot ('.jpg' -> 'jpg')
template
line_qa_with_gpt_and_dalle/templates/line_qa_with_gpt_and_dalle/base.html(new)
{% load static %}
<!DOCTYPE html>
<html lang="ja">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<meta http-equiv="X-UA-Compatible" content="ie=edge">
<title>ChatGPT-playground</title>
<!-- bootstrap and css -->
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.0.2/dist/css/bootstrap.min.css" rel="stylesheet" integrity="sha384-EVSTQN3/azprG1Anm3QDgpJLIm9Nao0Yz1ztcQTwFspd3yD65VohhpuuCOmLASjC" crossorigin="anonymous">
<link rel="stylesheet" href="{% static 'chat/css/home.css' %}">
<!-- font -->
<link href="https://fonts.googleapis.com/css?family=Sawarabi+Gothic" rel="stylesheet">
<!-- fontawesome -->
<link href="https://use.fontawesome.com/releases/v5.6.1/css/all.css" rel="stylesheet">
<!-- favicon -->
<link rel="shortcut icon" href="{% static 'chat/c_g.ico' %}">
</head>
<body>
<!-- nav -->
<h1></h1>
<div id="main">
{% block content %}{% endblock %}
</div>
<footer>
<p>© 2019 henojiya. / <a href="https://github.com/duri0214" target="_blank">github portfolio</a></p>
</footer>
</body>
</html>
line_qa_with_gpt_and_dalle/templates/line_qa_with_gpt_and_dalle/home.html(new)
{% extends 'line_qa_with_gpt_and_dalle/base.html' %}
{% load static %}
{% load split_ext %}
{% block content %}
<div class="container">
<div class="jumbotron">
<h1 class="display-4">LINE QA with ChatGPT4 and Dall-e-3</h1>
<p class="lead">The system that forms the basis of work</p>
<hr class="my-4">
<p>You can talk with ChatGPT4 and Dall-e-3 using LINE.</p>
</div>
{% for chat_log in chat_logs %}
{% if not chat_log.invisible %}
<div class="card mb-3">
<div class="card-body">
<h5 class="card-title">{{ chat_log.role }}</h5>
<h6 class="card-subtitle mb-2 text-muted">file: {{ chat_log.file_path }}</h6>
<p class="card-text">{{ chat_log.content }}</p>
{% if chat_log.file_path|split_ext == "jpg" %}
<img src="{{ chat_log.file_path }}" class="img-fluid" alt="Responsive image">
{% elif chat_log.file_path|split_ext == "mp3" %}
<audio controls>
<source src="{{ chat_log.file_path }}" type="audio/mpeg">
Your browser does not support the audio element.
</audio>
{% endif %}
<div>
<a href="#" class="card-link">Card link</a>
<a href="#" class="card-link">Another link</a>
</div>
</div>
</div>
{% endif %}
{% endfor %}
<form action="{% url 'line_qa_with_gpt:home' %}" method="POST">
{{ form }}
{% csrf_token %}
<input class="mt-3" type="submit" value="送信">
</form>
</div>
<script type="text/javascript">
window.scrollTo(0, document.body.scrollHeight);
</script>
{% endblock %}
views.py
コメントアウトしている部分は、その下のセクションで説明するドメインを使って実行するものを単体テスト的に置いた
line_qa_with_gpt_and_dalle/views.py
import json
from django.contrib.auth.models import User
from django.http import HttpResponse
from django.urls import reverse_lazy
from django.views import View
from django.views.decorators.csrf import csrf_exempt
from django.views.generic import FormView
from dotenv import load_dotenv
from line_qa_with_gpt_and_dalle.domain.usecase.llm_service_use_cases import (
GeminiUseCase,
OpenAIGptUseCase,
OpenAIDalleUseCase,
OpenAITextToSpeechUseCase,
OpenAISpeechToTextUseCase,
UseCase,
)
from line_qa_with_gpt_and_dalle.forms import UserTextForm
from line_qa_with_gpt_and_dalle.models import ChatLogsWithLine
# .env ファイルを読み込む
load_dotenv()
class HomeView(FormView):
template_name = "line_qa_with_gpt_and_dalle/home.html"
form_class = UserTextForm
success_url = reverse_lazy("line_qa_with_gpt:home")
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
login_user = User.objects.get(pk=1) # TODO: request.user.id
context["chat_logs"] = ChatLogsWithLine.objects.filter(
user=login_user
).order_by("created_at")
return context
def form_valid(self, form):
form_data = form.cleaned_data
login_user = User.objects.get(pk=1) # TODO: request.user.id
use_case_type = "OpenAISpeechToText" # TODO: ドロップダウンでモードを決める？
use_case: UseCase | None = None
content: str | None = form_data["question"]
if use_case_type == "Gemini":
use_case = GeminiUseCase()
content = form_data["question"]
elif use_case_type == "OpenAIGpt":
# Questionは何を入れてもいい（処理されない）
use_case = OpenAIGptUseCase()
content = form_data["question"]
elif use_case_type == "OpenAIDalle":
use_case = OpenAIDalleUseCase()
content = form_data["question"]
elif use_case_type == "OpenAITextToSpeech":
use_case = OpenAITextToSpeechUseCase()
content = form_data["question"]
elif use_case_type == "OpenAISpeechToText":
# Questionは何を入れてもいい（処理されない）
use_case = OpenAISpeechToTextUseCase()
content = None
use_case.execute(user=login_user, content=content)
return super().form_valid(form)
@csrf_exempt
class LineWebHookView(View):
@staticmethod
def post(request, *args, **kwargs):
"""ラインの友達追加時に呼び出され、ラインのIDを登録する"""
request_json = json.loads(request.body.decode("utf-8"))
events = request_json["events"]
# If you run the validation from the `LINE DEVELOPERS` screen, `events` will be returned as `[]`
if events:
line_user_id = events[0]["source"]["userId"]
# webhook connection check at fixed id 'dead...beef'
if line_user_id != "Udeadbeefdeadbeefdeadbeefdeadbeef":
# follow | unblock
if events[0]["type"] == "follow":
print("ここにきたらdbに追加")
# LinePush.objects.create(line_user_id)
# block
if events[0]["type"] == "unfollow":
print("ここにきたらdbから削除")
# LinePush.objects.filter(line_user_id).delete()
return HttpResponse(status=200)
ドメイン
repository
line_qa_with_gpt_and_dalle/domain/repository/chatlog.py(new)
from line_qa_with_gpt_and_dalle.domain.valueobject.chat import MyChatCompletionMessage
from line_qa_with_gpt_and_dalle.models import ChatLogsWithLine
class ChatLogRepository:
def __init__(self):
pass
@staticmethod
def find_chatlog_by_id(pk: int) -> list[ChatLogsWithLine]:
return ChatLogsWithLine.objects.get(pk=pk)
@staticmethod
def find_chatlog_by_user_id(user_id: int) -> list[ChatLogsWithLine]:
return ChatLogsWithLine.objects.filter(user_id=user_id)
@staticmethod
def insert(my_chat_completion_message: MyChatCompletionMessage):
ChatLogsWithLine.objects.create(
user=my_chat_completion_message.user,
role=my_chat_completion_message.role,
content=my_chat_completion_message.content,
file_path=my_chat_completion_message.file_path,
invisible=my_chat_completion_message.invisible,
)
@staticmethod
def bulk_insert(my_chat_completion_message_list: list[MyChatCompletionMessage]):
ChatLogsWithLine.objects.bulk_create(
[x.to_entity() for x in my_chat_completion_message_list]
)
@staticmethod
def upsert(my_chat_completion_message: MyChatCompletionMessage):
ChatLogsWithLine.objects.update_or_create(
id=my_chat_completion_message.id,
defaults={
"user": my_chat_completion_message.user,
"role": my_chat_completion_message.role,
"content": my_chat_completion_message.content,
"file_path": my_chat_completion_message.file_path,
"invisible": my_chat_completion_message.invisible,
},
)
service
line_qa_with_gpt_and_dalle/domain/service/llm.py(new)
import os
import secrets
from abc import ABC, abstractmethod
from io import BytesIO
from pathlib import Path
import requests.exceptions
from PIL import Image
from django.contrib.auth.models import User
from google import generativeai
from google.generativeai.types import GenerateContentResponse
from openai import OpenAI
from openai.types.chat import (
ChatCompletion,
)
from config.settings import MEDIA_ROOT
from line_qa_with_gpt_and_dalle.domain.repository.chatlog import (
ChatLogRepository,
)
from line_qa_with_gpt_and_dalle.domain.valueobject.chat import MyChatCompletionMessage
from line_qa_with_gpt_and_dalle.domain.valueobject.gender import Gender
def get_stored_chat_history(
user_id: int, chatlog_repository: ChatLogRepository
) -> list[MyChatCompletionMessage]:
chatlog_list = chatlog_repository.find_chatlog_by_user_id(user_id)
history = [
MyChatCompletionMessage(
pk=chatlog.pk,
user=chatlog.user,
role=chatlog.role,
content=chatlog.content,
invisible=False,
file_path=chatlog.file_path,
)
for chatlog in chatlog_list
]
return history
def get_prompt(gender: Gender) -> str:
return f"""
あなたはなぞなぞコーナーの担当者です。
#制約条件
- 「なぞなぞスタート」と言われたら質問に移る前に、あいさつをします
- 質問1のあとに質問2を行う。質問2が終わったら、感想を述べるとともに「本日はなぞなぞにご参加いただき、ありがとうございました。」と言って終わりましょう。判定結果は出力してはいけません
- 質問1は「論理的思考力」評価します
- 質問2は「洞察力」を評価します
- scoreが70を超えたら、judgeが「合格」になる
- {gender.name} の口調で会話を行う
- 「評価結果をjsonで出力してください」と入力されたら、判定結果例のように判定結果を出力する
#質問1
- はじめは4本足、途中から2本足、最後は3本足。それは何でしょう？
#質問2
- 私は黒い服を着て、赤い手袋を持っている。夜には立っているが、朝になると寝る。何でしょう？
#判定結果例
[{{"skill": "論理的思考力", "score": 50, "judge": "不合格"}},{{"skill": "洞察力", "score": 96, "judge": "合格"}}]
"""
def create_initial_prompt(user: User, gender: Gender) -> list[MyChatCompletionMessage]:
history = [
MyChatCompletionMessage(
user=user,
role="system",
content=get_prompt(gender),
invisible=True,
),
MyChatCompletionMessage(
user=user,
role="user",
content="なぞなぞスタート",
invisible=False,
),
]
return history
class LLMService(ABC):
def __init__(self):
self.chatlog_repository = ChatLogRepository()
@abstractmethod
def generate(self, **kwargs):
pass
@abstractmethod
def post_to_gpt(self, **kwargs):
pass
@abstractmethod
def save(self, **kwargs):
pass
class GeminiService(LLMService):
def __init__(self):
super().__init__()
def generate(
self, my_chat_completion_message: MyChatCompletionMessage, gender: str
) -> list[MyChatCompletionMessage]:
if my_chat_completion_message.content is None:
raise Exception("content is None")
chat_history = get_stored_chat_history(
user_id=my_chat_completion_message.user.pk,
chatlog_repository=self.chatlog_repository,
)
chat_history.append(
self.save(
MyChatCompletionMessage(
user=my_chat_completion_message.user,
role=my_chat_completion_message.role,
content=my_chat_completion_message.content,
invisible=False,
)
)
)
# TODO: Gemini用のMyChatCompletionMessageに詰め込みたい
# https://ai.google.dev/gemini-api/docs/get-started/tutorial?lang=python&hl=ja
response = self.post_to_gpt(chat_history)
latest_assistant = MyChatCompletionMessage(
user=my_chat_completion_message.user,
role="assistant",
content=response.text,
invisible=False,
)
chat_history.append(self.save(latest_assistant))
return chat_history
def post_to_gpt(
self, chat_history: list[MyChatCompletionMessage]
) -> GenerateContentResponse:
generativeai.configure(api_key=os.getenv("GEMINI_API_KEY"))
model = generativeai.GenerativeModel("gemini-1.5-flash")
# TODO: 「会話」にしたいね
response = model.generate_content(chat_history[-1].content)
return response
def save(
self, messages: MyChatCompletionMessage | list[MyChatCompletionMessage]
) -> MyChatCompletionMessage | list[MyChatCompletionMessage]:
if isinstance(messages, list):
self.chatlog_repository.bulk_insert(messages)
elif isinstance(messages, MyChatCompletionMessage):
self.chatlog_repository.insert(messages)
else:
raise ValueError(
f"Unexpected type {type(messages)}. Expected MyChatCompletionMessage or list[MyChatCompletionMessage]."
)
return messages
class OpenAIGptService(LLMService):
def __init__(self):
super().__init__()
self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def generate(
self, my_chat_completion_message: MyChatCompletionMessage, gender: str
) -> list[MyChatCompletionMessage]:
if my_chat_completion_message.content is None:
raise Exception("content is None")
chat_history = get_stored_chat_history(
user_id=my_chat_completion_message.user.pk,
chatlog_repository=self.chatlog_repository,
)
if not chat_history:
chat_history = create_initial_prompt(
user=my_chat_completion_message.user, gender=Gender(gender)
)
self.save(chat_history)
# 初回はユーザのボタン押下などのトリガーで「プロンプト」と「なぞなぞスタート」の2行がinsertされる
# 会話が始まっているならユーザの入力したチャットをinsertしてからChatGPTに全投げする
# つまり、3以上あれば会話が始まっているだろうとみなせる
if len(chat_history) > 2:
chat_history.append(
self.save(
MyChatCompletionMessage(
user=my_chat_completion_message.user,
role=my_chat_completion_message.role,
content=my_chat_completion_message.content,
invisible=False,
)
)
)
response = self.post_to_gpt(chat_history)
latest_assistant = MyChatCompletionMessage(
user=my_chat_completion_message.user,
role=response.choices[0].message.role,
content=response.choices[0].message.content,
invisible=False,
)
chat_history.append(self.save(latest_assistant))
if "本日はなぞなぞにご参加いただき" in latest_assistant.content:
chat_history.append(
self.save(
MyChatCompletionMessage(
user=latest_assistant.user,
role="user",
content="評価結果をjsonで出力してください",
invisible=True,
)
)
)
response = self.post_to_gpt(chat_history)
latest_assistant = MyChatCompletionMessage(
user=my_chat_completion_message.user,
role=response.choices[0].message.role,
content=response.choices[0].message.content,
invisible=True,
)
chat_history.append(self.save(latest_assistant))
return chat_history
def post_to_gpt(
self, chat_history: list[MyChatCompletionMessage]
) -> ChatCompletion:
return self.client.chat.completions.create(
model="gpt-4-turbo",
messages=[x.to_origin() for x in chat_history],
temperature=0.5,
)
def save(
self, messages: MyChatCompletionMessage | list[MyChatCompletionMessage]
) -> MyChatCompletionMessage | list[MyChatCompletionMessage]:
if isinstance(messages, list):
self.chatlog_repository.bulk_insert(messages)
elif isinstance(messages, MyChatCompletionMessage):
self.chatlog_repository.insert(messages)
else:
raise ValueError(
f"Unexpected type {type(messages)}. Expected MyChatCompletionMessage or list[MyChatCompletionMessage]."
)
return messages
class OpenAIDalleService(LLMService):
def __init__(self):
super().__init__()
self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def generate(self, my_chat_completion_message: MyChatCompletionMessage):
"""
画像urlの有効期限は1時間。それ以上使いたいときは保存する。
dall-e-3: 1024x1024, 1792x1024, 1024x1792 のいずれかしか生成できない
"""
if my_chat_completion_message.content is None:
raise Exception("content is None")
response = self.post_to_gpt(my_chat_completion_message.content)
image_url = response.data[0].url
try:
response = requests.get(image_url)
response.raise_for_status()
resized_picture = self.resize(picture=Image.open(BytesIO(response.content)))
my_chat_completion_message = self.save(
resized_picture,
my_chat_completion_message,
)
except requests.exceptions.HTTPError as http_error:
raise Exception(http_error)
except requests.exceptions.ConnectionError as connection_error:
raise Exception(connection_error)
except Exception as e:
raise Exception(e)
return my_chat_completion_message
def post_to_gpt(self, prompt: str):
return self.client.images.generate(
model="dall-e-3", prompt=prompt, size="1024x1024", quality="standard", n=1
)
def save(
self, picture: Image, my_chat_completion_message: MyChatCompletionMessage
) -> MyChatCompletionMessage:
folder_path = Path(MEDIA_ROOT) / "images"
if not folder_path.exists():
folder_path.mkdir(parents=True, exist_ok=True)
# This generates a random string of 10 characters
random_filename = secrets.token_hex(5) + ".jpg"
relative_path_str = "/media/images/" + random_filename
full_path = folder_path / random_filename
my_chat_completion_message.file_path = relative_path_str
picture.save(full_path)
self.chatlog_repository.upsert(my_chat_completion_message)
return my_chat_completion_message
@staticmethod
def resize(picture: Image) -> Image:
return picture.resize((128, 128))
class OpenAITextToSpeechService(LLMService):
def __init__(self):
super().__init__()
self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def generate(self, my_chat_completion_message: MyChatCompletionMessage):
if my_chat_completion_message.content is None:
raise Exception("content is None")
response = self.post_to_gpt(my_chat_completion_message.content)
self.save(response, my_chat_completion_message)
def post_to_gpt(self, text: str):
return self.client.audio.speech.create(
model="tts-1", voice="alloy", input=text, response_format="mp3"
)
def save(self, response, my_chat_completion_message: MyChatCompletionMessage):
folder_path = Path(MEDIA_ROOT) / "audios"
if not folder_path.exists():
folder_path.mkdir(parents=True, exist_ok=True)
# This generates a random string of 10 characters
random_filename = secrets.token_hex(5) + ".mp3"
relative_path_str = "/media/audios/" + random_filename
full_path = folder_path / random_filename
my_chat_completion_message.file_path = relative_path_str
response.write_to_file(full_path)
self.chatlog_repository.upsert(my_chat_completion_message)
return my_chat_completion_message
class OpenAISpeechToTextService(LLMService):
def __init__(self):
super().__init__()
self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def generate(self, my_chat_completion_message: MyChatCompletionMessage):
if my_chat_completion_message.file_path is None:
raise Exception("file_path is None")
full_path = Path(MEDIA_ROOT) / my_chat_completion_message.file_path
if full_path.exists():
response = self.post_to_gpt(str(full_path))
my_chat_completion_message.content = response.text
print(f"\n音声ファイルは「{response.text}」とテキスト化されました\n")
self.save(my_chat_completion_message)
else:
print(f"音声ファイル {my_chat_completion_message.file_path} は存在しません")
def post_to_gpt(self, path_to_audio: str):
audio = open(path_to_audio, "rb")
return self.client.audio.transcriptions.create(model="whisper-1", file=audio)
def save(self, my_chat_completion_message: MyChatCompletionMessage):
self.chatlog_repository.upsert(my_chat_completion_message)
use_cases
line_qa_with_gpt_and_dalle/domain/usecase/llm_service_use_cases.py(new)
from abc import ABC, abstractmethod
from pathlib import Path
from django.contrib.auth.models import User
from django.core.exceptions import ObjectDoesNotExist
from django.db.models import Q
from config import settings
from line_qa_with_gpt_and_dalle.domain.service.llm import (
GeminiService,
OpenAIGptService,
OpenAIDalleService,
OpenAITextToSpeechService,
OpenAISpeechToTextService,
)
from line_qa_with_gpt_and_dalle.domain.valueobject.chat import MyChatCompletionMessage
from line_qa_with_gpt_and_dalle.models import ChatLogsWithLine
class UseCase(ABC):
@abstractmethod
def execute(self, user: User, content: str | None):
pass
class GeminiUseCase(UseCase):
def execute(self, user: User, content: str | None):
"""
GeminiServiceを利用し、ユーザーからの入力（content）を基にテキストを生成します。
contentパラメータはNoneではないこと。
Args:
user (User): DjangoのUserモデルのインスタンス
content (str | None): ユーザーからの入力テキスト
Raises:
ValueError: contentがNoneの場合
Returns:
テキスト生成の結果
"""
if content is None:
raise ValueError("content cannot be None for GeminiUseCase")
llm_service = GeminiService()
my_chat_completion_message = MyChatCompletionMessage(
user=user,
role="user",
content=content,
invisible=False,
)
return llm_service.generate(my_chat_completion_message, gender="man")
class OpenAIGptUseCase(UseCase):
def execute(self, user: User, content: str | None):
"""
OpenAIGptServiceを利用し、ユーザーからの入力（content）を基にテキストを生成します。
contentパラメータはNoneではないこと。
Args:
user (User): DjangoのUserモデルのインスタンス
content (str | None): ユーザーからの入力テキスト
Raises:
ValueError: contentがNoneの場合
Returns:
テキスト生成の結果
"""
if content is None:
raise ValueError("content cannot be None for OpenAIGptUseCase")
llm_service = OpenAIGptService()
my_chat_completion_message = MyChatCompletionMessage(
user=user,
role="user",
content=content,
invisible=False,
)
return llm_service.generate(my_chat_completion_message, gender="man")
class OpenAIDalleUseCase(UseCase):
def execute(self, user: User, content: str | None):
"""
OpenAIDalleServiceを利用し、ユーザーからの入力テキスト（content）を基に画像を生成します。
contentパラメータはNoneではないこと。
Args:
user (User): DjangoのUserモデルのインスタンス
content (str | None): ユーザーからの入力テキスト
Raises:
ValueError: contentがNoneの場合
Returns:
画像生成の結果
"""
if content is None:
raise ValueError("content cannot be None for OpenAIDalleUseCase")
llm_service = OpenAIDalleService()
my_chat_completion_message = MyChatCompletionMessage(
user=user,
role="user",
content=content,
invisible=False,
)
return llm_service.generate(my_chat_completion_message)
class OpenAITextToSpeechUseCase(UseCase):
def execute(self, user: User, content: str | None):
"""
OpenAITextToSpeechServiceを利用し、ユーザーからの入力テキスト（content）を基に音声を生成します。
contentパラメータはNoneではないこと。
Args:
user (User): DjangoのUserモデルのインスタンス
content (str | None): ユーザーからの入力テキスト
Raises:
ValueError: contentがNoneの場合
Returns:
音声生成の結果
"""
if content is None:
raise ValueError("content cannot be None for OpenAITextToSpeechUseCase")
llm_service = OpenAITextToSpeechService()
my_chat_completion_message = MyChatCompletionMessage(
user=user,
role="user",
content=content,
invisible=False,
)
return llm_service.generate(my_chat_completion_message)
class OpenAISpeechToTextUseCase(UseCase):
def execute(self, user: User, content: str | None):
"""
TODO: ちょっとファイルが見つけられないバグがある issue7
OpenAISpeechToTextServiceを利用し、ユーザーの最新の音声ファイルをテキストに変換します。
contentパラメータは必ずNoneであること。
Args:
user (User): DjangoのUserモデルのインスタンス
content (str | None): この引数は現在利用されていません。
Raises:
ValueError: contentがNoneでない場合
Returns:
音声をテキストに変換した結果
"""
if content is not None:
raise ValueError("content must be None for OpenAISpeechToTextUseCase")
record = ChatLogsWithLine.objects.filter(
Q(user=user)
& Q(role="user")
& Q(file_path__endswith=".mp3")
& Q(invisible=False)
).last()
if record is None:
raise ObjectDoesNotExist("No audio file registered for the user")
llm_service = OpenAISpeechToTextService()
my_chat_completion_message = MyChatCompletionMessage(
pk=record.pk,
user=record.user,
role=record.role,
content=record.content,
file_path=str(Path(settings.MEDIA_ROOT) / record.file_path),
invisible=record.invisible,
)
return llm_service.generate(my_chat_completion_message)
valueobject
line_qa_with_gpt_and_dalle/domain/valueobject/chat.py(new)
from django.contrib.auth.models import User
from openai.types.chat import (
ChatCompletionSystemMessageParam,
ChatCompletionAssistantMessageParam,
ChatCompletionUserMessageParam,
)
from line_qa_with_gpt_and_dalle.models import ChatLogsWithLine
class MyChatCompletionMessage:
def __init__(
self,
user: User,
role: str,
invisible: bool,
pk: int = None,
content: str = None,
file_path: str = None,
):
self.id = pk
self.user = user
self.role = role
self.content = content
self.file_path = file_path
self.invisible = invisible
def to_origin(self):
if self.role == "system":
temp = ChatCompletionSystemMessageParam(role="system", content=self.content)
elif self.role == "assistant":
temp = ChatCompletionAssistantMessageParam(
role="assistant", content=self.content
)
else:
temp = ChatCompletionUserMessageParam(role="user", content=self.content)
return temp
def to_entity(self) -> ChatLogsWithLine:
return ChatLogsWithLine(
user=self.user,
role=self.role,
content=self.content,
invisible=self.invisible,
file_path=self.file_path,
)
def __str__(self):
return (
f"user_id: {self.user.pk}, "
f"role: {self.role}, "
f"content: {self.content}, "
f"invisible: {self.invisible}, "
f"file_path: {self.file_path}"
)
line_qa_with_gpt_and_dalle/domain/valueobject/gender.py(new)
class Gender:
def __init__(self, gender):
if gender not in {"man", "woman"}:
raise ValueError("Invalid gender")
self.gender = gender
@property
def name(self) -> str:
return "男性" if self.gender == "man" else "女性"
確認
console
python manage.py runserver
LINE
LINEビジネスアカウントのセットアップ
開発中はLINEアプリからのログインなんてやってらんないので、メールアドレスでログインできるようにビジネスアカウントをセットアップした
プロバイダの作成
LINEは予約語になっている
Messaging APIチャネルを作成
まず、LINE DevelopersコンソールからMessaging APIチャネルを作成します。チャネルを作成すると、そのチャネル用のLINE公式アカウントが作成されます。
作ったプロバイダに、チャネルを Messaging APIモードとしてつくると、そのモードの公式アカウントが作成できる、とイメージすると僕は理解しやすかった。
| 設定項目 | 設定値 |
|---|---|
| チャネルの種類 | Messaging API |
| プロバイダー | QA_WITH_GPT |
| 会社・事業者の所在国・地域 | 日本 |
| チャネル名 | GPTと会話する公式チャンネル |
| チャネル説明 | あのチャネルです |
| 大業種 | 通信・情報・メディア |
| 小業種 | 情報サービス |
| メールアドレス | （自分のメールアドレス） |
トークンを発行
［Messaging API設定］ のタブから、いくつかの種類のチャネルアクセストークンを発行できるが、いちばん簡単なのが長期のチャネルアクセストークン（APIKeyみたいなん）なので、「発行」ボタンで発行し、.env に 控える
- 任意の有効期間を指定できるチャネルアクセストークン（チャネルアクセストークンv2.1）
- ステートレスチャネルアクセストークン
- 短期のチャネルアクセストークン
- 長期のチャネルアクセストークン
| .env項目 | 取得できるページ |
|---|---|
| LINE_CHANNEL_SECRET | チャネル基本設定 タブ |
| LINE_CHANNEL_ACCESS_TOKEN | Messaging API設定 タブ |
Webhook URLを設定する
Webhook URLはボットサーバーのエンドポイントで、Webhookペイロードの送信先です。
この書きがよくわかんないけど、「ボットサーバーのエンドポイント」というのはどうやらdjangoアプリのことを言っているようだ。ボットサーバーなんだからLINEプラットフォームの公式チャンネルのことだろ？って考えると、思考の沼にハマる 
- LINE Developersコンソール にログインし、Messaging APIのチャネルがあるプロバイダーをクリックします
- Messaging APIのチャネルをクリックします
- ［Messaging API設定］タブをクリックします
- ［Webhook URL］の［編集］をクリックし、Webhook URL（LINEプラットフォームからボットにイベントを送信する際の送信先URL）を入力して、［更新］をクリックします
- Webhook URLにはHTTPSを使用し、一般的なブラウザ等で広く信頼されている認証局で発行されたSSL/TLS証明書を設定する必要があります。また、自己署名証明書は利用できない点に注意してください。SSL/TLSの設定で問題が発生した場合は、SSL/TLS証明書チェーンが完全で、中間証明書がサーバーに正しくインストールされていることを確かめてください
- ［検証］をクリックします。設定したWebhook URLでWebhookイベントを受け取ると、「成功」と表示されます
- ［Webhookの利用］を有効にします
自動応答をOFFにする
会話ができるようになると常にこのようなメッセージが送信されるが邪魔なのでOFFにする
webhookのPOSTが叩かれるエンドポイントを作ろう
つまるところ、公式チャンネルに何らかのイベントが発生すると、LINEがdjangoアプリのurlをPOSTで叩きに来る、と。
ということは urls.py に /line_webhook というエンドポイントを作ればよい。
ここで問題なのは以下のこと。
- 開発サーバ（http://127.0.0.1:8000/line_qa_with_gpt_and_dalle/line_webhook/） じゃなくて、本番サーバで動いているパスが必要
- オレオレ証明書が無効
本番サーバ一つしか持ってなくてポートフォリオが動いてるし、これだけのために本番サーバ増やすのもさすがに面倒なので、既存の僕ポートフォリオ（本番・証明書あり）で POST を処理するエンドポイント https://www.henojiya.net/linebot_engine/callback/ 作って検証した
urls.py
linebot_engine/urls.py（別のプロジェクトの内容なので注意）
from django.urls import path
from linebot_engine.views import CallbackView
app_name = "bot"
urlpatterns = [path("callback/", CallbackView.as_view(), name="callback")]
service
linebot_engine/domain/service/line_service.py
import base64
import hashlib
import hmac
import os
import secrets
import requests
from django.core.files.base import ContentFile
from django.http import HttpRequest
from linebot.api import LineBotApi
from linebot.models import TextSendMessage, ImageSendMessage
from config.settings import SITE_URL, MEDIA_URL
from linebot_engine.domain.valueobject.line import WebhookEvent
from linebot_engine.models import UserProfile, Message
class LineService:
"""A service class to encapsulate LINE related business logic."""
@staticmethod
def is_valid_signature(request: HttpRequest) -> bool:
"""Check LINE webhook request signature."""
body = request.body.decode("utf-8")
_hash = hmac.new(
os.environ.get("LINE_CHANNEL_SECRET").encode("utf-8"),
body.encode("utf-8"),
hashlib.sha256,
).digest()
signature = base64.b64encode(_hash).decode("utf-8")
# Retrieve X-Line-Signature header value
x_line_signature = request.META.get("HTTP_X_LINE_SIGNATURE")
# Compare X-Line-Signature header value and the signature calculated in the code
return x_line_signature == signature
@staticmethod
def _get_picture_content(url: str) -> ContentFile:
response = requests.get(url)
response.raise_for_status()
return ContentFile(response.content)
def handle_event(self, event: WebhookEvent, line_user_id: str):
"""
follow, unfollow and message という種類の異なるイベントを処理します
is_follow(): プロフィール画像を保存し、ユーザープロフィールを作成します
is_unfollow(): ユーザープロフィールを削除します
is_message():
text: メッセージの記録を作成します
image: 画像のコンテンツを取得し、`self.picture_save()` で保存します
Args:
event: イベントの情報を含むLineのイベントオブジェクト
line_user_id: イベントに関連付けられたLineのユーザーID
"""
line_bot_api = LineBotApi(os.environ.get("LINE_CHANNEL_ACCESS_TOKEN"))
if event.is_follow():
profile = line_bot_api.get_profile(event.source.user_id)
picture_file = self._get_picture_content(profile.picture_url)
picture_file.name = f"{secrets.token_hex(10)}.png"
UserProfile.objects.create(
line_user_id=line_user_id,
display_name=profile.display_name,
picture=picture_file,
)
elif event.is_unfollow():
UserProfile.objects.filter(line_user_id=line_user_id).delete()
elif event.is_message():
if event.event_data.type == "text":
Message.objects.create(
user_profile=UserProfile.objects.get(line_user_id=line_user_id),
source_type=event.event_data.type,
message=event.event_data.text,
)
text_message = TextSendMessage(text="textが記録されました")
line_bot_api.reply_message(event.reply_token, text_message)
elif event.event_data.type == "image":
message_content = line_bot_api.get_message_content(event.event_data.id)
picture_file = ContentFile(message_content.content)
picture_file.name = f"{secrets.token_hex(10)}.png"
Message.objects.create(
user_profile=UserProfile.objects.get(line_user_id=line_user_id),
source_type=event.event_data.type,
picture=picture_file,
)
full_picture_url = (
f"{SITE_URL}{MEDIA_URL}/linebot_engine/images/{picture_file.name}"
)
image_send_message = ImageSendMessage(
original_content_url=full_picture_url,
preview_image_url=full_picture_url,
)
text_message = TextSendMessage(
text="imageが記録されました。ほらこれでしょ？"
)
line_bot_api.reply_message(
event.reply_token, [image_send_message, text_message]
)
else:
line_bot_api.reply_message(
event.reply_token, TextSendMessage(text="Unsupported event type")
)
views.py
linebot_engine/views.py（別のプロジェクトの内容なので注意）
import json
from django.http import HttpResponse
from django.utils.decorators import method_decorator
from django.views import View
from django.views.decorators.csrf import csrf_exempt
from linebot_engine.domain.service.line_service import LineService
from linebot_engine.domain.valueobject.line import WebhookEvent
WEBHOOK_VERIFICATION_USER_ID = "Udeadbeefdeadbeefdeadbeefdeadbeef"
@method_decorator(csrf_exempt, name="dispatch")
class CallbackView(View):
@staticmethod
def post(request, *args, **kwargs):
"""
ラインの友達追加時に呼び出され、ラインのIDを登録する
Notes: LINE DEVELOPERS（ビジネスアカウントログイン）の画面からWebhookの接続をテストした場合
実際のイベント（ユーザーのアクションなど）がないため、eventsデータは存在せず、空の配列が返される
"""
line_service = LineService()
if not line_service.is_valid_signature(request):
return HttpResponse(status=403) # return 'Forbidden'
request_json = json.loads(request.body.decode("utf-8"))
events = [WebhookEvent(event) for event in request_json["events"]]
for event in events:
# TODO: .is_group() のときの処理はTBD
if event.source and event.source.is_user():
line_user_id = event.source.user_id
if line_user_id != WEBHOOK_VERIFICATION_USER_ID:
line_service.handle_event(event, line_user_id)
return HttpResponse(status=200)
@csrf_exempt は必要。検証ボタンで疎通確認するときに 403 が発生してしまうため
もういちどこの絵を置いておこう。
この絵を踏まえると、いろんなLINEのイベントのなかから特定のイベントを抜き出して、以下のようなことができそうだ
- 友だち登録時のイベントを識別して、その友達のuser_idをdbのuserと紐づける
- 友だちがトークでメッセージを投げてきたら、そのイベントを識別してChatGPTに投げる。その結果をdbに保存したりして、処理結果をトークで返す
ガツッとドメインにしまい込む
サーバーに届くWebhookの例
{
"destination": "xxxxxxxxxx",
"events": [
{
"type": "message",
"message": {
"type": "text",
"id": "468789577898262530", // 送られてきたメッセージのID
"quotedMessageId": "468789532432007169", // 引用対象となったメッセージのID
"quoteToken": "q3Plxr4AgKd...",
"text": "Chicken, please." // 送られてきたメッセージのテキスト
},
"webhookEventId": "01H810YECXQQZ37VAXPF6H9E6T",
"deliveryContext": {
"isRedelivery": false
},
"timestamp": 1692251666727,
"source": {
"type": "group",
"groupId": "Ca56f94637c...",
"userId": "U4af4980629..."
},
"replyToken": "38ef843bde154d9b91c21320ffd17a0f",
"mode": "active"
}
]
}
「上記のデータを想定して valueobjectのクラスを作って？」って聞くとあらかた作ってくれるよ。そう、jetbrain の AI Assistant ならね。もちろん漏れがないか見極める力は必要だけどね
valueobject
linebot_engine/domain/valueobject/line.py
class WebhookEvent:
"""
Notes: https://developers.line.biz/ja/reference/messaging-api/#webhook-event-objects
"""
class _Source:
"""
type: user or group
Notes: 3人以上でトークを利用することを「グループトーク」という
※1対1のトーク中に友だちを誘って会話することを「複数人トーク」というが、現在は、グループトークに機能が統合されているらしい
"""
def __init__(self, source_dict):
self.type = source_dict.get("type")
self.group_id = source_dict.get("groupId")
self.user_id = source_dict.get("userId")
def is_user(self):
return self.type == "user"
def is_group(self):
return self.type == "group"
class _Message:
class Emoji:
def __init__(self, emoji):
"""
index: テキストの先頭を0とした、textプロパティ内の絵文字の開始位置
length: LINE絵文字の文字列の長さ。LINE絵文字 `(hello)` であれば、7
productId: LINE絵文字の集合を示すプロダクトID
emojiId: プロダクトID内のLINE絵文字のID
productId, emojiId は https://developers.line.biz/ja/docs/messaging-api/emoji-list/ を参照
"""
self.index = emoji.get("index")
self.length = emoji.get("length")
self.product_id = emoji.get("productId")
self.emoji_id = emoji.get("emojiId")
class _Mention:
"""
Message.textプロパティにメンションが含まれる場合のみ、Messageクラスに含まれる
"""
class Mentionee:
def __init__(self, mentionee):
"""
index: テキストの先頭を0とした、textプロパティ内のメンションの開始位置。
length: メンションの長さ。@exampleであれば、8
type: メンションの対象 ["user": ユーザー, "all": グループ全体]
userId: メンションされたユーザーのユーザーID。プロフィール情報を取得することに、ユーザーが同意しているときにのみ含まれる
quotedMessageId: 引用されたメッセージのメッセージID。過去のメッセージを引用している場合にのみ含まれる
"""
self.index = mentionee.get("index")
self.length = mentionee.get("length")
self.type = mentionee.get("type")
self.user_id = mentionee.get("userId")
self.quoted_message_id = mentionee.get("quotedMessageId")
def __init__(self, mention):
self.mentionees = [
self.Mentionee(m) for m in mention.get("mentionees", [])
]
def __init__(self, message):
"""
id: メッセージID
text: エンドユーザーがLINE絵文字を送信した場合は、(hello)や(love)のように、LINE絵文字が文字列で含まれます。
エンドユーザーがメンションした場合は、`@example`のように、送信相手のLINEアカウントの表示名が文字列で含まれます。
メンションの詳細は、mentionプロパティで確認できます。
i.e. `@All @example Good Morning!! (love)`
quoteToken: メッセージの引用トークン
emojis: textプロパティに含まれる絵文字の配列
mention: extプロパティに含まれるメンションの情報
Notes: text に絵文字もメンションも含まれるので、 emojis, mention で文字加工する必要はなさそう
"""
self.id = message.get("id")
self.type = message.get("type")
self.text = message.get("text")
self.quote_token = message.get("quoteToken")
self.emojis = [self.Emoji(e) for e in message.get("emojis", [])]
self.mention = (
self._Mention(message.get("mention", {}))
if "mention" in message
else None
)
class _Follow:
"""
Notes: 初めて友だち追加されたとき、isUnblockedの値はfalseとなります。
これは、ユーザがブロックから解除されたのではなく、新たに友だちに追加されたことを示します
一方、ユーザがブロックから解除された場合、isUnblockedの値はtrueとなります。
これは、ユーザがあなたの公式アカウントのブロックを解除したことを示します
"""
def __init__(self, follow_event):
self.is_unblocked = follow_event.get("isUnblocked", False)
def __init__(self, event):
"""
replyToken: イベントに対して応答メッセージを送る際に使用する応答トークン
type: message or follow or unfollow and more
https://developers.line.biz/ja/reference/messaging-api/#message-event を参照
mode: active or standby チャネルの状態。standbyのときは、メッセージの送信を控えてください
timestamp: イベントの発生時刻 UNIX時間（ミリ秒）。再送されたWebhookの場合でも当初時刻を示す
source: イベントの送信元情報を含むユーザー、グループトーク。アカウントの連携に失敗した場合含まれない
webhookEventId: WebhookイベントID。Webhookイベントを一意に識別するためのID。ULID形式の文字列
deliveryContext: isRedelivery が True なら再送されたことを示す
event_data: メッセージかフォローアクションかなどによって別のオブジェクトをロードする
"""
self.reply_token = event.get("replyToken")
self.type = event.get("type")
self.mode = event.get("mode")
self.timestamp = event.get("timestamp")
self.source = self._Source(event.get("source"))
self.webhook_event_id = event.get("webhookEventId")
self.delivery_context = event.get("deliveryContext")
self.event_data = self._parse_event_data(event)
def _parse_event_data(self, event):
if self.is_message():
return self._Message(event.get("message", {}))
elif self.is_follow() or self.is_unfollow():
return self._Follow(event.get("follow", {}))
else:
raise ValueError(f"Unsupported event type: {self.type}")
def is_message(self):
return self.type == "message"
def is_follow(self):
return self.type == "follow"
def is_unfollow(self):
return self.type == "unfollow"
config/settings.py
:
+ SITE_URL = "https://www.henojiya.net"
STATIC_ROOT = BASE_DIR / "static" STATIC_ROOT = BASE_DIR / "static"
: