从 LangGraph 迁移到 CrewAI：面向工程师的实践指南

如果你已经用 LangGraph 构建过项目，了解如何快速把项目迁移到 CrewAI

你已经用 LangGraph 构建过智能体。你跟 StateGraph 较过劲，接过条件边，也在凌晨两点调试过状态字典。它能用——但在某个时刻，你开始怀疑是否有更好的通往生产环境的路径。

有的。CrewAI Flows 给你同样的能力——事件驱动编排、条件路由、共享状态——而样板代码大幅减少，并且其心智模型能清晰地映射到你实际思考多步骤 AI 工作流的方式。

本文并排讲解核心概念，展示真实的代码对比，并说明为什么 CrewAI Flows 是你接下来会想使用的框架。

---

心智模型的转变

LangGraph 要求你以图的方式思考：节点、边和状态字典。每个工作流都是有向图，你需要显式地连接计算步骤之间的转移。它很强大，但这层抽象带有开销——尤其是当你的工作流本质上是顺序执行、只有少数决策点时。

CrewAI Flows 要求你以事件的方式思考：启动事情的方法、监听结果的方法以及路由执行的方法。工作流的拓扑从装饰器标注中涌现，而不是来自显式的图构建。这不仅仅是语法糖——它改变了你设计、阅读和维护流水线的方式。

下面是核心映射：

LangGraph 概念	CrewAI Flows 对应项
StateGraph 类	Flow 类
add_node()	用 @start、@listen 装饰的方法
add_edge() / add_conditional_edges()	@listen() / @router() 装饰器
TypedDict 状态	Pydantic BaseModel 状态
START / END 常量	@start() 装饰器 / 方法自然返回
graph.compile()	flow.kickoff()
检查点器 / 持久化	内置记忆（基于 LanceDB）

我们来看看这在实践中是什么样子。

---

演示 1：一个简单的顺序流水线

设想你在构建一个流水线：接收一个主题，研究它，写一份摘要，并格式化输出。下面是每个框架的处理方式。

LangGraph 方法

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"])

你需要定义函数、把它们注册为节点，并手动连接每一次转换。对于这样简单的顺序流程来说，仪式感太重了。

CrewAI Flows 方法

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)

注意不同之处：没有图构建、没有边连接、没有编译步骤。执行顺序就声明在逻辑所在的位置。@start() 标记入口点，@listen(method_name) 把各个步骤串联起来。状态是一个规范的 Pydantic 模型，具备类型安全、校验和 IDE 自动补全。

---

演示 2：条件路由

有趣的地方从这里开始。假设你要构建一条内容流水线，根据检测到的内容类型路由到不同的处理路径。

LangGraph 方法

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"})

你需要一个单独的路由函数、显式的条件边映射，以及为每个分支准备的终止边。路由逻辑与产生路由决策的节点彼此解耦。

CrewAI Flows 方法

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)

@router() 装饰器把方法变成决策点。它返回一个与监听器匹配的字符串——不需要映射字典，也不需要单独的路由函数。分支逻辑读起来就像一个 Python if 语句，因为它确实就是。

---

演示 3：将 AI 智能体团队集成到流程中

CrewAI 真正的威力在这里显现。流程不只是用来串联 LLM 调用——它们还能编排由自主智能体组成的完整团队。这是 LangGraph 完全没有原生等价物的能力。

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

class ArticleState(BaseModel):
    topic: str = ""
    research: str = ""
    draft: str = ""
    final_article: str = ""

class ArticleFlow(Flow[ArticleState]):

    @start()
    def run_research_crew(self):
        """A full Crew of agents handles research."""
        researcher = Agent(
            role="Senior Research Analyst",
            goal=f"Produce comprehensive research on: {self.state.topic}",
            backstory="You're a veteran analyst known for thorough, "
                       "well-sourced research reports.",
            llm="gpt-4o"
        )

        research_task = Task(
            description=f"Research '{self.state.topic}' thoroughly. "
                        "Cover key trends, data points, and expert opinions.",
            expected_output="A detailed research brief with sources.",
            agent=researcher
        )

        crew = Crew(agents=[researcher], tasks=[research_task])
        result = crew.kickoff()
        self.state.research = result.raw
        return result.raw

    @listen(run_research_crew)
    def run_writing_crew(self, research_output):
        """A different Crew handles writing."""
        writer = Agent(
            role="Technical Writer",
            goal="Write a compelling article based on provided research.",
            backstory="You turn complex research into engaging, clear prose.",
            llm="gpt-4o"
        )

        editor = Agent(
            role="Senior Editor",
            goal="Review and polish articles for publication quality.",
            backstory="20 years of editorial experience at top tech publications.",
            llm="gpt-4o"
        )

        write_task = Task(
            description=f"Write an article based on this research:\n{self.state.research}",
            expected_output="A well-structured draft article.",
            agent=writer
        )

        edit_task = Task(
            description="Review, fact-check, and polish the draft article.",
            expected_output="A publication-ready article.",
            agent=editor
        )

        crew = Crew(agents=[writer, editor], tasks=[write_task, edit_task])
        result = crew.kickoff()
        self.state.final_article = result.raw
        return result.raw

# Run the full pipeline
flow = ArticleFlow()
flow.state.topic = "The Future of Edge AI"
flow.kickoff()
print(flow.state.final_article)

关键洞见在于：流程提供编排层，团队提供智能层。 流程中的每一步都可以启动一整支协作智能体团队，每个智能体都有自己的角色、目标和工具。你既得到结构化、可预测的控制流，又得到自主智能体协作——两全其美。

在 LangGraph 中，要实现类似的效果，就得在节点函数里手动实现智能体通信协议、工具调用循环和委派逻辑。这虽然可行，但每次都要从头搭建这些管道。

---

演示 4：并行执行与同步

现实世界的流水线常常需要把工作分发给多个分支再汇合结果。CrewAI Flows 通过 and_ 和 or_ 运算符优雅地处理这一点。

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

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

class AnalysisState(BaseModel):
    topic: str = ""
    market_data: str = ""
    tech_analysis: str = ""
    competitor_intel: str = ""
    final_report: str = ""

class ParallelAnalysisFlow(Flow[AnalysisState]):
    @start()
    def start_method(self):
        pass

    @listen(start_method)
    def gather_market_data(self):
        # Your agentic or deterministic code
        pass

    @listen(start_method)
    def run_tech_analysis(self):
        # Your agentic or deterministic code
        pass

    @listen(start_method)
    def gather_competitor_intel(self):
        # Your agentic or deterministic code
        pass

    @listen(and_(gather_market_data, run_tech_analysis, gather_competitor_intel))
    def synthesize_report(self):
        # Your agentic or deterministic code
        pass

flow = ParallelAnalysisFlow()
flow.state.topic = "AI-powered developer tools"
flow.kickoff()

多个 @start() 装饰器会并行触发。@listen 装饰器上的 and_() 组合器确保 synthesize_report 只在全部三个上游方法完成后才执行。还有 or_()，用于你希望在任意上游任务完成时就继续的场景。

在 LangGraph 中，你需要用并行分支、同步节点和谨慎的状态合并来构建扇出/扇入模式——全部都要通过边显式连接。

---

为什么生产环境选择 CrewAI Flows

除了更简洁的语法，流程还带来若干对生产至关重要的优势：

内置状态持久化。 流程状态由 LanceDB 支撑，这意味着你的工作流可以在崩溃后存续、恢复运行，并在多次运行之间积累知识。LangGraph 则要求你配置单独的检查点器。

类型安全的状态管理。 Pydantic 模型开箱即提供校验、序列化和 IDE 支持。LangGraph 的 TypedDict 状态在运行时不做校验。

一等的智能体编排能力。 团队是原生原语。你用角色、目标、背景故事和工具来定义智能体——它们会在流程的结构化外壳内自主协作。无需重新发明多智能体协同。

更简单的心智模型。 装饰器声明意图。@start 表示“从这里开始”。@listen(x) 表示“在 x 之后运行”。@router(x) 表示“在 x 之后决定往哪里走”。代码读起来就像它所描述的工作流。

CLI 集成。 用 crewai run 运行流程。没有单独的编译步骤，没有图的序列化。你的流程就是一个 Python 类，它就像类一样运行。

---

迁移速查表

如果你手上有一个 LangGraph 代码库并且想迁移到 CrewAI Flows，这里有一份实用的转换指南：

1. 梳理你的状态。 把 TypedDict 转换成 Pydantic BaseModel。为所有字段添加默认值。
2. 把节点转换为方法。 每个 add_node 函数都变成 Flow 子类上的一个方法。把读取 state["field"] 改为 self.state.field。
3. 用装饰器替换边。 你的 add_edge(START, "first_node") 变成第一个方法上的 @start()。顺序的 add_edge("a", "b") 变成方法 b 上的 @listen(a)。
4. 用 @router 替换条件边。 你的路由函数和 add_conditional_edges() 映射合并为一个返回路由字符串的 @router() 方法。
5. 用 kickoff 替换 compile + invoke。 去掉 graph.compile()。改为调用 flow.kickoff()。
6. 考虑团队适合放在哪里。 任何包含复杂多步智能体逻辑的节点，都是提取为团队的候选。你会在这里看到最大的质量提升。

---

开始使用

安装 CrewAI 并搭建一个新的 Flow 项目：

pip install crewai
crewai create flow my_first_flow
cd my_first_flow

这会生成一个项目结构，其中包含可直接编辑的 Flow 类、配置文件，以及已经设置好 type = "flow" 的 pyproject.toml。用下面的命令运行它：

crewai run

接下来，添加你的智能体、接好监听器，然后发布。

---

最后总结

LangGraph 让整个生态认识到：AI 工作流需要结构。这是重要的一课。但 CrewAI Flows 接过这一课，以一种编写更快、更易读、在生产中更强大的形式交付出来——尤其是当你的工作流涉及多个协作智能体时。

如果你要构建的不只是单智能体链路的任何东西，都值得认真看看 Flows。装饰器驱动的模型、原生的团队集成和内置的状态管理，意味着你会少花时间在管道上，多花时间在真正重要的问题上。

从 crewai create flow 开始。你不会再回头看。