classDiagram
class CreateQuizUseCase {
-quiz_repository: QuizRepository
-quiz_generator: QuizGenerator
-text_processing_service: TextProcessingService
-vectorstore_repository: VectorStoreRepository
+execute()
}
class QuizRepository {
<<interface>>
+save()
+find_by_id()
+save_user_answers()
}
class QuizRepositoryImpl {
-gcs_client: GCSClient
-bucket_name: str
+save()
+find_by_id()
+save_user_answers()
}
class QuizGenerator {
-llm: OpenAILLM
+generate_quiz()
-_create_prompt()
-_parse_llm_response()
}
class TextProcessingService {
-text_splitter: CharacterTextSplitter
-embedder: OpenAIEmbedding
-vectorstore_repository: VectorStoreRepository
-scraper: FireCrawlLoader
+process_text()
+process_from_url()
}
class VectorStoreRepository {
<<interface>>
+add_documents()
+as_retriever()
+delete()
}
class FAISSVectorStore {
-embedding: Embeddings
-db
+add_documents()
+as_retriever()
+delete()
}
class Quiz {
-quiz_id: QuizId
-questions: List<Question>
-difficulty: str
-gcs_path: str
-user_answers: dict
}
class Question{
-text: str
-options: List[str]
-correct_answer: str
-explanation: str
}
class GCSClient {
-client
+upload_blob()
+download_blob()
+delete_blob()
}
class OpenAILLM{
-llm
-prompt
+generate()
}
CreateQuizUseCase -- QuizRepository : uses
CreateQuizUseCase -- QuizGenerator : uses
CreateQuizUseCase -- TextProcessingService : uses
CreateQuizUseCase -- VectorStoreRepository: uses
QuizRepository <|.. QuizRepositoryImpl : implements
QuizRepositoryImpl -- GCSClient : uses
QuizGenerator -- OpenAILLM : uses
QuizGenerator -- Quiz : creates
QuizGenerator -- Question: creates
TextProcessingService -- CharacterTextSplitter: uses
TextProcessingService -- OpenAIEmbedding : uses
TextProcessingService -- VectorStoreRepository : uses
TextProcessingService -- FireCrawlLoader: uses
VectorStoreRepository <|.. FAISSVectorStore : implements
FAISSVectorStore -- OpenAIEmbedding: uses
Quiz *-- Question : contains
class CharacterTextSplitter{
}
class OpenAIEmbedding{
}
class FireCrawlLoader{
}
修正案に基づいたディレクトリ構成と、主要な部分の具体的な実装例を以下に示します。
ディレクトリ構成:
readum/
├── api/ # プレゼンテーション層 (FastAPI)
│ ├── routes/ # エンドポイント定義
│ │ ├── __init__.py
│ │ ├── quizzes.py # クイズ作成、回答関連のエンドポイント
│ │ └── ... # その他のエンドポイント
│ ├── schemas/ # リクエスト/レスポンスのデータ構造 (Pydantic モデル)
│ │ ├── __init__.py
│ │ ├── quiz.py # クイズ作成リクエスト、レスポンスのモデル
│ │ ├── answer.py # 回答リクエスト、レスポンスのモデル
│ │ └── ...
│ ├── dependencies.py # 依存性注入 (DI) のための設定
│ ├── __init__.py
│ └── main.py # FastAPI アプリケーションのエントリポイント
│
├── application/ # アプリケーション層 (ユースケース)
│ ├── use_cases/ # ユースケースクラス
│ │ ├── __init__.py
│ │ ├── create_quiz.py # クイズ作成ユースケース
│ │ ├── check_answer.py # 回答チェックユースケース
│ │ └── ... # その他のユースケース
│ ├── dto/ # Data Transfer Object (層間のデータ受け渡し用)
│ │ ├── __init__.py
│ │ ├── quiz.py # クイズ関連の DTO
│ │ └── ...
│ └── __init__.py
│
├── domain/ # ドメイン層
│ ├── models/ # エンティティ、値オブジェクト
│ │ ├── __init__.py
│ │ ├── quiz.py # Quiz エンティティ、Question エンティティ
│ │ ├── quiz_difficulty.py # QuizDifficulty 値オブジェクト (Enum)
│ │ ├── quiz_type.py # QuizType 値オブジェクト (Enum)
│ │ ├── question_count.py # QuestionCount 値オブジェクト
│ │ └── ...
│ ├── services/ # ドメインサービス
│ │ ├── __init__.py
│ │ ├── quiz_generator.py # クイズ生成サービス (LLM 呼び出しを含む)
│ │ └── text_processing_service.py # テキスト前処理、ベクトルDB操作
│ ├── repositories/ # リポジトリインターフェース
│ │ ├── __init__.py
│ │ ├── quiz_repository.py # Quiz リポジトリインターフェース
│ │ └── vectorstore_repository.py # VectorStore リポジトリインターフェース
│ ├── exceptions.py # ドメイン層のカスタム例外
│ └── __init__.py
│
├── infrastructure/ # インフラストラクチャ層
│ ├── embedding/ # 埋め込み生成 (OpenAI Embeddings)
│ │ ├── __init__.py
│ │ └── openai_embedding.py
│ ├── vectorstore/ # ベクトルDB (FAISS)
│ │ ├── __init__.py
│ │ └── faiss_vectorstore.py
│ ├── scraping/ # スクレイピング (FireCrawl)
│ │ ├── __init__.py
│ │ └── firecrawl_loader.py
│ ├── llm/ # LLM 呼び出し (LangChain, OpenAI)
│ │ ├── __init__.py
│ │ └── openai_llm.py
│ ├── text_splitter/ # テキスト分割 (CharacterTextSplitter)
│ │ ├── __init__.py
│ │ └── character_text_splitter.py
│ ├── persistence/ # 永続化 (GCS)
│ │ ├── __init__.py
│ │ ├── quiz_repository_impl.py # QuizRepository の GCS 実装
│ │ └── gcs_client.py # GCS クライアント
│ ├── utils/ # ユーティリティ
│ │ ├── __init__.py
│ │ ├── file_system.py # ファイルシステム操作
│ │ └── logger.py # ロギング
│ └── __init__.py
│
├── tests/ # テスト (pytest)
│ ├── api/ # API 層のテスト
│ │ ├── routes/
│ │ │ └── test_quizzes.py
│ │ └── ...
│ ├── application/ # アプリケーション層のテスト
│ │ ├── use_cases/
│ │ │ └── test_create_quiz.py
│ │ └── ...
│ ├── domain/ # ドメイン層のテスト
│ │ ├── models/
│ │ │ └── test_quiz.py
│ │ ├── services/
│ │ │ └── test_quiz_generator.py
│ │ └── ...
│ ├── infrastructure/ # インフラ層のテスト
│ │ ├── persistence/
│ │ │ └── test_quiz_repository_impl.py
│ │ └── ...
│ └── conftest.py # pytest の設定 (fixture など)
│
├── config/ # 設定ファイル
│ └── settings.py # 環境変数などから設定を読み込む
├── main.py # アプリケーションのエントリポイント (FastAPI の起動など)
├── Dockerfile # Dockerfile
├── Pipfile # 依存関係管理 (pipenv)
├── Pipfile.lock # 依存関係ロックファイル
└── README.md # プロジェクトの説明
主要なコード例:
Python
# domain/models/quiz.py
from typing import List, NewType
from dataclasses import dataclass, field
QuizId = NewType("QuizId", str)
@dataclass
class Question:
text: str
options: List[str]
correct_answer: str
explanation: str
@dataclass
class Quiz:
quiz_id: QuizId
questions: List[Question]
difficulty: str # QuizDifficulty (Enum) を使うべき
gcs_path: str = ""
user_answers: dict = field(default_factory=dict)
# domain/models/quiz_difficulty.py
from enum import Enum
class QuizDifficulty(str, Enum):
BEGINNER = "beginner"
INTERMEDIATE = "intermediate"
ADVANCED = "advanced"
# domain/models/quiz_type.py
from enum import Enum
class QuizType(str, Enum):
DIRECT = "direct"
URL = "url"
# domain/models/question_count.py
from dataclasses import dataclass
@dataclass(frozen=True)
class QuestionCount:
value: int
def __post_init__(self):
if self.value <= 0:
raise ValueError("問題数は1以上である必要があります")
# domain/services/quiz_generator.py
from domain.models.quiz import Quiz, Question, QuizDifficulty, QuestionCount
from infrastructure.llm.openai_llm import OpenAILLM # 仮: インターフェースに依存させるべき
from langchain_core.documents import Document
class QuizGenerator: # ドメインサービス
def __init__(self, llm: OpenAILLM): # DI
self.llm = llm
def generate_quiz(
self, context: List[Document], difficulty: QuizDifficulty, question_count: QuestionCount
) -> Quiz:
prompt = self._create_prompt(context, difficulty, question_count)
llm_response = self.llm.generate(prompt)
questions = self._parse_llm_response(llm_response)
return Quiz(quiz_id="generated_quiz_id", questions=questions, difficulty=difficulty.value) # type: ignore
def _create_prompt(
self, context: List[Document], difficulty: QuizDifficulty, question_count: QuestionCount
) -> str:
# プロンプトを作成
pass
def _parse_llm_response(self, llm_response: str) -> list[Question]:
# LLMの回答をパース
pass
# domain/services/text_processing_service.py
from typing import List
from langchain_core.documents import Document
from domain.repositories.vectorstore_repository import VectorStoreRepository
from infrastructure.text_splitter.character_text_splitter import (
CharacterTextSplitter,
)
from infrastructure.embedding.openai_embedding import OpenAIEmbedding
from infrastructure.scraping.firecrawl_loader import FireCrawlLoader # 仮: インターフェースに依存させるべき
class TextProcessingService:
def __init__(
self,
text_splitter: CharacterTextSplitter,
embedder: OpenAIEmbedding,
vectorstore_repository: VectorStoreRepository,
scraper: FireCrawlLoader,
):
self.text_splitter = text_splitter
self.embedder = embedder
self.vectorstore_repository = vectorstore_repository
self.scraper = scraper
def process_text(self, text: str) -> List[Document]:
chunks = self.text_splitter.split_text(text)
self.vectorstore_repository.add_documents(chunks)
return chunks
def process_from_url(self, url: str) -> List[Document]:
# スクレイピングの処理
scraped_text = self.scraper.scrape(url)
chunks = self.text_splitter.split_text(scraped_text)
self.vectorstore_repository.add_documents(chunks)
return chunks
# domain/repositories/quiz_repository.py
from abc import ABC, abstractmethod
from typing import List, Optional
from domain.models.quiz import Quiz, QuizId
class QuizRepository(ABC):
@abstractmethod
def save(self, quiz: Quiz) -> None:
raise NotImplementedError
@abstractmethod
def find_by_id(self, quiz_id: QuizId) -> Optional[Quiz]:
raise NotImplementedError
@abstractmethod
def save_user_answers(self, quiz_id: QuizId, user_answers: dict) -> None:
raise NotImplementedError
# domain/repositories/vectorstore_repository.py
from abc import ABC, abstractmethod
from typing import List
from langchain_core.documents import Document
class VectorStoreRepository(ABC):
@abstractmethod
def add_documents(self, documents: List[Document]) -> None:
pass
@abstractmethod
def as_retriever(self, **kwargs):
pass
@abstractmethod
def delete(self) -> None:
pass
# infrastructure/persistence/quiz_repository_impl.py
import json
from typing import List, Optional
from domain.models.quiz import Quiz, QuizId, Question
from domain.repositories.quiz_repository import QuizRepository
from infrastructure.persistence.gcs_client import GCSClient # GCS クライアント
class QuizRepositoryImpl(QuizRepository):
def __init__(self, gcs_client: GCSClient):
self.gcs_client = gcs_client # GCS クライアントの初期化
self.bucket_name = "your-gcs-bucket-name" # バケット名を指定
def save(self, quiz: Quiz) -> None:
# Quiz オブジェクトを JSON にシリアライズ
quiz_data = {
"quiz_id": quiz.quiz_id,
"questions": [
{
"text": q.text,
"options": q.options,
"correct_answer": q.correct_answer,
"explanation": q.explanation,
}
for q in quiz.questions
],
"difficulty": quiz.difficulty,
"gcs_path": quiz.gcs_path, # gcs_pathを保存
}
quiz_json = json.dumps(quiz_data)
# GCS に保存 (ファイル名は quiz_id.json とする)
if quiz.gcs_path == "":
file_path = f"quizzes/{quiz.quiz_id}.json"
quiz.gcs_path = file_path # gcs_pathをドメインオブジェクトに格納
else:
file_path = quiz.gcs_path
self.gcs_client.upload_blob(self.bucket_name, quiz_json, file_path)
def find_by_id(self, quiz_id: QuizId) -> Optional[Quiz]:
file_path = f"quizzes/{quiz_id}.json"
quiz_json = self.gcs_client.download_blob(self.bucket_name, file_path)
if quiz_json:
quiz_data = json.loads(quiz_json)
questions = [
Question(
text=q["text"],
options=q["options"],
correct_answer=q["correct_answer"],
explanation=q["explanation"],
)
for q in quiz_data["questions"]
]
return Quiz(
quiz_id=quiz_data["quiz_id"],
questions=questions,
difficulty=quiz_data["difficulty"],
gcs_path=quiz_data["gcs_path"],
)
else:
return None
def save_user_answers(self, quiz_id: QuizId, user_answers: dict) -> None:
# GCS から既存のクイズデータを取得
file_path = f"quizzes/{quiz_id}.json"
quiz_json = self.gcs_client.download_blob(self.bucket_name, file_path)
if quiz_json:
quiz_data = json.loads(quiz_json)
# ユーザーの回答を追加
quiz_data["user_answers"] = user_answers
# 更新したデータをGCSにアップロード
updated_quiz_json = json.dumps(quiz_data)
self.gcs_client.upload_blob(
self.bucket_name, updated_quiz_json, file_path
)
else:
# TODO 404を返すようにする
raise FileNotFoundError(f"Quiz with id {quiz_id} not found in GCS.")
# infrastructure/vectorstore/faiss_vectorstore.py
from langchain_core.vectorstores import VectorStore, VectorStoreRetriever
from domain.repositories.vectorstore_repository import VectorStoreRepository
from langchain_core.embeddings import Embeddings
from langchain_core.documents import Document
class FAISSVectorStore(VectorStore, VectorStoreRepository):
def __init__(self, embedding: Embeddings):
self.embedding = embedding
self.db = None # from_documents, from_textsを呼び出した時に初期化
def add_documents(self, documents: list[Document]) -> None:
if self.db is None:
self.db = self.from_documents(documents, self.embedding)
else:
self.db.add_documents(documents)
def as_retriever(self, **kwargs) -> VectorStoreRetriever:
if self.db is None:
raise ValueError("FAISSVectorStore is empty. Please add_documents first.")
return self.db.as_retriever(**kwargs)
def delete(self) -> None:
self.db = None # メモリから削除
def from_documents(self, documents, embedding: Embeddings):
# FAISSのfrom_documentsを呼び出す
pass
def from_texts(self, texts, embedding):
#
Python
# infrastructure/llm/openai_llm.py (続き)
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from config.settings import Settings # 設定情報をimport
class OpenAILLM:
def __init__(self):
self.llm = ChatOpenAI(model_name=Settings.model.GPT_MODEL)
self.prompt = PromptTemplate.from_template(
"""
あなたは文章理解度チェッククイズ作成のプロフェッショナルです。
与えられた文章から、内容理解度をチェックするためのクイズを作成してください。
# 指示
- クイズは4択形式で{question_count}問作成してください。
- 各問題には、問題文、選択肢(1, 2, 3, 4)、正解の選択肢番号、詳しい解説を含めてください。
- クイズの難易度は{difficulty}です。
# 文章
{context}
# 出力形式
1. 問題文\\n\\n選択肢\\n\\n正解\\n\\n解説\\n\\n
2. 問題文\\n\\n選択肢\\n\\n正解\\n\\n解説\\n\\n ...
"""
)
def generate(self, prompt:str) -> str:
return self.llm.invoke(prompt).content # type: ignore
# infrastructure/scraping/firecrawl_loader.py
# (仮の実装。実際には FireCrawl ライブラリを使用)
class FireCrawlLoader:
def scrape(self, url: str) -> str:
# FireCrawl を使って Web ページをスクレイピング
# 実際には、FireCrawlLoader を適切に設定して使用
return "Scraped text from " + url
# infrastructure/text_splitter/character_text_splitter.py
from langchain_text_splitters import CharacterTextSplitter
from config.settings import Settings
from langchain_core.documents import Document
class CustomCharacterTextSplitter(CharacterTextSplitter):
def __init__(self, chunk_size: int = Settings.text_splitter.CHUNK_SIZE, chunk_overlap: int = Settings.text_splitter.CHUNK_OVERLAP, separator: str = "\\n\\n", ):
super().__init__(chunk_size=chunk_size, chunk_overlap=chunk_overlap, separator=separator)
def split_text(self, text: str) -> list[Document]:
docs = super().split_text(text)
# Documentオブジェクトに変換
return [Document(page_content=x) for x in docs]
# application/use_cases/create_quiz.py
from domain.models.quiz import QuizDifficulty, QuestionCount, QuizType, Quiz
from domain.services.quiz_generator import QuizGenerator
from domain.services.text_processing_service import TextProcessingService
from domain.repositories.quiz_repository import QuizRepository
from domain.repositories.vectorstore_repository import VectorStoreRepository
from domain.exceptions import QuizCreationError
class CreateQuizUseCase:
def __init__(
self,
quiz_repository: QuizRepository,
quiz_generator: QuizGenerator,
text_processing_service: TextProcessingService,
vectorstore_repository: VectorStoreRepository,
):
self.quiz_repository = quiz_repository
self.quiz_generator = quiz_generator
self.text_processing_service = text_processing_service
self.vectorstore_repository = vectorstore_repository
def execute(
self,
input_text: str,
input_url: str,
quiz_type: QuizType,
difficulty: QuizDifficulty,
question_count: QuestionCount,
) -> Quiz:
try:
if quiz_type == QuizType.DIRECT:
chunks = self.text_processing_service.process_text(input_text)
elif quiz_type == QuizType.URL:
chunks = self.text_processing_service.process_from_url(input_url) # type: ignore
else:
raise ValueError("Invalid quiz type")
# ベクトルDBから関連情報を取得 (Retriever を使用)
retriever = self.vectorstore_repository.as_retriever()
context = retriever.get_relevant_documents(
f"次の条件に従ってクイズを作成してください。難易度:{difficulty.value}, 問題数:{question_count.value}"
)
quiz = self.quiz_generator.generate_quiz(context, difficulty, question_count)
self.quiz_repository.save(quiz)
self.vectorstore_repository.delete()
return quiz
except Exception as e:
raise QuizCreationError("Failed to create quiz") from e
# api/dependencies.py
from infrastructure.persistence.quiz_repository_impl import QuizRepositoryImpl
from infrastructure.llm.openai_llm import OpenAILLM
from infrastructure.embedding.openai_embedding import OpenAIEmbedding
from domain.services.quiz_generator import QuizGenerator
from domain.repositories.quiz_repository import QuizRepository
from domain.repositories.vectorstore_repository import VectorStoreRepository
from infrastructure.vectorstore.faiss_vectorstore import FAISSVectorStore
from infrastructure.persistence.gcs_client import GCSClient
from infrastructure.text_splitter.character_text_splitter import CustomCharacterTextSplitter
from infrastructure.scraping.firecrawl_loader import FireCrawlLoader # 仮
from domain.services.text_processing_service import TextProcessingService
def get_quiz_repository() -> QuizRepository:
return QuizRepositoryImpl(GCSClient())
def get_quiz_generator() -> QuizGenerator:
return QuizGenerator(OpenAILLM())
def get_vectorstore_repository() -> VectorStoreRepository:
return FAISSVectorStore(OpenAIEmbedding())
def get_text_processing_service() -> TextProcessingService:
return TextProcessingService(CustomCharacterTextSplitter(), OpenAIEmbedding(), get_vectorstore_repository(), FireCrawlLoader())
# api/routes/quizzes.py
from fastapi import APIRouter, HTTPException, Depends
from application.use_cases.create_quiz import CreateQuizUseCase
from api.models.request import QuizRequest, QuizType
from api.models.response import QuizResponse
from api.dependencies import (
get_quiz_repository,
get_quiz_generator,
get_text_processing_service,
get_vectorstore_repository
)
from domain.repositories.quiz_repository import QuizRepository
from domain.services.quiz_generator import QuizGenerator
from domain.services.text_processing_service import TextProcessingService
from domain.repositories.vectorstore_repository import VectorStoreRepository
from domain.exceptions import QuizCreationError
router = APIRouter()
@router.post("/create_quiz", response_model=QuizResponse)
async def create_quiz(
quiz_request: QuizRequest,
quiz_repository: QuizRepository = Depends(get_quiz_repository),
quiz_generator: QuizGenerator = Depends(get_quiz_generator),
text_processing_service: TextProcessingService = Depends(get_text_processing_service),
vectorstore_repository: VectorStoreRepository = Depends(get_vectorstore_repository)
):
try:
use_case = CreateQuizUseCase(
quiz_repository=quiz_repository,
quiz_generator=quiz_generator,
text_processing_service=text_processing_service,
vectorstore_repository=vectorstore_repository
)
quiz = use_case.execute(
input_text=quiz_request.content if quiz_request.type == QuizType.TEXT else "",
input_url=quiz_request.content if quiz_request.type == QuizType.URL else "",
quiz_type=quiz_request.type,
difficulty=quiz_request.difficulty,
question_count=quiz_request.question_count,
)
return QuizResponse(id=quiz.quiz_id, preview=quiz.questions) # 一旦questionsをpreviewとする
except QuizCreationError as e:
raise HTTPException(status_code=500, detail="Quiz creation failed.")
except Exception as e:
print(e)
raise HTTPException(status_code=500, detail="Internal server error.")
# config/settings.py
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
class Config:
env_file = '.env'
extra = "ignore"
class model:
GPT_MODEL: str = "gpt-3.5-turbo-1106"
TEXT_EMBEDDINGS_MODEL: str = "text-embedding-ada-002"
class text_splitter:
CHUNK_SIZE: int = 1000
CHUNK_OVERLAP: int = 100
class lang_chain:
LANGCHAIN_API_KEY: str
LANGCHAIN_TRACING_V2: str
LANGCHAIN_ENDPOINT: str
LANGCHAIN_PROJECT: str
settings = Settings()
ポイント:
CreateQuizUseCase は、ドメインサービス (TextProcessingService) とリポジトリ (QuizRepository, VectorStoreRepository) を利用するように変更。TextProcessingService は、テキスト分割、埋め込み、ベクトルDB への追加、スクレイピングを担当。FAISSVectorStore は VectorStoreRepository インターフェースを実装。Depends を使って、依存オブジェクトを注入。config/settings.py で設定情報を一元管理 (環境変数から読み込む)。infrastructure 層の各クラスは、外部ライブラリ (LangChain, OpenAI, Google Cloud Storage) を利用。models,repositoriesとservicesに__init__.pyを追加これで、Readum アプリケーションの主要な部分のコードが、DDD に基づく構成になりました。各層がそれぞれの責務に集中し、疎結合で変更に強い設計になっていることを確認してください。