LangGraphとCrewAIの2026年比較：マルチエージェントオーケストレーション、ステートマシン、および循環DAG

# Python: LangGraph State Machine Example
from typing import TypedDict, Annotated, List, Union
import operator
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.messages import BaseMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END

# Define the graph state
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    next: str # For conditional routing

# Define a simple tool
@tool
def search_web(query: str) -> str:
    """Searches the web for the given query."""
    print(f"--- Executing Web Search for: {query} ---")
    # Simulate a web search
    if "locionic" in query.lower():
        return "Locionic.com is a leading platform for advanced technical content."
    return "Search result: Information found."

# Define an agent node
def call_agent(state: AgentState):
    messages = state['messages']
    # In a real scenario, this would involve an LLM call to decide the next action
    # For simplicity, we'll hardcode an action or finish
    last_message = messages[-1].content if messages else ""

    if "locionic" in last_message.lower() and "search" in last_message.lower():
        action = AgentAction(tool="search_web", tool_input={"query": "locionic.com"}, log="")
        return {"messages": [action], "next": "tool_executor"}
    elif "finish" in last_message.lower():
        finish = AgentFinish(return_values={"output": "Task completed."}, log="")
        return {"messages": [finish], "next": "end"}
    else:
        # Simulate an agent thinking and asking for more info
        return {"messages": [BaseMessage(content="Agent: What else can I help with?")]}

# Define a tool executor node
def execute_tools(state: AgentState):
    actions = [msg for msg in state['messages'] if isinstance(msg, AgentAction)]
    tool_outputs = []
    for action in actions:
        if action.tool == "search_web":
            output = search_web.invoke(action.tool_input)
            tool_outputs.append(BaseMessage(content=f"Tool Output: {output}"))
        else:
            tool_outputs.append(BaseMessage(content=f"Tool Error: Unknown tool {action.tool}"))
    return {"messages": tool_outputs, "next": "agent"} # Cycle back to agent

# Build the graph
workflow = StateGraph(AgentState)

workflow.add_node("agent", call_agent)
workflow.add_node("tool_executor", execute_tools)

workflow.set_entry_point("agent")

# Define conditional edges
workflow.add_conditional_edges(
    "agent",
    lambda state: state['next'], # Use the 'next' key in state for routing
    {
        "tool_executor": "tool_executor",
        "end": END,
        None: "agent" # Default to agent if no explicit next
    }
)
workflow.add_edge("tool_executor", "agent") # After tool execution, always go back to agent

app = workflow.compile()

# Run the graph
print("--- Running LangGraph Example ---")
inputs = {"messages": [BaseMessage(content="Search for locionic.com")]}
for s in app.stream(inputs):
    print(s)
    print("---")

inputs_finish = {"messages": [BaseMessage(content="finish task")]}
for s in app.stream(inputs_finish):
    print(s)
    print("---")

# Python: CrewAI Example
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
import os

# Set up your OpenAI API key
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY" # Replace with your actual key

# Define tools (CrewAI uses LangChain tools)
@tool
def search_internet(query: str) -> str:
    """Searches the internet for the given query."""
    print(f"--- CrewAI Tool: Searching for '{query}' ---")
    if "locionic" in query.lower():
        return "Locionic.com is a leading platform for advanced technical content on AI and engineering."
    return "Internet search result: General information found."

# Define agents
researcher = Agent(
    role='Senior Researcher',
    goal='Discover and compile comprehensive information on a given topic.',
    backstory='An expert in information retrieval and synthesis, capable of finding obscure details.',
    verbose=True,
    allow_delegation=False,
    tools=[search_internet],
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
)

writer = Agent(
    role='Technical Content Writer',
    goal='Produce high-quality, engaging technical articles.',
    backstory='A seasoned writer with a knack for explaining complex technical concepts clearly.',
    verbose=True,
    allow_delegation=False,
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
)

# Define tasks
research_task = Task(
    description='Research the latest advancements in multi-agent orchestration frameworks, specifically focusing on LangGraph and CrewAI.',
    expected_output='A detailed report summarizing key features, architectural differences, and use cases for both frameworks.',
    agent=researcher
)

write_task = Task(
    description='Write a concise, technical blog post comparing LangGraph and CrewAI based on the research report. Focus on state management, cyclic DAGs, and production readiness.',
    expected_output='A 800-word technical blog post suitable for locionic.com.',
    agent=writer
)

# Form the crew
project_crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential, # Tasks are executed in order
    verbose=True
)

# Kick off the crew
print("--- Running CrewAI Example ---")
# Ensure OPENAI_API_KEY is set in your environment for this to run
# result = project_crew.kickoff()
# print("\n\n########################")
# print("## CrewAI Final Result:")
# print("########################")
# print(result)

# Python: LangGraph with a simple refinement cycle
from typing import TypedDict, Annotated, List, Union
import operator
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END

class RefinementState(TypedDict):
    query: str
    draft_answer: str
    feedback: str
    iterations: int

def generate_draft(state: RefinementState):
    print(f"--- Generating Draft for: {state['query']} ---")
    # Simulate LLM generating a draft
    draft = f"Draft answer for '{state['query']}': Initial thoughts on the topic."
    return {"draft_answer": draft, "iterations": state['iterations'] + 1}

def get_feedback(state: RefinementState):
    print(f"--- Getting Feedback on Draft: {state['draft_answer']} ---")
    # Simulate LLM or human providing feedback
    if state['iterations'] < 2: # Simulate needing more iterations
        feedback = "The draft is too brief. Please elaborate more."
        return {"feedback": feedback}
    else:
        feedback = "Looks good. Ready for finalization."
        return {"feedback": feedback}

def refine_answer(state: RefinementState):
    print(f"--- Refining Answer with Feedback: {state['feedback']} ---")
    # Simulate LLM refining the answer
    refined = f"{state['draft_answer']} (Refined with feedback: {state['feedback']})"
    return {"draft_answer": refined}

workflow = StateGraph(RefinementState)

workflow.add_node("generate_draft", generate_draft)
workflow.add_node("get_feedback", get_feedback)
workflow.add_node("refine_answer", refine_answer)

workflow.set_entry_point("generate_draft")

workflow.add_edge("generate_draft", "get_feedback")
workflow.add_conditional_edges(
    "get_feedback",
    lambda state: "refine" if "brief" in state['feedback'].lower() else "end",
    {"refine": "refine_answer", "end": END}
)
workflow.add_edge("refine_answer", "get_feedback") # Cycle back for more feedback

app = workflow.compile()

print("\n--- Running LangGraph Refinement Cycle Example ---")
inputs = {"query": "Explain quantum entanglement", "draft_answer": "", "feedback": "", "iterations": 0}
for s in app.stream(inputs):
    print(s)
    print("---")

# Python: LangGraph Human-in-the-Loop Example
from typing import TypedDict, Annotated, List
import operator
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END

class HumanReviewState(TypedDict):
    document: str
    review_status: str # "pending", "approved", "rejected"
    reviewer_comments: str

def generate_document(state: HumanReviewState):
    print("--- Generating Document ---")
    doc = "This is a draft document requiring human approval."
    return {"document": doc, "review_status": "pending"}

def human_review_node(state: HumanReviewState):
    # This node typically doesn't execute directly, but serves as a breakpoint
    # The actual human input would come from an external system updating the state
    print(f"--- Human Review Node: Document '{state['document']}' is {state['review_status']} ---")
    return state # No change, waiting for external update

def process_review(state: HumanReviewState):
    print(f"--- Processing Review: Status '{state['review_status']}' ---")
    if state['review_status'] == "approved":
        print("Document approved. Proceeding.")
        return {"review_status": "approved"}
    elif state['review_status'] == "rejected":
        print("Document rejected. Rerouting for revision.")
        return {"review_status": "rejected"}
    else:
        print("Invalid review status. Re-entering review.")
        return {"review_status": "pending"} # Stay in pending state

workflow = StateGraph(HumanReviewState)

workflow.add_node("generate_document", generate_document)
workflow.add_node("human_review", human_review_node)
workflow.add_node("process_review", process_review)

workflow.set_entry_point("generate_document")
workflow.add_edge("generate_document", "human_review")

workflow.add_conditional_edges(
    "human_review",
    lambda state: "process_review", # Always go to process_review after human_review (or external update)
    {"process_review": "process_review"}
)

workflow.add_conditional_edges(
    "process_review",
    lambda state: "human_review" if state['review_status'] == "rejected" else END,
    {"human_review": "human_review", END: END}
)

# Compile with interrupt_before for human_review
app = workflow.compile(
    checkpointer=None, # For simplicity, no checkpointer here
    interrupt_before=["human_review"] # Pause before human_review node
)

print("\n--- Running LangGraph Human-in-the-Loop Example ---")
# Initial run will pause at 'human_review'
thread = {"document": "", "review_status": "", "reviewer_comments": ""}
for s in app.stream(thread, config={"configurable": {"thread_id": "1"}}):
    print(s)
    if "human_review" in s:
        print("--- PAUSED FOR HUMAN REVIEW ---")
        break # Simulate pausing

# Simulate human approving the document externally
print("\n--- Simulating Human Approval ---")
# In a real system, an API call would update the state and resume
# For demonstration, we manually update and continue
thread_state_after_pause = app.get_state(config={"configurable": {"thread_id": "1"}})
print(f"State after pause: {thread_state_after_pause.values}")

# Update the state with human input
updated_state = thread_state_after_pause.values
updated_state['review_status'] = "approved"
updated_state['reviewer_comments'] = "Content is accurate."

# Resume the graph with the updated state
for s in app.stream(updated_state, config={"configurable": {"thread_id": "1"}}):
    print(s)
    print("---")

# Simulate human rejecting the document
print("\n--- Simulating Human Rejection ---")
thread_reject = {"document": "", "review_status": "", "reviewer_comments": ""}
for s in app.stream(thread_reject, config={"configurable": {"thread_id": "2"}}):
    print(s)
    if "human_review" in s:
        print("--- PAUSED FOR HUMAN REVIEW (Rejection Scenario) ---")
        break

thread_state_after_reject_pause = app.get_state(config={"configurable": {"thread_id": "2"}})
updated_state_reject = thread_state_after_reject_pause.values
updated_state_reject['review_status'] = "rejected"
updated_state_reject['reviewer_comments'] = "Needs more detail on performance."

for s in app.stream(updated_state_reject, config={"configurable": {"thread_id": "2"}}):
    print(s)
    print("---")