LangChainとLangGraphで構築するブログ記事作成マルチAIエージェントの実装

はじめに

AI技術の進歩により、単一のAIエージェントだけでなく、複数のAIエージェントが協調して作業するマルチエージェントシステムが注目を集めています。今回は、PythonのLangChainとLangGraphフレームワークを使用して、ブログ記事作成に特化したマルチAIエージェントシステムを構築する方法を詳しく解説します。

マルチAIエージェントとは

マルチAIエージェントシステムは、異なる専門性を持つ複数のAIエージェントが連携して、より複雑で高品質な作業を実現するアーキテクチャです。各エージェントは特定の役割に特化し、以下のような利点があります：

• 専門性の向上: 各エージェントが特定のタスクに特化
• 品質の向上: 複数の視点からのチェックとフィードバック
• スケーラビリティ: 新しいエージェントの追加が容易
• 可読性: 各処理の責任が明確

LangChainとLangGraphの概要

LangChainは、大規模言語モデル（LLM）を活用したアプリケーション開発のためのフレームワークです。プロンプト管理、チェーン構築、メモリ管理などの機能を提供します。

LangGraphは、LangChainエコシステムの一部として開発された、グラフベースの実行フローを構築するためのライブラリです。複雑なワークフローや条件分岐を含むエージェントシステムの構築に適しています。

必要なライブラリのインストール

pip install langchain
pip install langgraph
pip install langchain-openai
pip install python-dotenv

環境設定

import os
from dotenv import load_dotenv

load_dotenv()

# OpenAI API Keyの設定（.envファイルに記載）
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"

ブログ記事作成マルチAIエージェントの実装

以下に、リサーチャー、ライター、レビューアー、エディターの4つのエージェントで構成されるブログ記事作成システムの完全な実装を示します。

import os
from typing import TypedDict, List, Dict, Any
from langchain_openai import ChatOpenAI
from langchain.schema import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolExecutor
from langchain.tools import Tool
import json
import datetime

# 状態管理用のクラス
class BlogState(TypedDict):
    topic: str
    research_data: List[str]
    initial_draft: str
    reviewed_draft: str
    final_article: str
    feedback: List[str]
    current_agent: str
    iteration_count: int
    max_iterations: int

class BlogMultiAgent:
    def __init__(self, api_key: str):
        os.environ["OPENAI_API_KEY"] = api_key
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.7)
        self.graph = self._build_graph()
    
    def _build_graph(self):
        """マルチエージェントのワークフローグラフを構築"""
        workflow = StateGraph(BlogState)
        
        # ノード（エージェント）の追加
        workflow.add_node("researcher", self.researcher_agent)
        workflow.add_node("writer", self.writer_agent)
        workflow.add_node("reviewer", self.reviewer_agent)
        workflow.add_node("editor", self.editor_agent)
        
        # エッジ（フロー）の設定
        workflow.set_entry_point("researcher")
        workflow.add_edge("researcher", "writer")
        workflow.add_edge("writer", "reviewer")
        workflow.add_conditional_edges(
            "reviewer",
            self._should_continue_review,
            {
                "continue": "writer",
                "edit": "editor",
                "end": END
            }
        )
        workflow.add_edge("editor", END)
        
        return workflow.compile()
    
    def researcher_agent(self, state: BlogState) -> BlogState:
        """リサーチエージェント：トピックに関する情報収集"""
        print(f"🔍 Researcher Agent: Researching topic '{state['topic']}'")
        
        research_prompt = f"""
        あなたは専門のリサーチャーです。以下のトピックについて詳細な調査を行い、
        ブログ記事作成に必要な情報を収集してください。

        トピック: {state['topic']}

        以下の観点から情報を整理してください：
        1. トピックの基本的な定義と背景
        2. 最新のトレンドや動向
        3. 実践的な応用例
        4. 関連する技術やツール
        5. 注意点や課題

        各項目について具体的で実用的な情報を提供してください。
        """
        
        response = self.llm.invoke([SystemMessage(content=research_prompt)])
        research_data = [response.content]
        
        state["research_data"] = research_data
        state["current_agent"] = "researcher"
        print("✅ Research completed")
        
        return state
    
    def writer_agent(self, state: BlogState) -> BlogState:
        """ライターエージェント：記事の執筆"""
        print(f"✍️ Writer Agent: Writing article on '{state['topic']}'")
        
        research_content = "\n".join(state["research_data"])
        
        if state.get("reviewed_draft"):
            # レビュー後の修正
            feedback_content = "\n".join(state["feedback"])
            writing_prompt = f"""
            あなたは経験豊富なテクニカルライターです。
            以下のレビューフィードバックを参考に、記事を修正してください。

            元の記事:
            {state["reviewed_draft"]}

            フィードバック:
            {feedback_content}

            修正版の記事を作成してください。記事は以下の構造を持つようにしてください：
            - 魅力的なタイトル
            - 導入部
            - 詳細な本文（複数のセクション）
            - 実践的な例
            - まとめ

            読者に価値のある情報を提供し、実践的で分かりやすい内容にしてください。
            """
        else:
            # 初回執筆
            writing_prompt = f"""
            あなたは経験豊富なテクニカルライターです。
            以下のリサーチ情報を基に、魅力的で実用的なブログ記事を作成してください。

            トピック: {state['topic']}

            リサーチ情報:
            {research_content}

            記事は以下の構造を持つようにしてください：
            - 魅力的なタイトル
            - 導入部（読者の興味を引く）
            - 詳細な本文（複数のセクション）
            - 実践的な例やコード例
            - まとめと次のステップ

            読者に価値のある情報を提供し、実践的で分かりやすい内容にしてください。
            文字数は2000-3000字程度を目安にしてください。
            """
        
        response = self.llm.invoke([SystemMessage(content=writing_prompt)])
        
        state["initial_draft"] = response.content
        state["current_agent"] = "writer"
        print("✅ Article draft completed")
        
        return state
    
    def reviewer_agent(self, state: BlogState) -> BlogState:
        """レビューアーエージェント：記事の品質チェック"""
        print("🔍 Reviewer Agent: Reviewing article quality")
        
        article_to_review = state["initial_draft"]
        
        review_prompt = f"""
        あなたは経験豊富なエディターです。以下のブログ記事を詳細にレビューし、
        改善点を指摘してください。

        記事:
        {article_to_review}

        以下の観点から評価してください：
        1. 内容の正確性と完全性
        2. 構成と流れの論理性
        3. 読みやすさと分かりやすさ
        4. 実践的価値
        5. SEO対策の観点

        改善が必要な場合は具体的なフィードバックを提供し、
        記事の品質スコア（1-10）も提示してください。

        スコアが8以上の場合は「APPROVED」、それ以下の場合は「NEEDS_IMPROVEMENT」と
        最初に記載してください。
        """
        
        response = self.llm.invoke([SystemMessage(content=review_prompt)])
        review_content = response.content
        
        # レビュー結果の判定
        if "APPROVED" in review_content:
            state["reviewed_draft"] = article_to_review
            state["feedback"] = [review_content]
        else:
            state["feedback"] = state.get("feedback", []) + [review_content]
            state["iteration_count"] = state.get("iteration_count", 0) + 1
        
        state["current_agent"] = "reviewer"
        print("✅ Review completed")
        
        return state
    
    def editor_agent(self, state: BlogState) -> BlogState:
        """エディターエージェント：最終編集と公開準備"""
        print("📝 Editor Agent: Final editing and formatting")
        
        article = state["reviewed_draft"]
        
        edit_prompt = f"""
        あなたは熟練のエディターです。以下の承認済み記事に最終的な編集を加え、
        公開準備を整えてください。

        記事:
        {article}

        以下の作業を行ってください：
        1. 最終的な文章校正
        2. 見出しの調整とフォーマット
        3. メタデータの追加（推定読了時間、タグなど）
        4. SEOメタディスクリプションの作成
        5. 読者にとってより魅力的になる微調整

        完成した記事を出力してください。
        """
        
        response = self.llm.invoke([SystemMessage(content=edit_prompt)])
        
        state["final_article"] = response.content
        state["current_agent"] = "editor"
        print("✅ Final editing completed")
        
        return state
    
    def _should_continue_review(self, state: BlogState) -> str:
        """レビュー結果に基づく条件分岐"""
        feedback = state.get("feedback", [])
        iteration_count = state.get("iteration_count", 0)
        max_iterations = state.get("max_iterations", 3)
        
        if not feedback:
            return "end"
        
        latest_feedback = feedback[-1]
        
        if "APPROVED" in latest_feedback:
            return "edit"
        elif iteration_count >= max_iterations:
            print(f"⚠️ Maximum iterations ({max_iterations}) reached. Moving to editor.")
            return "edit"
        else:
            print(f"🔄 Revision needed. Iteration {iteration_count + 1}")
            return "continue"
    
    def create_blog_article(self, topic: str, max_iterations: int = 3) -> Dict[str, Any]:
        """ブログ記事作成のメイン実行関数"""
        print(f"🚀 Starting blog article creation for topic: '{topic}'")
        print("=" * 60)
        
        # 初期状態の設定
        initial_state = {
            "topic": topic,
            "research_data": [],
            "initial_draft": "",
            "reviewed_draft": "",
            "final_article": "",
            "feedback": [],
            "current_agent": "",
            "iteration_count": 0,
            "max_iterations": max_iterations
        }
        
        # グラフの実行
        result = self.graph.invoke(initial_state)
        
        print("=" * 60)
        print("✅ Blog article creation completed!")
        
        return {
            "topic": result["topic"],
            "final_article": result["final_article"],
            "research_data": result["research_data"],
            "feedback_history": result["feedback"],
            "iterations": result["iteration_count"],
            "creation_date": datetime.datetime.now().isoformat()
        }

# 使用例とテスト
def main():
    """メイン実行関数"""
    
    # APIキーの設定（実際の使用時は環境変数から取得）
    api_key = "your-openai-api-key-here"  # 実際のAPIキーに置き換えてください
    
    # ブログ作成エージェントのインスタンス化
    blog_agent = BlogMultiAgent(api_key)
    
    # ブログ記事の作成
    topic = "PythonでAPIを活用したWebスクレイピングの基礎と実践"
    
    try:
        result = blog_agent.create_blog_article(topic, max_iterations=2)
        
        # 結果の表示
        print("\n" + "=" * 80)
        print("📄 FINAL BLOG ARTICLE")
        print("=" * 80)
        print(result["final_article"])
        
        # メタデータの表示
        print("\n" + "=" * 80)
        print("📊 CREATION METADATA")
        print("=" * 80)
        print(f"Topic: {result['topic']}")
        print(f"Iterations: {result['iterations']}")
        print(f"Creation Date: {result['creation_date']}")
        
        # 記事をファイルに保存
        filename = f"blog_article_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.md"
        with open(filename, 'w', encoding='utf-8') as f:
            f.write(f"# {result['topic']}\n\n")
            f.write(f"*Created on: {result['creation_date']}*\n\n")
            f.write(result['final_article'])
        
        print(f"\n📁 Article saved to: {filename}")
        
    except Exception as e:
        print(f"❌ Error during article creation: {e}")

if __name__ == "__main__":
    main()

実行とテスト

実際にシステムを動作させる際の手順：

1. 環境設定: OpenAI APIキーの設定
2. トピック選定: 作成したいブログ記事のトピックを決定
3. 実行: create_blog_articleメソッドを呼び出し
4. 結果確認: 生成された記事とメタデータの確認

パフォーマンス最適化のポイント

コスト効率の向上

• 適切なモデル選択（GPT-3.5 vs GPT-4）
• プロンプトの最適化
• 不要な反復の削減

実行速度の向上

• 並列処理の実装
• キャッシュ機能の活用
• ストリーミング処理の導入

実運用での考慮事項

セキュリティ

• APIキーの適切な管理
• 入力データの検証
• 出力コンテンツのフィルタリング

監視とロギング

• 各エージェントの実行ログ
• エラーハンドリング
• パフォーマンスメトリクス

まとめ

LangChainとLangGraphを使用したマルチAIエージェントシステムにより、高品質なブログ記事を自動生成することが可能になります。各エージェントの専門性を活かし、段階的な品質向上を実現できるこのアプローチは、コンテンツマーケティングや技術文書作成の効率化に大きく貢献します。

実装のポイントは、明確な役割分担、適切な品質チェック、そして柔軟なワークフロー設計です。今回のコードをベースに、用途に応じてカスタマイズを行い、より高度なコンテンツ生成システムを構築してください。

次のステップ

1. カスタマイゼーション: 特定のドメインに特化したエージェントの追加
2. 統合: CMSやブログプラットフォームとの連携
3. 自動化: スケジュール実行や自動公開機能の実装
4. 分析: 生成された記事のパフォーマンス分析機能

このマルチAIエージェントシステムを活用して、効率的で高品質なコンテンツ作成を実現してください。