LangChainとLangGraphで実現！反復型ペルソナ・インタビューによる要件定義書・仕様書自動生成ツールの実装解説

本記事は、書籍「LangChainとLangGraphによるRAG・AIエージェント［実践］入門」

の10章を参考にして、その内容を拡張したものです。
拡張内容としては、ユーザーのリクエストから自動で要件定義書や仕様書を作り出すまでを一気通貫でコーディングしたものです。LangChainなどの詳細な解説は本書をご参照ください。

また、上記書籍に関しては、例えば以下の記事においてしっかりまとめられておりますので、参考にしてください。

初めてLangChainに触れる方でも理解しやすいよう、各工程の背景や仕組み、コード内の各関数がどのような役割を果たしているのかを、できるだけ具体的に丁寧に解説していきます。

⚠ 注意
この記事は個人で作成したものであり、内容や意見は所属企業・部門見解を代表するものではありません。

---

環境構築

本システムはGoogle Colab上での動作を前提としています。以下のコードをColabのセルにそのままコピー＆ペーストして実行してください。

• 注：今後出てくるファイルパスやAPIキー該当部分は筆者環境にあわせているので、お試しになる場合はご自分の環境に合わせて適宜修正してください。

これにより、Google Driveへのマウント、必要なリポジトリのクローン、依存ライブラリのインストール、さらに各APIキーやLangChainの設定を環境変数として登録することができます。

from google.colab import drive
drive.mount('/content/drive')

!git clone https://github.com/GenerativeAgents/agent-book.git

!pip install langchain-core==0.3.0 langchain-openai==0.2.0 langgraph==0.2.22 python-dotenv==1.0.1

import os
from google.colab import userdata

# OpenAI API キー
os.environ["OPENAI_API_KEY"] = userdata.get("OPENAI_API_KEY")

# Gemini (Google Generative AI) API キー
os.environ["GOOGLE_API_KEY"] = userdata.get("GOOGLE_API_KEY")

# LangChainの設定
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
os.environ["LANGCHAIN_API_KEY"] = userdata.get("LANGCHAIN_API_KEY")
os.environ["LANGCHAIN_PROJECT"] = "agent-book"

これで必要な環境のセットアップは完了です。あとは、用意された各モジュールをファイルに保存し、下記のコマンドを実行するだけで、本システムが自動的に処理を開始し、要件定義書や仕様書の生成が行われます。

実行方法

APPENDIXの全コードも設置の上で、以下のコマンドで実行してください:

%cd /content/drive/MyDrive/AIエージェント 
!python generate_specifications.py --task "○○県職員採用ページを改良して採用試験への応募者数を増やしたい"#プロンプトを入力

---

動作デモ：実際の出力例

以下は、本システムを実行した際のサンプルコマンドと、その出力結果の一例です。
（実際の実行環境によって、日付やタイムスタンプ、文面などは異なる場合があります）

---

1. コマンド例

%cd /content/drive/MyDrive/AIエージェント
!python generate_specifications.py --task "県立動物園ページのHPを改良して来場者を増やしたい"

このコマンドにより、

1. 要件定義書
2. 仕様書
3. 契約書添付用仕様書
   が順番に自動生成され、Googleドライブに保存されます。

---

2. 出力結果イメージ

/content/drive/MyDrive/AIエージェント
🔹 要件定義書を生成中...
🔹 要件定義書を元にした仕様書を生成中...
🔹 契約書添付用仕様書を生成中...

=== 契約書添付用仕様書 ===

はい、承知いたしました。以下に、契約書添付用仕様書案を作成します。契約書に添付することを想定し、より詳細で明確な記述を心がけました。

---

**契約書添付用仕様書**

**○○県立動物園ホームページ改良プロジェクト**

**version 1.0**
**作成日：2023年12月28日**

## 1. プロジェクト概要
### 1.1. プロジェクト名
○○県立動物園ホームページ改良プロジェクト

### 1.2. 目的
...（以下省略）...

---

1. システム全体の概要

これまで、要件定義書や仕様書の作成は、ヒアリングや手作業に頼っていたため、情報の抜け落ちや作成工数が大きな課題でした。

そこで本システムでは、以下のプロセスを自動化することで、効率的かつ高精度な文書生成を実現しています。

ペルソナ生成

• ユーザーのリクエストに合わせ、例えば「マーケティング担当」「システムエンジニア」「広報担当」など、さまざまな視点を持つキャラクター（ペルソナ）を自動で作成します。
• 各ペルソナには名前や背景が設定され、後のインタビューで多面的な情報収集に活用されます。

インタビュー実施

• 作成された各ペルソナごとに、ユーザーリクエストに基づいた質問が自動生成され、その回答を収集します。
• こうして得られた多角的な意見・情報を統合することで、より充実したドキュメントが生成されます。

情報評価と反復処理

• インタビュー結果が十分かどうかを自動で評価し、不足している場合は設定した上限回数内で再度情報収集を行います。
• この反復処理により、常に最新で十分な情報をもとに文書を作成できます。

文書自動生成

• 十分な情報が集まると、まずは要件定義書が作成され、次にその内容を基に仕様書が生成されます。
• さらに、Google Generative AI（Gemini API）を利用して、契約書に添付可能な形に仕上げます。

---

2. システム全体のプロセスフロー

以下のMermaid図は、ユーザーがリクエストを入力してから最終的な仕様書が完成するまでの全体の流れを示しています。

どの工程でも、情報が足りない場合は自動で反復処理が行われる仕組みになっていますので、安心して高品質な文書を生成できます。

---

各工程の詳細

ユーザー要求入力 (A)

ユーザーは具体的なリクエスト（例：「○○県職員採用ページの改良」など）を入力します。
この入力がシステム全体の出発点となり、後続の処理に引き渡されます。

要件定義書生成 (B～B2)

• 最初に、システムはユーザーリクエストに基づいたペルソナを自動生成し、各ペルソナに質問を投げかけて回答を収集します。
• 収集された情報が十分でない場合は、同じプロセスが自動で再実行され、必要な情報が集まるまでループします。
• 十分な情報が得られたら、統合された回答を元に要件定義書が完成し、保存されます。

仕様書生成 (C～C2)

• 完成した要件定義書をもとに、さらに詳細な仕様書が生成されます。
• こちらもペルソナ生成やインタビュー、情報評価の反復処理を通じて、実務に直結した具体的な文書を自動生成します。

契約書添付用仕様書生成 (D)

• 生成された仕様書をさらにブラッシュアップし、契約書に添付できる形式に整形します。
• 特に、業務範囲や技術仕様、スケジュールなどの項目が強化され、誤解を防ぐために簡潔で明確な文章が生成されます。

最終仕様書作成完了 (E)

すべての工程が完了し、最終的な仕様書が出力されます。

---

3. 各モジュールの実装詳細と設計のポイント

ここでは、各モジュールの主要な関数やクラスについて、さらに具体的に丁寧に解説します。

3.1 generate_specifications.py

このファイルは、システム全体の統括モジュールとして動作します。以下のポイントを押さえています。

環境設定

• 最初にGoogle Colab環境かどうかを判断し、必要ならGoogle Driveをマウントします。また、.envファイルからAPIキーなどを読み込みます。
• これにより、Colab環境とローカル環境の両方で柔軟に動作できるように工夫されています。

要件定義書生成

• get_requirements_definition() 関数は、subprocessを利用して documentation_agent/main.py を実行します。
• ユーザーリクエストが引数として渡され、生成された要件定義書は標準出力から取得され、タイムスタンプ付きファイルとして保存されます。
• エラーが発生した場合は、詳細なエラーメッセージを出力して処理を中断します。

仕様書生成

• get_specification() 関数は、先ほどの要件定義書と元のリクエストを統合し、aiエージェントフォルダ/main.py を実行して仕様書を生成します。
• 仕様書も同様にタイムスタンプ付きでファイル保存され、後続の処理で利用されます。

契約書添付用仕様書生成

• generate_announcement() 関数は、Gemini APIを呼び出して、生成された仕様書を契約書に添付できる形式に整形します。
• プロンプトには、業務範囲や技術仕様、スケジュールの強化指示が含まれており、実務で使える高精度な文書を作り出します。

3.2 documentation_agent/main.py

このモジュールは、要件定義書生成エージェントとしての役割を担っています。

PersonaGenerator

• ユーザーリクエストに応じて、複数のペルソナ（キャラクター）を自動生成します。
• プロンプトテンプレートでは、名前、背景、年齢、性別、職業などの多様な情報を求める指示が含まれており、後のインタビューで多角的な情報を得るための基盤となります。

InterviewConductor

• 生成されたペルソナに対して、ユーザーリクエストに基づいた具体的な質問を自動生成し、その回答を収集します。
• 質問生成は、各ペルソナの特徴をプロンプトに含めることで、具体性と深みのある質問を実現しており、回答生成もバッチ処理で効率的に行われます。

InformationEvaluator

• インタビュー結果を統合し、文書作成に必要な情報が十分かどうかを自動評価します。
• この評価結果（is_sufficient とその理由）は、情報が不足している場合の反復処理の判断に利用されます。

RequirementsDocumentGenerator

• 十分な情報が収集された段階で、各ペルソナから得たインタビュー結果を統合し、具体的な要件定義書を作成します。
• プロンプトには、事業概要、主要機能、非機能要件など、必要なセクションが詳細に指示されており、日本語で明確な文章が生成されます。

反復処理

• StateGraphを利用して、情報が不十分な場合に自動的に「ペルソナ生成→インタビュー→情報評価」のループが実行される仕組みになっています。
• 反復回数に上限を設けることで、無限ループを防止しています。

3.3 aiエージェントフォルダ/main.py

こちらのモジュールは、仕様書生成エージェントとして機能します。

• 要件定義書を入力として、さらに詳細な仕様書を自動生成するために、同様のペルソナ生成・インタビュー・情報評価の処理が実施されます。

仕様書生成プロセス

• documentation_agent と同様の流れで、ペルソナ生成、インタビュー、情報評価が実施されますが、仕様書生成に特化したプロンプトが用意されています。
• これにより、事業概要、目的、業務範囲、技術的要件、スケジュール、予算など、実務に直結する詳細な文書が自動生成されます。

プロンプト設計

• 仕様書生成用のプロンプトは、各セクションごとに具体的な説明を求める指示が含まれており、生成された文章が実際の業務でそのまま利用できるように工夫されています。

---

4. コードの分割と一斉実行の考え方

プログラミング初心者の方には、コードが複数のファイルに分かれていること、そしてそれらが一斉に動くという感覚が分かりにくいかもしれません。

本システムは、以下のような考え方で設計されています。

モジュールごとの役割分担

• 各ファイルは、システム全体の中で明確な役割を持っています。
  • generate_specifications.py は、全体の統括と各エージェントの実行を管理します。
  • documentation_agent/main.py は、要件定義書の生成に関する処理を担当します。
  • aiエージェントフォルダ/main.py は、仕様書の生成に特化しています。
• このように役割を分けることで、各部分を個別に開発・テストでき、全体としての再利用性や保守性が向上します。

統合実行の仕組み

• メインとなる generate_specifications.py が、Pythonのsubprocessモジュールを使って、他のファイル（モジュール）を実行します。
• つまり、各ファイルは独立した処理単位として動作し、その結果を標準出力やファイルに出力します。
• generate_specifications.py はそれらをまとめ、システム全体としての処理結果（要件定義書や仕様書）を得るためのオーケストレーターとして働きます。

全体としての一斉実行

• それぞれのモジュールは、個別に動作するため、初心者の方には「ファイルがバラバラでどうやって一緒に動くのか？」と感じられるかもしれませんが、実際には generate_specifications.py が各モジュールを順番に呼び出し、結果を連携させることで、一連の流れが一斉に実行される仕組みになっています。
• この設計により、システム全体をモジュール化し、分かりやすく管理しやすいコードベースを実現しています。

---

5. 実装上のポイントと注意事項

エラーハンドリング

• 各モジュールではtry-exceptブロックを使い、実行中のエラー（subprocessのエラー、API呼び出し時のエラーなど）を適切にキャッチしています。
• これにより、実際の運用時に問題が発生した際の原因究明が容易になります。

環境構築

• Google Colab環境でもローカル環境でも動作するよう、Driveのマウントや出力先の設定が工夫されています。
• また、上記「環境構築」セクションで紹介したコードを参考に、必要なライブラリのインストールや環境変数の設定を行ってください。

タイムスタンプ付きファイル出力

• 出力される各文書は、生成日時が分かるタイムスタンプ付きのファイル名で保存されるため、過去の出力結果を管理しやすくなっています。

反復処理による精度向上

• ペルソナ生成、インタビュー、情報評価のループ処理により、必要な情報が十分に集まるまで自動で再実行され、品質の高い要件定義書や仕様書が作成されます。

---

まとめ

この記事では、書籍「LangChainとLangGraphによるRAG・AIエージェント［実践］入門」10章を参考に、LangChainとLangGraphを活用してユーザーのリクエストから自動で要件定義書や仕様書を作成するツールの実装例を、分かりやすく丁寧に解説しました。

各モジュールの背景や各関数の役割、反復処理やエラーハンドリングのポイント、そしてGoogle Colab環境での環境構築手順についても詳しく説明しています。

さらに、コードが複数のファイルに分割されている構成と、それらを統合して一斉に実行する仕組みについても丁寧に解説しました。

ぜひ、この実装例を参考にして、LangChainやLangGraphの面白さ、自動生成ツールの可能性に触れてみてください！

プログラミング初心者が試しながら手探りで作成した記事ですので、コードが冗長で読みにくい点もあろうかと思います。本記事にお気づきの点や何か誤りがあった場合はぜひコメントで教えていただきますと幸いです。

---

Appendix: コード全文

以下に、本システムの各ファイルの フルコード を掲載します。

このコードを実際に動かしながら、LangChain や LangGraph を活用した 自動生成ツールの動作を確認 してみてください。

📌 構成:

1. generate_specifications.py（システム全体を統括するメインスクリプト）
2. documentation_agent/main.py（要件定義書生成エージェント）
3. aiエージェントフォルダ/main.py（仕様書生成エージェント）

🔹 使い方:

1. 記事の解説を参考に環境をセットアップ

%cd /content/drive/MyDrive/AIエージェント 
!python generate_specifications.py --task "○○県職員採用ページを改良して採用試験への応募者数を増やしたい"#プロンプトを入力

　 を実行
3. 生成された要件定義書・仕様書を確認！

1. /content/drive/MyDrive/AIエージェント/generate_specifications.py

# -*- coding: utf-8 -*- 
"""generate_specifications"""

import subprocess
import os
import google.generativeai as genai
import argparse
from dotenv import load_dotenv
import datetime

# Google Colab 環境かどうか判定して Drive をマウント
try:
    # Colab 環境なら get_ipython() が存在する
    if "google.colab" in str(get_ipython()):
        from google.colab import drive
        drive.mount('/content/drive')
        SAVE_DIR = "/content/drive/MyDrive/AIエージェント/"
    else:
        SAVE_DIR = "./output/"
except Exception as e:
    SAVE_DIR = "./output/"

# SAVE_DIR が存在しない場合は作成
os.makedirs(SAVE_DIR, exist_ok=True)

# Pandoc のインストール（今回は使用しないので、ここはコメントアウトまたは削除）
# try:
#     subprocess.run(["apt-get", "install", "-y", "pandoc"], check=True)
# except Exception as e:
#     print("Pandoc のインストールに失敗しました。ローカル環境の場合は手動でインストールしてください。", e)

# .env から環境変数を読み込む
load_dotenv()
API_KEY = os.getenv("GOOGLE_API_KEY")

# Google Generative AI の設定
genai.configure(api_key=API_KEY)

# タイムスタンプ生成関数
def generate_timestamp():
    now = datetime.datetime.now()
    return now.strftime("%Y%m%d%H%M")

# 要件定義書の取得
def get_requirements_definition(user_task):
    try:
        result = subprocess.run(
            [
                "python", "-m", "documentation_agent.main",
                "--task", user_task,
                "--k", "5"
            ],
            cwd="/content/agent-book/chapter10",  # モジュールがあるディレクトリを指定
            capture_output=True,
            text=True,
            check=True
        )
        requirements_text = result.stdout.strip()

        # タイムスタンプ付きファイル名生成（.txt形式に変更）
        timestamp = generate_timestamp()
        file_name = f"requirements_definition_{timestamp}.txt"
        file_path = os.path.join(SAVE_DIR, file_name)

        # テキスト形式で保存
        with open(file_path, "w", encoding="utf-8") as f:
            f.write(requirements_text)

        return requirements_text
    except subprocess.CalledProcessError as e:
        print(f"要件定義書生成エラー: {e}\n詳細: {e.stderr}")
        return None

# 仕様書の取得（要件定義書を元に作成）
def get_specification(requirements_text, user_task):
    task_with_requirements = f"{user_task}\n\n【要件定義書】\n{requirements_text}"

    try:
        result = subprocess.run(
            [
                "python", "/content/drive/MyDrive/AIエージェント/main.py",
                "--task", task_with_requirements,
                "--k", "5"
            ],
            capture_output=True,
            text=True,
            check=True
        )
        spec_text = result.stdout.strip()

        # タイムスタンプ付きファイル名生成（.txt形式に変更）
        timestamp = generate_timestamp()
        file_name = f"detailed_specification_{timestamp}.txt"
        file_path = os.path.join(SAVE_DIR, file_name)

        # テキスト形式で保存
        with open(file_path, "w", encoding="utf-8") as f:
            f.write(spec_text)

        return spec_text
    except subprocess.CalledProcessError as e:
        print(f"仕様書生成エラー: {e}\n詳細: {e.stderr}")
        return None

# Geminiに渡して契約書添付用の仕様書を生成
def generate_announcement(spec_text, user_task):
    gemini_model = "gemini-2.0-flash-thinking-exp-01-21"

    prompt = f"""
    以下の仕様書を基に、○○県として契約書に添付できる分量と精度に肉付けして契約書添付用仕様書を作成してください。
    - **契約書への添付を想定した形式**で作成
    - **業務範囲・技術仕様・スケジュールを強化**
    - **誤解を防ぐため、簡潔かつ明確に**
    - **適切な見出しや箇条書きを使用**
    - **出力は必ず日本語で記述してください。
    元の業務内容:
    {user_task}

    仕様書:
    {spec_text}

    """

    model = genai.GenerativeModel(gemini_model)

    try:
        response = model.generate_content(prompt)
        announcement_text = response.text.strip()

        # タイムスタンプ付きファイル名生成（.txt形式に変更）
        timestamp = generate_timestamp()
        file_name = f"announcement_specification_{timestamp}.txt"
        file_path = os.path.join(SAVE_DIR, file_name)

        # テキスト形式で保存
        with open(file_path, "w", encoding="utf-8") as f:
            f.write(announcement_text)

        print("\n=== 契約書添付用仕様書 ===\n")
        print(announcement_text)
    except Exception as e:
        print(f"Gemini API エラー: {e}")

# メイン処理の実行
def main():
    parser = argparse.ArgumentParser(description="仕様書自動生成ツール")
    parser.add_argument("--task", type=str, required=True, help="委託事業の内容")
    args = parser.parse_args()

    user_task = args.task

    print("🔹 要件定義書を生成中...")
    requirements_text = get_requirements_definition(user_task)
    if not requirements_text:
        print("要件定義書の取得に失敗しました。")
        return

    print("🔹 要件定義書を元にした仕様書を生成中...")
    spec_text = get_specification(requirements_text, user_task)
    if not spec_text:
        print("仕様書の取得に失敗しました。")
        return

    print("🔹 契約書添付用仕様書を生成中...")
    generate_announcement(spec_text, user_task)
    
    # 以下で、生成された要件定義書と仕様書も表示する
    print("\n=== 仕様書 ===\n")
    print(spec_text)
    print("\n=== 要件定義書 ===\n")
    print(requirements_text)

    
    print("\n✅ すべての仕様書が生成され、Googleドライブに保存されました！")

if __name__ == "__main__":
    main()

2. /content/agent-book/chapter10/documentation_agent/main.py

import operator
from typing import Annotated, Any, Optional

from dotenv import load_dotenv
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import END, StateGraph
from pydantic import BaseModel, Field

# .envファイルから環境変数を読み込む
load_dotenv()


# ペルソナを表すデータモデル
class Persona(BaseModel):
    name: str = Field(..., description="ペルソナの名前")
    background: str = Field(..., description="ペルソナの持つ背景")


# ペルソナのリストを表すデータモデル
class Personas(BaseModel):
    personas: list[Persona] = Field(
        default_factory=list, description="ペルソナのリスト"
    )


# インタビュー内容を表すデータモデル
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 EvaluationResult(BaseModel):
    reason: str = Field(..., description="判断の理由")
    is_sufficient: bool = Field(..., description="情報が十分かどうか")


# 要件定義生成AIエージェントのステート
class InterviewState(BaseModel):
    user_request: str = Field(..., description="ユーザーからのリクエスト")
    personas: Annotated[list[Persona], operator.add] = Field(
        default_factory=list, description="生成されたペルソナのリスト"
    )
    interviews: Annotated[list[Interview], operator.add] = 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="情報が十分かどうか"
    )


# ペルソナを生成するクラス
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"
                    "ユーザーリクエスト: {user_request}\n\n"
                    "各ペルソナには名前と簡単な背景を含めてください。年齢、性別、職業、技術的専門知識において多様性を確保してください。",
                ),
            ]
        )
        # ペルソナ生成のためのチェーンを作成
        chain = prompt | self.llm
        # ペルソナを生成
        return chain.invoke({"user_request": user_request})


# インタビュー実施クラス
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._create_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",
                    "以下の要求内容と関係者情報に基づき、関係者視点で重要かつ詳細な情報を引き出す質問を1つ生成してください。\n\n"
                    "要求内容: {user_request}\n"
                    "関係者: {persona_name} - {persona_background}\n\n"
                    "質問:"
                ),
            ]
        )
        # 質問生成のためのチェーンを作成
        question_chain = question_prompt | self.llm | StrOutputParser()
        # 各ペルソナに対する質問クエリを作成
        question_queries = [
            {
                "user_request": user_request,
                "persona_name": persona.name,
                "persona_background": persona.background,
            }
            for persona in personas
        ]
        # 質問をバッチ処理で生成
        return question_chain.batch(question_queries)

    def _generate_answers(self, personas: list[Persona], questions: list[str]) -> list[str]:
        # 回答生成のためのプロンプトを定義
        answer_prompt = ChatPromptTemplate.from_messages(
            [
                (
                    "system",
                    "あなたは、以下のペルソナとして回答しています: {persona_name} - {persona_background}",
                ),
                ("human", "質問: {question}"),
            ]
        )
        # 回答生成のためのチェーンを作成
        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 _create_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)
        ]


# 情報の十分性評価クラス
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\n"
                    "インタビュー結果:\n{interview_results}\n\n"
                    "十分であれば 'true'、不足している場合は 'false' と、その理由を詳細に説明してください。"
                ),
            ]
        )
        chain = prompt | self.llm
        interview_text = "\n".join(
            f"ペルソナ: {i.persona.name} - {i.persona.background}\n質問: {i.question}\n回答: {i.answer}\n"
            for i in interviews
        )
        return chain.invoke({"user_request": user_request, "interview_results": interview_text})


# 仕様書生成クラス
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"
                    "仕様書には以下のセクションを含め、各セクションは可能な限り詳細な説明と具体例、数値データ、過去の事例、"
                    "改善提案、参考文献リスト、そして将来的な展望までを含めた、長文かつ包括的な文章で記述してください:\n"
                    "1. プロジェクト概要\n"
                    "2. 事業目的および背景\n"
                    "3. 業務範囲および提供サービス\n"
                    "4. 技術的要件\n"
                    "5. スケジュールおよびマイルストーン\n"
                    "6. 予算見積\n\n"
                    "出力は必ず日本語で、極めて詳細な文章で記述してください。\n\n仕様書:"
                ),
            ]
        )
        chain = prompt | self.llm | StrOutputParser()
        interview_text = "\n".join(
            f"ペルソナ: {i.persona.name} - {i.persona.background}\n質問: {i.question}\n回答: {i.answer}\n"
            for i in interviews
        )
        return chain.invoke({"user_request": user_request, "interview_results": interview_text})


# 委託事業仕様書生成AIエージェントのクラス
class DocumentationAgent:
    def __init__(self, llm: ChatOpenAI, k: Optional[int] = None):
        self.persona_generator = PersonaGenerator(llm=llm, k=k)
        self.interview_conductor = InterviewConductor(llm=llm)
        self.information_evaluator = InformationEvaluator(llm=llm)
        self.requirements_generator = RequirementsDocumentGenerator(llm=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.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[str, Any]:
        new_personas: Personas = self.persona_generator.run(state.user_request)
        return {
            "personas": new_personas.personas,
            "iteration": state.iteration + 1,
        }

    def _conduct_interviews(self, state: InterviewState) -> dict[str, Any]:
        new_interviews: InterviewResult = self.interview_conductor.run(
            state.user_request, state.personas[-5:]
        )
        return {"interviews": new_interviews.interviews}

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

    def _generate_requirements(self, state: InterviewState) -> dict[str, Any]:
        requirements_doc: str = self.requirements_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"]


def main():
    import argparse

    parser = argparse.ArgumentParser(
        description="ユーザー要求に基づいて要件定義を生成します"
    )
    parser.add_argument(
        "--task",
        type=str,
        help="作成したいアプリケーションについて記載してください",
    )
    parser.add_argument(
        "--k",
        type=int,
        default=5,
        help="生成するペルソナの人数を設定してください（デフォルト:5）",
    )
    args = parser.parse_args()

    llm = ChatOpenAI(model="gpt-4o", temperature=0.0)
    agent = DocumentationAgent(llm=llm, k=args.k)
    final_output = agent.run(user_request=args.task)
    print(final_output)


if __name__ == "__main__":
    main()

3. /content/drive/MyDrive/AIエージェント/main.py

import operator
from typing import Annotated, Any, Optional

from dotenv import load_dotenv
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.graph import END, StateGraph

# .envファイルから環境変数を読み込む
load_dotenv()

# 関係者（ペルソナ）を表すデータモデル
class Persona(BaseModel):
    name: str = Field(..., description="関係者の名前")
    background: str = Field(..., description="関係者の役割や背景情報")

# 関係者リストのデータモデル
class Personas(BaseModel):
    personas: list[Persona] = Field(default_factory=list, description="関係者（ペルソナ）のリスト")

# インタビュー内容を表すデータモデル
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 EvaluationResult(BaseModel):
    is_sufficient: bool = Field(..., description="仕様書作成に十分な情報かどうか")
    reason: str = Field(..., description="判断の理由")

# 仕様書作成エージェントの状態モデル
class InterviewState(BaseModel):
    user_request: str = Field(..., description="ユーザーからの委託事業に関する要求")
    personas: Annotated[list[Persona], operator.add] = Field(default_factory=list, description="生成された関係者（ペルソナ）のリスト")
    interviews: Annotated[list[Interview], operator.add] = 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="仕様書作成に必要な情報が十分かどうか")

# 関係者（ペルソナ）生成クラス
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要求内容: {{user_request}}\n\n各ペルソナには、名前、役割、背景などを詳細に記載してください。")
        ])
        chain = prompt | self.llm
        return chain.invoke({"user_request": user_request})

# インタビュー実施クラス
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._create_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", "以下の要求内容と関係者情報に基づき、質問を1つ生成してください。\n\n要求内容: {user_request}\n関係者: {persona_name} - {persona_background}\n\n質問:")
        ])
        question_chain = question_prompt | self.llm | StrOutputParser()
        question_queries = [{"user_request": user_request, "persona_name": persona.name, "persona_background": persona.background} for persona in personas]
        return question_chain.batch(question_queries)

    def _generate_answers(self, personas: list[Persona], questions: list[str]) -> list[str]:
        answer_prompt = ChatPromptTemplate.from_messages([
            ("system", "あなたは、以下の関係者として質問に対する回答を提供する専門家です: {persona_name} - {persona_background}"),
            ("human", "質問: {question}")
        ])
        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 _create_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)]

# 情報の十分性評価クラス
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\nインタビュー結果:\n{interview_results}\n\n十分なら 'true'、不足なら 'false' とその理由を回答してください。")
        ])
        chain = prompt | self.llm
        interview_text = "\n".join([f"ペルソナ: {i.persona.name} - {i.persona.background}\n質問: {i.question}\n回答: {i.answer}" for i in interviews])
        return chain.invoke({"user_request": user_request, "interview_results": interview_text})

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仕様書:")
        ])
        chain = prompt | self.llm | StrOutputParser()
        interview_text = "\n".join([f"ペルソナ: {i.persona.name} - {i.persona.background}\n質問: {i.question}\n回答: {i.answer}" for i in interviews])
        return chain.invoke({"user_request": user_request, "interview_results": interview_text})

class DocumentationAgent:
    def __init__(self, llm: ChatOpenAI, k: Optional[int] = None):
        self.persona_generator = PersonaGenerator(llm=llm, k=k)
        self.interview_conductor = InterviewConductor(llm=llm)
        self.information_evaluator = InformationEvaluator(llm=llm)
        self.requirements_generator = RequirementsDocumentGenerator(llm=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.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[str, Any]:
        new_personas: Personas = self.persona_generator.run(state.user_request)
        return {"personas": new_personas.personas, "iteration": state.iteration + 1}

    def _conduct_interviews(self, state: InterviewState) -> dict[str, Any]:
        new_interviews: InterviewResult = self.interview_conductor.run(state.user_request, state.personas[-5:])
        return {"interviews": new_interviews.interviews}

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

    def _generate_requirements(self, state: InterviewState) -> dict[str, Any]:
        requirements_doc: str = self.requirements_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"]

def main():
    import argparse
    parser = argparse.ArgumentParser(description="○○県発注の委託事業仕様書を生成します")
    parser.add_argument("--task", type=str, default="デフォルトの委託事業要求", help="作成したい委託事業の要求内容を記載してください（例：県有施設のHP改修、デジタルプロモーション、イベントプロモーション等）")
    parser.add_argument("--k", type=int, default=5, help="生成するペルソナの人数を設定してください（デフォルト:5）")
    args, _ = parser.parse_known_args()
    llm = ChatOpenAI(model="gpt-4o", temperature=0.0)
    agent = DocumentationAgent(llm=llm, k=args.k)
    final_output = agent.run(user_request=args.task)
    print(final_output)

if __name__ == "__main__":
    main()