楽しく学ぶAIエージェントの作り方！書籍『LangChainとLangGraphによるRAG・AIエージェント[実践]入門』から得た学び3選

chain = LLMChain(prompt = prompt, llm = llm, output_parser = output_parser)

chain = prompt | llm | output_parser

from pydantic import BaseModel, Field
from typing import Any, Annotated
import operator
from langchain_openai import ChatOpenAI
from single_path_plan_generation.main import QueryDecomposer, DecomposedTasks
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.tools import TavilySearchResults
from langchain_core.output_parsers import StrOutputParser
from langgraph.prebuilt import create_react_agent
from langgraph.graph import StateGraph, END

# ==============================
# データモデル定義
# ==============================

# 役割 (Role) を表すデータモデル
class Role(BaseModel):
    name: str  = Field(..., description="役割の名前")
    description: str = Field(..., description="役割の詳細な名前")
    key_skills: list[str] = Field(..., description="この役割に必要な主要なスキルや属性")

# タスク (Task) を表すデータモデル
class Task(BaseModel):
    description: str = Field(..., description="タスクの説明")
    role: Role = Field(default=None, description="タスクに割り当てられた役割")

# 役割が割り当てられたタスクのリストを表すデータモデル
class TaskWithRoles(BaseModel):
    tasks: list[Task] = Field(..., description="役割が割り当てられたタスクのリスト")

# エージェントの状態を管理するデータモデル
class AgentState(BaseModel):
    query: str = Field(..., description="ユーザーが入力したクエリ")
    tasks: list[Task] = Field(
        default_factory=list, description="実行するタスクのリスト"
    )
    current_task_index: int = Field(default=0, description="現在実行中のタスクの番号")
    results: Annotated[list[str], operator.add] = Field(
        default_factory=list, description="実行済みタスクの結果リスト"
    )
    final_report: str = Field(default="", description="最終的な出力結果")

# ==============================
# 各機能のクラス定義
# ==============================

# クエリを分解してタスクを生成するプランナー
class Planner:
    def __init__(self, llm: ChatOpenAI): 
        self.query_decomposer = QueryDecomposer(llm=llm)

    def run(self, query: str) -> list[Task]:
        # クエリを分解し、それぞれの部分タスクを作成
        decomposed_tasks: DecomposedTasks = self.query_decomposer.run(query=query)
        return [Task(description=task) for task in decomposed_tasks.values]

# 各タスクに役割を割り当てるロールアサイナー
class RoleAssigner:
    def __init__(self, llm: ChatOpenAI):
        # LLM の出力を TaskWithRoles 型として扱う
        self.llm = llm.with_structured_output(TaskWithRoles)

    def run(self, tasks: list[Task]) -> list[Task]:
        # タスクに基づいて適切な役割を生成するプロンプトを作成
        prompt = ChatPromptTemplate(
            [
                ("system",
                    "あなたは創造的な役割設計の専門家です。与えられたタスクに対して、ユニークで適切な役割を生成してください。"
                ),
                ("human",
                    "タスク: \n{tasks}\n\n"
                    "これらのタスクに対して、以下の指示に従って役割を割り当ててください。\n"
                    "1. 各タスクに対して、独自の創造的な役割を考案してください。\n"
                    "2. 役割名は、そのタスクの本質を反映した魅力的で記憶に残るものにしてください。\n"
                    "3. 各役割に対して、その役割がなぜそのタスクに最適なのかを説明してください。\n"
                    "4. その役割に必要な主要スキルを3つ挙げてください。\n\n"
                    "創造性を発揮し、タスクの本質を捉えた革新的な役割を生成してください。"
                )
            ]
        )
        chain = prompt | self.llm 
        tasks_with_roles = chain.invoke(
            {"tasks": "\n".join([task.description for task in tasks])}
        )
        return tasks_with_roles.tasks 

# タスクを実行するエージェント
class Executor:
    def __init__(self, llm: ChatOpenAI):
        self.llm = llm 
        self.tools = [TavilySearchResults(max_results=3)]
        self.base_agent = create_react_agent(self.llm, self.tools)

    def run(self, task: Task) -> str:
        # 役割に基づいてタスクを実行
        result = self.base_agent.invoke(
            {
                "messages": [
                    ("system", 
                        f"あなたは{task.role.name}です\n"
                        f"説明: {task.role.description}\n"
                        f"主要なスキル: {', '.join(task.role.key_skills)}\n"
                        "与えられたタスクを最高の能力で遂行してください。"
                    ),
                    ("human", 
                        f"以下のタスクを実行してください: \n\n{task.description}"
                    )
                ]
            }
        )
        return result["messages"][-1].content 

# タスクの結果を統合してレポートを作成するクラス
class Reporter:
    def __init__(self, llm: ChatOpenAI):
        self.llm = llm 

    def run(self, query: str, results: list[str]) -> str:
        # 収集した情報を元にレポートを作成
        prompt = ChatPromptTemplate(
            [
                ("system", 
                    "あなたは総合的なレポート作成の専門家です。"
                ),
                ("human",
                    "タスク: 以下の情報に基づいて、包括的で一貫性のある回答を作成してください。\n"
                    "ユーザーの依頼: {query}\n\n"
                    "収集した情報: \n{results}"
                )
            ]
        )
        chain = prompt | self.llm | StrOutputParser()
        return chain.invoke({"query": query, "results": "\n\n".join(
            f"info {i + 1}: \n{result}" for i, result in enumerate(results)
        )})

# 役割に基づいて協力しながらタスクを実行するクラス
class RoleBasedCooperation:
    def __init__(self, llm: ChatOpenAI):
        self.llm = llm
        self.planner = Planner(llm)
        self.role_assigner = RoleAssigner(llm)
        self.executor = Executor(llm)
        self.reporter = Reporter(llm)
        self.graph = self._create_graph()

    def _create_graph(self) -> StateGraph:
        graph = StateGraph(AgentState)
        graph.add_node("planner", self._plan_tasks)
        graph.add_node("role_assigner", self._assign_roles)
        graph.add_node("executor", self._execute_task)
        graph.add_node("reporter", self._generate_report)
        
        # タスクの流れを定義
        graph.set_entry_point("planner")
        graph.add_edge("planner", "role_assigner")
        graph.add_edge("role_assigner", "executor")
        graph.add_conditional_edges("executor", lambda state: state.current_task_index < len(state.tasks), {True: "executor", False: "reporter"})
        graph.add_edge("reporter", END)

        return graph.compile()

def main():
    import argparse

    from tmp.settings import Settings 

    settings = Settings()

    parser = argparse.ArgumentParser(
        description="RoleBasedCooperationを使用してタスクを実行します"
    )
    parser.add_argument("--task", type = str, required=True, help = "実行するタスク")
    args = parser.parse_args()

    llm = ChatOpenAI(
        model = settings.openai_smart_model, temperature=settings.temperature
    )
    agent = RoleBasedCooperation(llm = llm)
    result = agent.run(query = args.task)
    print(result)

if __name__ == "__main__":
    main()

python -m role_based_cooperation.main --task "1年以内にフルマラソンをsub3で完走したいと考えています。本番に向けたメニューを考えてください。"

{
  "tasks": [
    {
      "description": "フルマラソンsub3達成に必要なトレーニングプランをインターネットで調査し、具体的な週間スケジュールを作成する。",
      "role": {
        "name": "マラソンメトロノーム",
        "description": "この役割は、フルマラソンsub3達成に向けたトレーニングプランを精密に調整し、最適な週間スケジュールを作成することに特化しています。マラソンメトロノームは、トレーニングのリズムとペースを完璧に調整し、ランナーが目標を達成するための最適なプランを提供します。",
        "key_skills": [
          "データ分析能力",
          "計画立案能力",
          "時間管理能力"
        ]
      }
    },
    {
      "description": "インターネットでsub3達成者の体験談や成功事例を調査し、どのようなトレーニングや戦略が効果的かを分析する。",
      "role": {
        "name": "成功ストーリーアーキビスト",
        "description": "成功ストーリーアーキビストは、sub3達成者の体験談を収集し、そこから有効なトレーニング戦略を抽出する役割です。過去の成功事例を分析し、最も効果的なアプローチを特定することで、ランナーにとっての最適な戦略を構築します。",
        "key_skills": [
          "リサーチ能力",
          "分析力",
          "コミュニケーション能力"
        ]
      }
    },
    {
      "description": "インターネットで栄養管理や食事プランについて調査し、sub3達成に向けた食事計画を立てる。",
      "role": {
        "name": "栄養ナビゲーター",
        "description": "栄養ナビゲーターは、ランナーがsub3を達成するために必要な栄養素を的確に把握し、最適な食事プランを設計する役割です。栄養の知識を駆使して、エネルギー補給と回復を最大化する食事計画を提供します。",
        "key_skills": [
          "栄養学の知識",
          "プランニング能力",
          "情報収集能力"
        ]
      }
    },
    {
      "description": "インターネットでフルマラソンにおける怪我予防やリカバリー方法を調査し、トレーニングに組み込むべきストレッチやケア方法を決定する。",
      "role": {
        "name": "リカバリーエンジニア",
        "description": "リカバリーエンジニアは、怪我予防とリカバリーのための最適なストレッチやケア方法を設計し、トレーニングプランに組み込む役割です。ランナーが安全にトレーニングを続けられるよう、最新のリカバリー技術を活用します。",
        "key_skills": [
          "生体力学の知識",
          "問題解決能力",
          "注意深さ"
        ]
      }
    },
    {
      "description": "インターネットでレース当日の戦略やペース配分について調査し、具体的なレースプランを作成する。",
      "role": {
        "name": "レースストラテジスト",
        "description": "レースストラテジストは、レース当日の戦略を緻密に計画し、ペース配分を最適化する役割です。ランナーが最高のパフォーマンスを発揮できるよう、レースの全体像を見据えた戦略を提供します。",
        "key_skills": [
          "戦略的思考",
          "ペース管理能力",
          "状況判断力"
        ]
      }
    }
  ]
}

フルマラソンでsub3を達成するためのトレーニングプランを以下に提案します。これは、16週間の計画を基にしたもので、トレーニング、栄養、怪我予防、レース戦略を統合しています。

### トレーニングプラン
1. **週間スケジュール**:
   - **月曜日**: 休息日。体を回復させる。
   - **火曜日**: スピードトレーニング（400mインターバルを8〜10本）。
   - **水曜日**: イージーラン（10kmをゆっくりとしたペースで）。
   - **木曜日**: テンポラン（8〜10kmをマラソンペースより少し速く）。
   - **金曜日**: クロストレーニング（サイクリングや水泳を45分）。
   - **土曜日**: ロングラン（25〜30kmをゆっくりとしたペースで）。
   - **日曜日**: コアトレーニングとストレッチ。

2. **トレーニング戦略**:
   - 簡単な日とハードな日を明確に分け、プログレッションとビルドアップを意識。
   - 栄養計画を最適化し、トレーニング中に試す。

### 栄養プラン
- **炭水化物**: 体重1ポンドあたり4-6gを摂取。
- **タンパク質**: 体重1ポンドあたり0.8-1.0gを摂取。
- **脂肪**: 体重1ポンドあたり0.4-0.6gを摂取。
- **食事例**: 朝食にオートミール、昼食に全粒粉パンのサンドイッチ、夕食にグリルしたサーモン。

### 怪我予防とリカバリー
- **ストレッチ**: 股関節と足首の可動性を高める。
- **筋膜リリース**: ランニング後に行う。
- **睡眠**: 十分な睡眠を確保。

### レース戦略
- **スタート前**: 軽いジョギングとストレッチでウォームアップ。
- **序盤 (0-5km)**: 目標ペースより少し遅めにスタート。
- **中盤 (5-15km)**: 目標ペースを維持し、水分補給を忘れずに。
- **終盤 (15-20km)**: 体調が良ければペースを上げる。
- **フィニッシュ**: ゴールが見えたら全力でスプリント。

このプランは、個々の体力や経験に応じて調整が必要です。トレーニングの進捗に応じて距離やペースを調整し、栄養とリカバリーをしっかりと行うことで、sub3達成に向けた準備を整えましょう。