Thought for a couple of seconds
以下が Readum の LangGraph(Supervisor 版)実装における 要件まとめ と 実装コード です。
QuizRequest
)
type
: TEXT
or URL
content
: 読書メモ本文 or URLdifficulty
: BEGINNER
/ INTERMEDIATE
/ ADVANCED
questionCount
: 3~20web_crawl(content: str) -> str
:URL→FireCrawl→テキストchunk(text: str) -> List[str]
:固定 CHUNK_SIZE=2000
, CHUNK_OVERLAP=100
で分割embed(chunks: List[str]) -> List[List[float]]
:HuggingFace 埋め込み→FAISS に保存vectors: List[List[float]]
retrieve(vectors: List[List[float]]) -> List[str]
:FAISS検索 (top_k=8
)generate_quiz(chunks: List[str], question_count: int, difficulty: str) -> Quiz
:
.with_structured_output(Quiz)
で Pydantic Quiz
モデルを強制Quiz
インスタンスevaluate_quiz(quiz: Quiz) -> EvaluationResult
:品質チェックfinalize_quiz(quiz: Quiz, difficulty: str) -> QuizResponse
:
QuizResponse(id, preview=DomainQuiz(**quiz.dict()), difficultyValue)
を生成response.json()
を GCS にアップロード → UUID を返却QuizResponse
インスタンス(辞書化して最終出力)preprocess_agent
→ rag_agent
→ finalize_agent
→ END
finalize_agent
内で評価結果 is_sufficient=False
なら最大 5 回まで rag_agent
にループQuizResponse
dict を返すreadum_langgraph.py
)from typing import List, Dict
from uuid import uuid4
from langchain_openai import ChatOpenAI
from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent
from your_models import QuizRequest, QuizResponse, DomainQuiz, Quiz, EvaluationResult
from your_tools import (
FireCrawl,
simple_chunk,
embedding_tool,
faiss_retriever,
rag_chain,
eval_chain,
gcs_uploader,
)
# --- 1. LLM モデル ---
model = ChatOpenAI(model="gpt-4o", temperature=0.0)
# --- 2. ツール関数定義 ---
def web_crawl(content: str) -> str:
# URLの場合は FireCrawl、それ以外はそのまま
return FireCrawl.extract(content)
def chunk(text: str) -> List[str]:
# CHUNK_SIZE=2000, CHUNK_OVERLAP=100
return simple_chunk(text, chunk_size=2000, overlap=100)
def embed(chunks: List[str]) -> List[List[float]]:
vectors = embedding_tool.embed_documents(chunks)
faiss_retriever.add_documents(chunks, vectors)
return vectors
def retrieve(vectors: List[List[float]]) -> List[str]:
return faiss_retriever.get_relevant(vectors, top_k=8)
def generate_quiz(chunks: List[str], question_count: int, difficulty: str) -> Quiz:
# RAG Chain に構造化出力を指定
result: Quiz = rag_chain.invoke({
"context": "\\n".join(chunks),
"question_count": question_count,
"difficulty": difficulty
})
return result
def evaluate_quiz(quiz: Quiz) -> EvaluationResult:
return eval_chain.invoke({"quiz": quiz.dict()})
def finalize_quiz(quiz: Quiz, difficulty: str) -> QuizResponse:
# QuizResponse を組み立て → GCS に保存
response = QuizResponse(
id=str(uuid4()),
preview=DomainQuiz(**quiz.dict()),
difficultyValue=difficulty
)
gcs_uploader.upload_and_get_uuid(response.json())
return response
# --- 3. Agent 定義 ---
preprocess_agent = create_react_agent(
model=model,
tools=[web_crawl, chunk, embed],
name="preprocess_agent",
prompt=(
"You are PreprocessAgent. "
"If input is URL, use web_crawl; then always chunk and embed to FAISS."
)
)
rag_agent = create_react_agent(
model=model.with_structured_output(Quiz),
tools=[retrieve, generate_quiz],
name="rag_agent",
prompt=(
"You are RAGAgent. "
"First retrieve relevant chunks, then generate a Quiz model."
)
)
finalize_agent = create_react_agent(
model=model.with_structured_output(QuizResponse),
tools=[evaluate_quiz, finalize_quiz],
name="finalize_agent",
prompt=(
"You are FinalizeAgent. "
"Evaluate the Quiz; if is_sufficient is false, retry quiz generation. "
"Otherwise finalize and return QuizResponse."
)
)
# --- 4. Supervisor ワークフロー ---
supervisor = create_supervisor(
agents=[preprocess_agent, rag_agent, finalize_agent],
model=model,
prompt=(
"You are a supervisor orchestrating the Readum quiz pipeline: "
"1) preprocess_agent, 2) rag_agent, 3) finalize_agent. "
"Retry up to 5 times if evaluation fails."
)
)
app = supervisor.compile()
# --- 5. 実行サンプル ---
request = QuizRequest(
type="text",
content="ユーザーの読書メモ本文…",
difficulty="intermediate",
questionCount=10
).dict()
result = app.invoke({
"messages": [
{"role": "user", "content": request}
]
})
print(result)
上記で、
with_structured_output(Quiz)
/ (QuizResponse)
) を使い、QuizResponse
dict を返却する流れを実装しています。ぜひご確認ください!何か追加調整やご質問があれば教えてください 😊