🙌

LangGraph×Streamlitで構築する要件定義書作成エージェント―「 RAG・AIエージェント実践入門」を試してみた―

酒井駿

2025/02/23に公開

こんにちは、酒井です！
株式会社EGGHEAD（エッグヘッド）という「製造業で生成AIを活用したシステム開発」をしている会社の代表をしております。

今回は『RAG・AIエージェント実践入門』に掲載されている要件定義書作成エージェントを実際に試してみました。書籍のサンプルコードに加えて、画面上からエージェントを操作できるようにしました。そして、重要なポイントや処理内容を、コード片や図を使って補足解説しました。

本記事が、LangGraphを活用したAIエージェント開発の初学者の方の参考になれば幸いです。

システム概要

最終成果物

GIF動画なのでかなり画像が荒いですが、最終成果物はこちらになります。

また、LangSmithでLLMの実行ログを取っているので、エージェントが実行したLLMのログを取ることができます。

処理のフロー

以下の図に示すフローとなっています。

まずはユーザーの入力した内容から複数の仮想のペルソナと質問をそれぞれ生成します。
次に回答を生成します。
回答が充分かどうかをAIが判断します。
充分だった場合、要件定義書を作成します。

システム構成

レイヤー	技術スタック	役割
フロントエンド	Streamlit	画面を生成
バックエンド	FastAPI	APIエンドポイントを提供
AIエージェント	LangGraph、LangChain	要件定義プロセスの自動化
ログ収集レイヤー	LangSmith	LLMの実行ログの収集

システムの処理フロー

プロジェクト構成

ディレクトリ構造

Google Colabで実行する用なので、ファイルを分けるようになってなかったのですが、今回は画面も含めて動かしたかったので、きちんとファイル分割してプロジェクトのようにしました。

/langgraph_sandbox
├── backend/
│   ├── agents/
│   │   └── documentation_agent.py
│   ├── models/
│   │   ├── evaluation.py
│   │   ├── interview.py
│   │   └── persona.py
│   ├── services/
│   │   ├── information_evaluator.py
│   │   ├── interview_conductor.py
│   │   ├── persona_generator.py
│   │   └── requirements_document_generator.py
│   ├── main.py
│   └── requirements.txt
└── frontend/
    ├── app.py
    └── requirements.txt

バックエンド実装

main.pyファイル

main.pyはごく一般的なエンドポイントです。

main.py
from fastapi import FastAPI, Depends
from langchain_openai import ChatOpenAI
from pydantic import BaseModel
from functools import lru_cache
from dotenv import load_dotenv

from agents.documentation_agent import DocumentationAgent

load_dotenv()

app = FastAPI()


@lru_cache()
def get_llm():
    return ChatOpenAI()


def get_documentation_agent(llm: ChatOpenAI = Depends(get_llm)) -> DocumentationAgent:
    return DocumentationAgent(llm)


class UserRequest(BaseModel):
    request: str


@app.post("/generate_requirements")
async def generate_requirements(
    user_request: UserRequest,
    documentation_agent: DocumentationAgent = Depends(get_documentation_agent),
):
    requirements_doc = documentation_agent.run(user_request.request)
    return {"requirements_doc": requirements_doc}

エージェントを実行するファイル

こちらがワークフローを実行していくファイルです。StateGraphで初期化してノードを追加していくのが特徴です。

workflow.add_nodeの第一引数にノード名、第二引数にstateを引数に持つ関数を指定します。この関数はstateのプロパティをキーとする辞書を返していて、これでstateが更新されるようになっています。

同様にノードを追加していき、ワークフローを構成していきます。

また、workflow.add_conditional_edgesでは分岐をさせていて、まだインタビュー結果が充分でない場合はプルソナ生成、質問生成、インタビュー、結果が十分かどうかの評価を行い、充分な場合は要件定義書を生成しています。

agents/documentation_agent.py
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import Optional

from services.persona_generator import PersonaGenerator
from services.interview_conductor import InterviewConductor
from services.requirements_document_generator import (
    RequirementsDocumentGenerator,
)
from services.information_evaluator import InformationEvaluator
from models.interview import InterviewState, InterviewResult


class DocumentationAgent:
    def __init__(self, llm: ChatOpenAI, k: Optional[int] = None):
        self.persona_generator = PersonaGenerator(llm, k=5)
        self.interview_generator = InterviewConductor(llm)
        self.information_evaluator = InformationEvaluator(llm)
        self.requirements_document_generator = RequirementsDocumentGenerator(llm)

        self.graph = self._create_graph()

    def _create_graph(self) -> StateGraph:
        workflow = StateGraph(InterviewState)

        workflow.add_node("generate_personas", self._generate_personas)
        workflow.add_node("conduct_interviews", self._conduct_interviews)
        workflow.add_node("evaluate_information", self._evaluate_information)
        workflow.add_node("generate_requirements", self._generate_requirements)

        # これは workflow.add_edge(START, "agent") と同等
        workflow.set_entry_point("generate_personas")

        workflow.add_edge("generate_personas", "conduct_interviews")
        workflow.add_edge("conduct_interviews", "evaluate_information")

        workflow.add_conditional_edges(
            "evaluate_information",
            lambda state: not state.is_information_sufficient and state.iteration < 5,
            {True: "generate_personas", False: "generate_requirements"},
        )

        workflow.add_edge("generate_requirements", END)

        return workflow.compile()

    def _generate_personas(self, state: InterviewState) -> dict:
        new_personas = self.persona_generator.run(state.user_request)
        return {
            "personas": new_personas.personas,
            "iteration": state.iteration + 1,
        }

    def _conduct_interviews(self, state: InterviewState) -> dict:
        new_interviews: InterviewResult = self.interview_generator.run(
            state.user_request,
            state.personas,
        )
        return {"interviews": new_interviews.interviews}

    def _evaluate_information(self, state: InterviewState) -> dict:
        evaluation_result = self.information_evaluator.run(
            state.user_request,
            state.interviews,
        )
        return {
            "is_information_sufficient": evaluation_result.is_sufficient,
            "reason": evaluation_result.reason,
        }

    def _generate_requirements(self, state: InterviewState) -> dict:
        requirements_doc = self.requirements_document_generator.run(
            state.user_request,
            state.interviews,
        )
        return {"requirements_doc": requirements_doc}

    def run(self, user_request: str) -> str:
        initial_state = InterviewState(user_request=user_request)
        final_state = self.graph.invoke(initial_state)

        return final_state["requirements_doc"]

モデルレイヤー

evaluation.py

評価結果を格納するモデルです。

evaluation.py
from pydantic import BaseModel, Field


class EvaluationResult(BaseModel):
    reason: str = Field(..., description="理由")
    is_sufficient: bool = Field(..., description="情報が十分かどうか")

interview.py

こちらにはインタビューに関係するモデルを入れています。

interview.py
from pydantic import BaseModel, Field
from .persona import Persona


class Interview(BaseModel):
    persona: Persona = Field(..., description="人物")
    question: str = Field(..., description="質問")
    answer: str = Field(..., description="回答")


class InterviewResult(BaseModel):
    interviews: list[Interview] = Field(
        default_factory=list, description="インタビュー"
    )


class InterviewState(BaseModel):
    user_request: str = Field(..., description="ユーザーのリクエスト")
    personas: list[Persona] = Field(
        default_factory=list, description="生成されたペルソナのリスト"
    )
    interviews: list[Interview] = Field(
        default_factory=list, description="生成されたインタビューのリスト"
    )
    requirements_doc: str = Field(default="", description="生成された要件定義書")
    iteration: int = Field(
        default=0, description="ペルソナ生成とインタビューの反復回数"
    )
    is_information_sufficient: bool = Field(
        default=False, description="情報が十分かどうか"
    )

persona.py

ペルソナの情報を格納するモデルです。より詳細にペルソナの情報を定義したい場合は、ここにage(年齢)などを追加します。
ペルソナの解像度は高ければ高いほどいいので、可能であれば年齢や職業、年収なども入れるといいでしょう。

persona.py
from pydantic import BaseModel, Field


class Persona(BaseModel):
    name: str = Field(..., description="名前")
    background: str = Field(..., description="背景")


class Personas(BaseModel):
    personas: list[Persona] = Field(default_factory=list, description="人物")

サービスレイヤー

information_evaluator.py

ポイントはllm.with_structured_output(EvaluationResult)になっていて、こちらの引数でPydanticのモデルを指定することでchain.invokeの戻り値をPydanticのモデルインスタンスにすることができます。これはめちゃくちゃ便利です。

information_evaluator.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from models.interview import Interview
from models.evaluation import EvaluationResult


class InformationEvaluator:
    def __init__(self, llm: ChatOpenAI):
        self.llm = llm.with_structured_output(EvaluationResult)

    def run(self, user_request: str, interviews: list[Interview]) -> EvaluationResult:
        prompt = ChatPromptTemplate.from_messages(
            [
                (
                    "system",
                    "あなたは包括的な要件文書を作成するための情報の十分製を評価する専門家です。",
                ),
                (
                    "human",
                    "以下のユーザーリクエストとインタビュー結果に基づいて、包括的な要件定義書を作成するのに十分な情報が集まったかどうかを判断してください。\n\n"
                    "ユーザーリクエスト：{user_request}\n"
                    "インタビュー結果：{interview_results}",
                ),
            ]
        )

        chain = prompt | self.llm

        return chain.invoke(
            {
                "user_request": user_request,
                "interview_results": "\n".join(
                    f"ペルソナ: {i.persona.name} - {i.persona.background}\n"
                    f"質問: {i.question}\n"
                    f"回答: {i.answer}\n"
                    for i in interviews
                ),
            }
        )

interview_conductor.py

こちらでは、質問の生成、ペルソナからの回答の生成を行います。chain.batchを使うことで、並列でのAPI呼び出しをしていて、実行時間の短縮が行えます。
並列処理は重要で、エージェントはLLMの呼び出しが多くなりユーザー体験が悪くなりがちです。なので、可能な限り並列で実行するのがいいでしょう。

persona_generator.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from models.persona import Persona
from models.interview import Interview, InterviewResult


class InterviewConductor:
    def __init__(self, llm: ChatOpenAI):
        self.llm = llm

    def run(self, user_request: str, personas: list[Persona]) -> InterviewResult:
        questions = self._generate_questions(
            user_request=user_request, personas=personas
        )

        answers = self._generate_answers(
            personas=personas,
            questions=questions,
        )

        interviews = self._generate_interviews(
            personas=personas, questions=questions, answers=answers
        )

        return InterviewResult(interviews=interviews)

    def _generate_questions(
        self, user_request: str, personas: list[Persona]
    ) -> list[str]:
        question_prompt = ChatPromptTemplate.from_messages(
            [
                (
                    "system",
                    "あなたはユーザー要件に基づいて適切な質問を生成する専門家です。",
                ),
                (
                    "human",
                    "以下のペルソナに関連するユーザーリクエストについて、１つ質問を生成してください。\n\n"
                    "ユーザーリクエスト：{user_request}\n"
                    "ペルソナ：{persona_name} - {persona_background}\n\n"
                    "質問は具体的で、このペルソナの視点から重要な情報を引き出すように設計してください。",
                ),
            ]
        )

        question_chain = question_prompt | self.llm | StrOutputParser()

        question_queries = [
            {
                "persona_name": persona.name,
                "persona_background": persona.background,
                "user_request": user_request,
            }
            for persona in personas
        ]

        result = question_chain.batch(question_queries)
        return result

    def _generate_answers(
        self, personas: list[Persona], questions: list[str]
    ) -> list[str]:
        answer_prompt = ChatPromptTemplate.from_messages(
            [
                (
                    "system",
                    "あなたは以下のペルソナとして回答しています：{persona_name} - {persona_background}",
                ),
                ("human", "質問：{question}\n"),
            ]
        )

        answer_chain = answer_prompt | self.llm | StrOutputParser()

        answer_queries = [
            {
                "persona_name": persona.name,
                "persona_background": persona.background,
                "question": question,
            }
            for persona, question in zip(personas, questions)
        ]

        return answer_chain.batch(answer_queries)

    def _generate_interviews(
        self, personas: list[Persona], questions: list[str], answers: list[str]
    ) -> list[Interview]:
        return [
            Interview(persona=persona, question=question, answer=answer)
            for persona, question, answer in zip(personas, questions, answers)
        ]

persona_generator.py

こちらでは、ペルソナの生成を行っています。デフォルトでは5人生成しています。ここでもllm.with_structured_outputを使って、戻り値をPydanticのモデルインスタンスにしています。

persona_generator.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from models.persona import Personas
from langchain_core.output_parsers import PydanticOutputParser


class PersonaGenerator:
    def __init__(self, llm: ChatOpenAI, k: int = 5):
        self.llm = llm.with_structured_output(Personas)
        self.k = k

    def run(self, user_request: str) -> Personas:
        prompt = ChatPromptTemplate.from_messages(
            [
                (
                    "system",
                    "あなたはユーザーインタビュー用の多様なペルソナを作成する専門家です。",
                ),
                (
                    "human",
                    f"以下のユーザーリクエストに関するインタビュー用に、{self.k}人の多様なペルソナを作成してください。\n\n"
                    f"ユーザーリクエスト: {user_request}\n\n"
                    "各ペルソナには名前と簡単な背景を含めてください。年齢、性別、職業、技術的専門知識において多様性を確保してください。",
                ),
            ]
        )

        chain = prompt | self.llm
        return chain.invoke({"user_request": user_request})

requirements_document_generator.py

最後に要件定義書の生成するところです。ペルソナへのインタビュー情報を考慮した要件定義書を作成します。この例ではかなり簡易的なものになっているので、もっと良くしたい場合はここのプロンプトを編集することになります。

persona_generator.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

from models.interview import Interview


class RequirementsDocumentGenerator:
    def __init__(self, llm: ChatOpenAI):
        self.llm = llm

    def run(self, user_request: str, interviews: list[Interview]) -> str:
        prompt = ChatPromptTemplate.from_messages(
            [
                (
                    "system",
                    "あなたは収集した情報に基づいて要件文書を作成する専門家です。",
                ),
                (
                    "human",
                    "以下のユーザーリクエストと複数のペルソナからのインタビュー結果に基づいて、要件定義書を作成してください。\n\n"
                    "ユーザーリクエスト：{user_request}\n\n"
                    "インタビュー結果：\n{interview_results}\n"
                    "要件定義書には以下のセクションを含めてください:\n"
                    "1. プロジェクト概要\n"
                    "2. 主要機能\n"
                    "3. 非機能要件\n"
                    "4. 制約条件\n"
                    "5. ターゲットユーザー\n"
                    "6. 優先順位\n"
                    "7. リスクと軽減策\n"
                    "出力は必ず日本語でお願いします。\n\n要件定義書：",
                ),
            ]
        )

        chain = prompt | self.llm | StrOutputParser()

        return chain.invoke(
            {
                "user_request": user_request,
                "interview_results": "\n".join(
                    f"ペルソナ: {i.persona.name} - {i.persona.background}\n"
                    f"質問: {i.question}\n"
                    f"回答: {i.answer}\n"
                    for i in interviews
                ),
            }
        )

フロントエンド実装

フロントエンドは簡易的なものを作るためにStreamlitを使いました。CSSも書かずにこれだけの記述でデモ画面が作れるのはいいです。地味に凝ったUIにも出来てモック的に見せるには十分だと思っています。

https://streamlit.io/

app.py
import streamlit as st
import requests

st.title("要件定義書作成アシスタント")

st.write("プロジェクトの要件を自動で分析・整理し、要件定義書を作成します。")

user_request = st.text_area(
    "プロジェクトの概要を入力してください",
    height=200,
    placeholder="例: オンライン予約システムを作りたい。ユーザーは予約日時を選択でき、管理者は予約状況を確認できるようにしたい。"
)

if st.button("要件定義書を作成"):
    if user_request:
        with st.spinner("要件定義書を作成中..."):
            try:
                response = requests.post(
                    "http://localhost:8000/generate_requirements",
                    json={"request": user_request}
                )
                if response.status_code == 200:
                    requirements_doc = response.json()["requirements_doc"]
                    st.success("要件定義書が作成されました！")
                    st.text_area("作成された要件定義書", value=requirements_doc, height=400)
                else:
                    st.error("エラーが発生しました。もう一度お試しください。")
            except requests.exceptions.ConnectionError:
                st.error("バックエンドサーバーに接続できません。サーバーが起動していることを確認してください。")
    else:
        st.warning("プロジェクトの概要を入力してください。")

セットアップ

バックエンド環境構築

まずは、環境変数をセットします。

LANGSMITH_TRACING="true"
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="*************"
LANGSMITH_PROJECT="Project Name"
OPENAI_API_KEY="************"

以下コマンドでサーバーを起動します。

cd backend
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
uvicorn main:app --reload

フロントエンド環境構築

以下コマンドでフロントエンドを起動します。

cd frontend
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
streamlit run app.py

まとめ

最後まで読んでくださり、ありがとうございました！

このコードを書いていて感じたのはAIエージェントを開発する際には、処理のワークフローを明確に定義することがとても重要であると再認識しました。特に複雑な業務をAIエージェントに落とし込むには、ワークフローを可視化してシンプルなものにすることが重要で、それをきちんと行うには業務理解が必要になると感じました。

私は車部品メーカーで3年間勤めていたバックグラウンドがあり業務理解があります。もしこの記事を読んでいる製造業の方で生成AIを使って業務改善したいと思っている方がいらっしゃったらお気軽に下のフォームから問い合わせいただけると幸いです。

EGGHEAD テックブログPublication

製造業を中心に生成AIを使った業務改善のシステム開発を行っています。

Discussion

ログインするとコメントできます