AIエージェント・オーケストレーションフレームワーク2026：LangGraph vs CrewAI vs AutoGen完全ガイド

from langgraph.graph import StateGraph
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    task: str
    result: str

graph_builder = StateGraph(AgentState)

# ノードの定義
def process_task(state: AgentState) -> AgentState:
    # タスク処理ロジック
    return {"result": "タスク完了"}

# 条件付きエッジの定義
def should_continue(state: AgentState) -> str:
    if state["result"]:
        return "end"
    return "retry"

graph_builder.add_node("process", process_task)
graph_builder.add_conditional_edges("process", should_continue)

graph = graph_builder.compile()
result = graph.invoke({"messages": [], "task": "分析"})

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

class DocumentState(TypedDict):
    document: str
    analysis: str
    insights: list

graph = StateGraph(DocumentState)
llm = ChatOpenAI(model="gpt-4")

def analyze_document(state: DocumentState) -> DocumentState:
    response = llm.invoke(f"分析: {state['document']}")
    return {"analysis": response.content}

def extract_insights(state: DocumentState) -> DocumentState:
    response = llm.invoke(f"主要な洞察を抽出: {state['analysis']}")
    return {"insights": [item.strip() for item in response.content.split('\n')]}

graph.add_node("analyze", analyze_document)
graph.add_node("extract", extract_insights)
graph.add_edge("analyze", "extract")
graph.add_edge("extract", END)

workflow = graph.compile()

from crewai import Agent, Task, Crew
from crewai_tools import tool

class ResearchTools:
    @tool
    def search_web(query: str) -> str:
        """ウェブ検索を実行"""
        return f"検索結果: {query}"

# エージェントの定義
researcher = Agent(
    role="Research Analyst",
    goal="トピックについて正確な情報を見つけること",
    backstory="10年の経験を持つ専門研究者",
    tools=[ResearchTools.search_web()],
    verbose=True
)

writer = Agent(
    role="Content Writer",
    goal="魅力的で有益なコンテンツを執筆すること",
    backstory="テック出版物の専門家ライター"
)

# タスクの定義
research_task = Task(
    description="2026年のAIエージェントについて研究",
    agent=researcher,
    expected_output="詳細な研究報告書"
)

writing_task = Task(
    description="研究に基づいてブログ記事を執筆",
    agent=writer,
    context=[research_task]
)

# クルーの構成
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    verbose=2
)

result = crew.kickoff()

from autogen import AssistantAgent, UserProxyAgent

# アシスタントエージェントの作成
assistant = AssistantAgent(
    name="Scientist",
    system_message="あなたは有用なAI科学アシスタントです"
)

# ユーザープロキシエージェント
user_proxy = UserProxyAgent(
    name="User",
    human_input_mode="TERMINATE",
    code_execution_config={"use_docker": False}
)

# 対話の開始
user_proxy.initiate_chat(
    assistant,
    message="CSVファイルを分析して統計を生成するPythonコードを書いてください"
)

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

# 複数の専門家エージェント
data_scientist = AssistantAgent(
    name="DataScientist",
    system_message="あなたは専門的なデータサイエンティストです"
)

engineer = AssistantAgent(
    name="Engineer",
    system_message="あなたは専門的なソフトウェアエンジニアです"
)

# グループチャットの設定
groupchat = GroupChat(
    agents=[data_scientist, engineer],
    messages=[],
    max_round=10
)

manager = GroupChatManager(groupchat=groupchat)

user_proxy.initiate_chat(
    manager,
    message="売上予測のための機械学習パイプラインを構築してください"
)

version: 1.0
name: 'カスタマーサポートワークフロー'

nodes:
  - id: input
    type: input
    config:
      title: '顧客質問'

  - id: classify
    type: llm
    config:
      model: gpt-4
      prompt: 'このクエリを分類してください: {{input.query}}'

  - id: route
    type: switch
    config:
      cases:
        - value: technical
          next: technical_agent
        - value: billing
          next: billing_agent

  - id: technical_agent
    type: agent
    config:
      role: 'Technical Support'

  - id: output
    type: output
    config:
      title: '応答'

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

class ArticleState(TypedDict):
    article: str
    summary: str
    sentiment: str

graph = StateGraph(ArticleState)
llm = ChatOpenAI()

def summarize(state):
    result = llm.invoke(f"要約: {state['article']}")
    return {"summary": result.content}

def analyze_sentiment(state):
    result = llm.invoke(f"感情分析: {state['summary']}")
    return {"sentiment": result.content}

graph.add_node("summarize", summarize)
graph.add_node("sentiment", analyze_sentiment)
graph.add_edge("summarize", "sentiment")
graph.add_edge("sentiment", END)

workflow = graph.compile()
result = workflow.invoke({"article": "...", "summary": "", "sentiment": ""})

from crewai import Agent, Task, Crew

summarizer = Agent(
    role="News Summarizer",
    goal="ニュース記事の正確な要約を作成すること"
)

analyst = Agent(
    role="Sentiment Analyst",
    goal="コンテンツの感情的トーンを分析すること"
)

summary_task = Task(
    description="記事を要約",
    agent=summarizer
)

sentiment_task = Task(
    description="感情を分析",
    agent=analyst,
    context=[summary_task]
)

crew = Crew(
    agents=[summarizer, analyst],
    tasks=[summary_task, sentiment_task]
)

result = crew.kickoff()

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

summarizer = AssistantAgent(name="Summarizer")
analyst = AssistantAgent(name="Sentiment Analyst")

groupchat = GroupChat(agents=[summarizer, analyst], messages=[])
manager = GroupChatManager(groupchat=groupchat)

user_proxy = UserProxyAgent(name="Admin", human_input_mode="TERMINATE")
user_proxy.initiate_chat(
    manager,
    message="この記事を要約して感情分析してください: ..."
)

# 良い例：単一責任
analyzer_agent = Agent(
    role="Data Analyzer",
    goal="構造化データから洞察を抽出すること"
)

# 避けるべき例：責任が広すぎる
universal_agent = Agent(
    role="Universal Assistant",
    goal="すべてをやること"
)

from crewai_tools import tool

@tool
def fetch_market_data(symbol: str) -> str:
    """特定シンボルの市場データを取得"""
    # 実装
    pass

@tool
def analyze_trends(data: str) -> str:
    """データから市場トレンドを分析"""
    # 実装
    pass

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_external_api(endpoint: str):
    # API呼び出し実装
    pass

# 互換性ラッパー例
class LegacyAgentWrapper:
    def __init__(self, new_agent):
        self.agent = new_agent

    def execute(self, task):
        # レガシーインターフェース
        return self.agent.run(task)