How to migrate a LangGraph pipeline to CrewAI Flows

LangGraph pipelines are graphs: you register nodes, wire edges, compile, then invoke. CrewAI Flows maps the same sequential and branching work onto a Flow class whose methods use @start, @listen, and @router. There is no graph.compile() step—you call flow.kickoff(). This walkthrough follows CrewAI’s official migration guide and rebuilds both of its core demos: a research → summarize → format pipeline, and a classify-then-route pipeline.

Map concepts before you copy code

LangGraph asks you to think in graphs: nodes, edges, and state dictionaries. CrewAI Flows asks you to think in events: methods that start work, methods that listen for results, and methods that route execution. Topology comes from decorator annotations rather than explicit add_node / add_edge construction.

Convert each TypedDict to a Pydantic BaseModel and give every field a default. Inside methods, read and write self.state.field instead of state["field"]. Pydantic validates at runtime; TypedDict does not.

Demo 1: research → summarize → format

The first official demo takes a topic, researches it, writes a summary, and formats the output.

In LangGraph you define functions, register them as nodes, and manually wire every transition, including START and END. Then you compile and invoke:

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class ResearchState(TypedDict):
    topic: str
    raw_research: str
    summary: str
    formatted_output: str

def research_topic(state: ResearchState) -> dict:
    # Call an LLM or search API
    result = llm.invoke(f"Research the topic: {state['topic']}")
    return {"raw_research": result}

def write_summary(state: ResearchState) -> dict:
    result = llm.invoke(
        f"Summarize this research:\n{state['raw_research']}"
    )
    return {"summary": result}

def format_output(state: ResearchState) -> dict:
    result = llm.invoke(
        f"Format this summary as a polished article section:\n{state['summary']}"
    )
    return {"formatted_output": result}

# Build the graph
graph = StateGraph(ResearchState)
graph.add_node("research", research_topic)
graph.add_node("summarize", write_summary)
graph.add_node("format", format_output)

graph.add_edge(START, "research")
graph.add_edge("research", "summarize")
graph.add_edge("summarize", "format")
graph.add_edge("format", END)

# Compile and run
app = graph.compile()
result = app.invoke({"topic": "quantum computing advances in 2026"})
print(result["formatted_output"])

That is a lot of ceremony for a straight sequence.

The Flow version declares order next to the logic. @start() marks the entry point. @listen(method_name) chains the next step. The same class can mix a direct LLM call, a single Agent, and a Crew:

from crewai import LLM, Agent, Crew, Process, Task
from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel

llm = LLM(model="openai/gpt-5.2")

class ResearchState(BaseModel):
    topic: str = ""
    raw_research: str = ""
    summary: str = ""
    formatted_output: str = ""

class ResearchFlow(Flow[ResearchState]):
    @start()
    def research_topic(self):
        # Option 1: Direct LLM call
        result = llm.call(f"Research the topic: {self.state.topic}")
        self.state.raw_research = result
        return result

    @listen(research_topic)
    def write_summary(self, research_output):
        # Option 2: A single agent
        summarizer = Agent(
            role="Research Summarizer",
            goal="Produce concise, accurate summaries of research content",
            backstory="You are an expert at distilling complex research into clear, "
            "digestible summaries.",
            llm=llm,
            verbose=True,
        )
        result = summarizer.kickoff(
            f"Summarize this research:\n{self.state.raw_research}"
        )
        self.state.summary = str(result)
        return self.state.summary

    @listen(write_summary)
    def format_output(self, summary_output):
        # Option 3: a complete crew (with one or more agents)
        formatter = Agent(
            role="Content Formatter",
            goal="Transform research summaries into polished, publication-ready article sections",
            backstory="You are a skilled editor with expertise in structuring and "
            "presenting technical content for a general audience.",
            llm=llm,
            verbose=True,
        )
        format_task = Task(
            description=f"Format this summary as a polished article section:\n{self.state.summary}",
            expected_output="A well-structured, polished article section ready for publication.",
            agent=formatter,
        )
        crew = Crew(
            agents=[formatter],
            tasks=[format_task],
            process=Process.sequential,
            verbose=True,
        )
        result = crew.kickoff()
        self.state.formatted_output = str(result)
        return self.state.formatted_output

# Run the flow
flow = ResearchFlow()
flow.state.topic = "quantum computing advances in 2026"
result = flow.kickoff()
print(flow.state.formatted_output)

Set flow.state.topic, call flow.kickoff(), and read flow.state.formatted_output. No graph object, no edge list, no compile.

Demo 2: classify, then route

The second demo classifies content as technical, creative, or business, then sends it down a matching path.

LangGraph needs a separate routing function, add_conditional_edges with a mapping dictionary, and an END edge on every branch. The routing logic sits apart from the node that produced the decision:

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END

class ContentState(TypedDict):
    input_text: str
    content_type: str
    result: str

def classify_content(state: ContentState) -> dict:
    content_type = llm.invoke(
        f"Classify this content as 'technical', 'creative', or 'business':\n{state['input_text']}"
    )
    return {"content_type": content_type.strip().lower()}

def process_technical(state: ContentState) -> dict:
    result = llm.invoke(f"Process as technical doc:\n{state['input_text']}")
    return {"result": result}

def process_creative(state: ContentState) -> dict:
    result = llm.invoke(f"Process as creative writing:\n{state['input_text']}")
    return {"result": result}

def process_business(state: ContentState) -> dict:
    result = llm.invoke(f"Process as business content:\n{state['input_text']}")
    return {"result": result}

# Routing function
def route_content(state: ContentState) -> Literal["technical", "creative", "business"]:
    return state["content_type"]

# Build the graph
graph = StateGraph(ContentState)
graph.add_node("classify", classify_content)
graph.add_node("technical", process_technical)
graph.add_node("creative", process_creative)
graph.add_node("business", process_business)

graph.add_edge(START, "classify")
graph.add_conditional_edges(
    "classify",
    route_content,
    {
        "technical": "technical",
        "creative": "creative",
        "business": "business",
    }
)
graph.add_edge("technical", END)
graph.add_edge("creative", END)
graph.add_edge("business", END)

app = graph.compile()
result = app.invoke({"input_text": "Explain how TCP handshakes work"})

In Flows, @router() is the decision point. It returns a string that matches a listener—no mapping dict. The branch reads like a Python if because it is one:

from crewai import LLM, Agent
from crewai.flow.flow import Flow, listen, router, start
from pydantic import BaseModel

llm = LLM(model="openai/gpt-5.2")

class ContentState(BaseModel):
    input_text: str = ""
    content_type: str = ""
    result: str = ""

class ContentFlow(Flow[ContentState]):
    @start()
    def classify_content(self):
        self.state.content_type = (
            llm.call(
                f"Classify this content as 'technical', 'creative', or 'business':\n"
                f"{self.state.input_text}"
            )
            .strip()
            .lower()
        )
        return self.state.content_type

    @router(classify_content)
    def route_content(self, classification):
        if classification == "technical":
            return "process_technical"
        elif classification == "creative":
            return "process_creative"
        else:
            return "process_business"

    @listen("process_technical")
    def handle_technical(self):
        agent = Agent(
            role="Technical Writer",
            goal="Produce clear, accurate technical documentation",
            backstory="You are an expert technical writer who specializes in "
            "explaining complex technical concepts precisely.",
            llm=llm,
            verbose=True,
        )
        self.state.result = str(
            agent.kickoff(f"Process as technical doc:\n{self.state.input_text}")
        )

    @listen("process_creative")
    def handle_creative(self):
        agent = Agent(
            role="Creative Writer",
            goal="Craft engaging and imaginative creative content",
            backstory="You are a talented creative writer with a flair for "
            "compelling storytelling and vivid expression.",
            llm=llm,
            verbose=True,
        )
        self.state.result = str(
            agent.kickoff(f"Process as creative writing:\n{self.state.input_text}")
        )

    @listen("process_business")
    def handle_business(self):
        agent = Agent(
            role="Business Writer",
            goal="Produce professional, results-oriented business content",
            backstory="You are an experienced business writer who communicates "
            "strategy and value clearly to professional audiences.",
            llm=llm,
            verbose=True,
        )
        self.state.result = str(
            agent.kickoff(f"Process as business content:\n{self.state.input_text}")
        )

flow = ContentFlow()
flow.state.input_text = "Explain how TCP handshakes work"
flow.kickoff()
print(flow.state.result)

@listen("process_technical"), @listen("process_creative"), and @listen("process_business") bind those route strings. After flow.kickoff(), the processed text is on flow.state.result.

Extract heavy nodes into Crews

Flows orchestrate; Crews supply the agent team. Each listen step can spin up agents with roles, goals, backstories, and tools. The migration guide’s third listing chains a research crew into a writer-and-editor crew on an ArticleFlow. When a LangGraph node already hides multi-step agent logic, that node is the first place to extract a Crew.

Cheat sheet

1. Map your state. Convert your TypedDict to a Pydantic BaseModel. Add default values for all fields.
2. Convert nodes to methods. Each add_node function becomes a method on your Flow subclass. Replace state["field"] reads with self.state.field.
3. Replace edges with decorators. Your add_edge(START, "first_node") becomes @start() on the first method. Sequential add_edge("a", "b") becomes @listen(a) on method b.
4. Replace conditional edges with @router. Your routing function and add_conditional_edges() mapping become a single @router() method that returns a route string.
5. Replace compile + invoke with kickoff. Drop graph.compile(). Call flow.kickoff() instead.
6. Consider where Crews fit. Any node where you have complex multi-step agent logic is a candidate for extraction into a Crew.

Scaffold a project when you are ready to run one:

pip install crewai
crewai create flow my_first_flow
cd my_first_flow

That generates a Flow class, configuration files, and a pyproject.toml with type = "flow" already set. Run it with:

crewai run

Convert one LangGraph graph you already trust—start with its TypedDict and node list—then scaffold a Flow and run crewai run. Keep the official migrating-from-LangGraph page open for the full listings, including the Crew-inside-Flow example.

Sources

• Migrating from LangGraph