Thought for a couple of seconds

以下が Readum の LangGraph（Supervisor 版）実装における 要件まとめ と 実装コード です。

要件まとめ

  1. 入力モデル (QuizRequest)
  2. 前処理 (PreprocessAgent)
  3. RAG 処理 (RAGAgent)
  4. 評価 ＆ 最終化 (FinalizeAgent)
  5. Supervisor ワークフロー

実装コード例 (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)

上記で、

ぜひご確認ください！何か追加調整やご質問があれば教えてください 😊