LLM Agentフレームワーク比較：AutoGen vs CrewAI vs LangGraph 実践選択ガイド

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat, SelectorGroupChat
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient

# 1. LLMクライアント設定
model_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    api_key="your-api-key",
)

# 2. ツール定義
async def search_web(query: str) -> str:
    """ウェブから情報を検索します。"""
    # 実際にはTavily、SerpAPIなどを使用
    return f"'{query}'に対する検索結果: [検索結果の要約]"

async def analyze_data(data: str) -> str:
    """データを分析しインサイトを抽出します。"""
    return f"分析結果: {data[:100]}に対する深層分析完了"

# 3. エージェント定義
researcher = AssistantAgent(
    name="Researcher",
    model_client=model_client,
    system_message="""あなたは専門リサーチャーです。
    与えられたテーマについてウェブ検索を行い、関連情報を収集します。
    収集した情報を構造化してAnalystに渡します。""",
    tools=[search_web],
)

analyst = AssistantAgent(
    name="Analyst",
    model_client=model_client,
    system_message="""あなたはデータアナリストです。
    Researcherが収集した情報を分析し、核心的なインサイトを抽出します。
    分析結果をWriterに渡します。""",
    tools=[analyze_data],
)

writer = AssistantAgent(
    name="Writer",
    model_client=model_client,
    system_message="""あなたはレポート作成者です。
    Analystの分析結果をもとに最終レポートを作成します。
    レポート作成が完了したら'TERMINATE'を出力します。""",
)

# 4. チーム構成と実行
termination = MaxMessageTermination(max_messages=15) | TextMentionTermination("TERMINATE")

# SelectorGroupChat: LLMが次の発言者を選択
team = SelectorGroupChat(
    participants=[researcher, analyst, writer],
    model_client=model_client,
    termination_condition=termination,
    selector_prompt="""次の発言者を選択してください。
    情報収集が必要ならResearcher、分析が必要ならAnalyst、
    最終レポート作成が必要ならWriterを選択します。""",
)

async def main():
    result = await team.run(
        task="2026年のAIエージェントフレームワーク市場動向を調査し、レポートを作成してください。"
    )
    for message in result.messages:
        print(f"[{message.source}]: {message.content[:200]}")
    print(f"\n総メッセージ数: {len(result.messages)}")

asyncio.run(main())

from crewai import Agent, Task, Crew, Process
from crewai.tools import tool
from langchain_openai import ChatOpenAI

# 1. LLM設定
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

# 2. カスタムツール定義
@tool("Search Tool")
def search_tool(query: str) -> str:
    """指定されたクエリでウェブ検索を実行します。"""
    # 実際にはTavily、SerpAPIなどを使用
    return f"'{query}' 検索結果: [関連情報の要約]"

@tool("SEO Analyzer")
def seo_analyzer(content: str) -> str:
    """コンテンツのSEOスコアを分析します。"""
    word_count = len(content.split())
    return f"単語数: {word_count}, SEOスコア: {'良好' if word_count > 500 else '不足'}"

# 3. エージェント定義
research_agent = Agent(
    role="技術リサーチャー",
    goal="最新の技術トレンドを調査し、正確な情報を収集する",
    backstory="""10年のキャリアを持つ技術ジャーナリストで、AIとソフトウェアエンジニアリング
    分野の最新動向を常に把握しています。""",
    tools=[search_tool],
    llm=llm,
    verbose=True,
    allow_delegation=True,  # 他のエージェントへのタスク委譲が可能
)

writer_agent = Agent(
    role="技術ブログライター",
    goal="読者が容易に理解できる深みのある技術ブログを作成する",
    backstory="""技術ブログ専門のライターで、複雑な技術概念を
    明確かつ実用的な文章に落とし込むことに長けています。""",
    llm=llm,
    verbose=True,
)

editor_agent = Agent(
    role="編集長",
    goal="コンテンツの品質、正確性、SEO最適化を検査する",
    backstory="""技術メディアの編集長で、コンテンツの技術的正確性と
    可読性、SEO最適化を同時に追求しています。""",
    tools=[seo_analyzer],
    llm=llm,
    verbose=True,
)

# 4. タスク定義
research_task = Task(
    description="""'{topic}'に関する総合リサーチを実施します。
    - 最新動向を3つ以上収集
    - 核心的な技術概念の整理
    - 実際の活用事例を2件以上含める""",
    expected_output="構造化されたリサーチレポート（マークダウン形式）",
    agent=research_agent,
)

writing_task = Task(
    description="""リサーチ結果をもとに技術ブログ記事を作成します。
    - 2000字以上の本文
    - コード例を2つ以上含める
    - 実務者視点のインサイトを含める""",
    expected_output="完成したブログ記事（マークダウン形式）",
    agent=writer_agent,
    context=[research_task],  # research_taskの結果をコンテキストとして受け取る
)

editing_task = Task(
    description="""作成されたブログ記事を検査します。
    - 技術的正確性の確認
    - 可読性の改善
    - SEO分析の実施
    - 最終修正版の出力""",
    expected_output="検査完了した最終ブログ記事",
    agent=editor_agent,
    context=[writing_task],
)

# 5. Crew構成と実行
crew = Crew(
    agents=[research_agent, writer_agent, editor_agent],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,  # 順次実行
    verbose=True,
    memory=True,                 # 短期メモリを有効化
    max_rpm=10,                  # API呼び出しレート制限
)

# 実行
result = crew.kickoff(inputs={"topic": "LLMエージェントフレームワーク比較"})
print(f"最終結果:\n{result}")
print(f"トークン使用量: {crew.usage_metrics}")

from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, SystemMessage

# 1. 状態定義
class SupportState(TypedDict):
    messages: Annotated[list, add_messages]
    category: str           # 問い合わせカテゴリ (billing, technical, general)
    sentiment: str          # 顧客感情 (positive, neutral, negative)
    escalated: bool         # 上位担当者へのエスカレーション有無
    resolution: str         # 解決状態

# 2. ツール定義
@tool
def lookup_order(order_id: str) -> str:
    """注文情報を照会します。"""
    orders = {
        "ORD-001": {"status": "配送中", "eta": "2026-03-11"},
        "ORD-002": {"status": "配送完了", "delivered": "2026-03-07"},
    }
    if order_id in orders:
        return f"注文 {order_id}: {orders[order_id]}"
    return f"注文 {order_id}が見つかりません。"

@tool
def create_ticket(summary: str, priority: str) -> str:
    """カスタマーサポートチケットを作成します。"""
    return f"チケット作成完了 - 概要: {summary}, 優先度: {priority}, チケットID: TKT-{hash(summary) % 10000}"

@tool
def refund_order(order_id: str, reason: str) -> str:
    """注文の返金を処理します。（承認が必要）"""
    return f"返金リクエスト受付: {order_id}, 理由: {reason}。管理者承認待ち。"

tools = [lookup_order, create_ticket, refund_order]
llm = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)

# 3. ノード関数定義
def classify_intent(state: SupportState) -> dict:
    """顧客の問い合わせを分類します。"""
    classifier_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    last_message = state["messages"][-1].content

    response = classifier_llm.invoke([
        SystemMessage(content="""顧客の問い合わせを分類してください。
        カテゴリ: billing, technical, general
        感情: positive, neutral, negative
        JSON形式で応答: {"category": "...", "sentiment": "..."}"""),
        HumanMessage(content=last_message)
    ])

    import json
    try:
        result = json.loads(response.content)
    except json.JSONDecodeError:
        result = {"category": "general", "sentiment": "neutral"}

    return {
        "category": result.get("category", "general"),
        "sentiment": result.get("sentiment", "neutral"),
    }

def agent_respond(state: SupportState) -> dict:
    """エージェントが応答します。"""
    system_prompt = f"""あなたはカスタマーサポートエージェントです。
    現在の問い合わせカテゴリ: {state.get('category', 'unknown')}
    顧客感情: {state.get('sentiment', 'unknown')}
    エスカレーション有無: {state.get('escalated', False)}

    顧客に親切かつ正確に応答してください。
    必要に応じてツールを使用して情報を照会したりチケットを作成してください。"""

    messages = [SystemMessage(content=system_prompt)] + state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}

def check_escalation(state: SupportState) -> dict:
    """エスカレーションの必要性を判断します。"""
    if state.get("sentiment") == "negative" and state.get("category") == "billing":
        return {"escalated": True}
    return {"escalated": False}

def human_escalation(state: SupportState) -> dict:
    """上位担当者にエスカレーションします。"""
    return {
        "messages": [SystemMessage(content="[システム] この問い合わせは上位担当者に転送されました。しばらくお待ちください。")],
        "resolution": "escalated_to_human",
    }

# 4. ルーティング関数
def route_after_classification(state: SupportState) -> Literal["check_escalation", "agent_respond"]:
    """分類結果に基づいてルーティングします。"""
    if state.get("sentiment") == "negative":
        return "check_escalation"
    return "agent_respond"

def route_after_escalation_check(state: SupportState) -> Literal["human_escalation", "agent_respond"]:
    """エスカレーション判断結果に基づいてルーティングします。"""
    if state.get("escalated"):
        return "human_escalation"
    return "agent_respond"

# 5. グラフ構成
workflow = StateGraph(SupportState)

# ノード追加
workflow.add_node("classify_intent", classify_intent)
workflow.add_node("check_escalation", check_escalation)
workflow.add_node("agent_respond", agent_respond)
workflow.add_node("tools", ToolNode(tools))
workflow.add_node("human_escalation", human_escalation)

# エッジ定義
workflow.add_edge(START, "classify_intent")
workflow.add_conditional_edges("classify_intent", route_after_classification)
workflow.add_conditional_edges("check_escalation", route_after_escalation_check)
workflow.add_conditional_edges("agent_respond", tools_condition, {"tools": "tools", END: END})
workflow.add_edge("tools", "agent_respond")
workflow.add_edge("human_escalation", END)

# チェックポインター（状態永続化）
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

# 6. 実行
config = {"configurable": {"thread_id": "customer-123"}}
result = app.invoke(
    {"messages": [HumanMessage(content="注文ORD-001の配送が遅すぎます。返金してください。")]},
    config=config,
)

for msg in result["messages"]:
    print(f"[{msg.__class__.__name__}]: {msg.content[:200]}")
print(f"カテゴリ: {result['category']}, 感情: {result['sentiment']}")

# LangGraphでのエラーリカバリパターン例
from tenacity import retry, stop_after_attempt, wait_exponential
from langchain_core.messages import AIMessage

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=10))
def resilient_agent_node(state: SupportState) -> dict:
    """リトライロジックを含むエージェントノード。"""
    try:
        response = llm.invoke(state["messages"])
        return {"messages": [response]}
    except Exception as e:
        # フォールバック: 事前定義された応答を返す
        fallback_response = AIMessage(
            content="申し訳ございません。一時的なシステムエラーが発生しました。"
                    "しばらくしてから再度お試しください。お急ぎの場合は"
                    "カスタマーセンター 0120-000-000 までご連絡ください。"
        )
        return {
            "messages": [fallback_response],
            "resolution": f"error_fallback: {str(e)}",
        }