从 LangGraph 迁移到 CrewAI：工程师实用指南

你已经用 LangGraph 构建过 agent。你和 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()
Checkpointer / 持久化	内置内存（基于 LanceDB）

让我们看看实际效果。

---

Demo 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:
    # 调用 LLM 或搜索 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}

# 构建图
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)

# 编译并运行
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):
        # 方案 1：直接 LLM 调用
        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):
        # 方案 2：单个 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):
        # 方案 3：完整 crew（可包含一个或多个 agent）
        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

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

注意差异：没有图构建、没有边连接、没有 compile 步骤。执行顺序直接声明在逻辑所在的位置。@start() 标记入口点，@listen(method_name) 将步骤串联起来。状态是一个真正的 Pydantic 模型，带类型安全、校验和 IDE 自动补全。

---

Demo 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}

# 路由函数
def route_content(state: ContentState) -> Literal["technical", "creative", "business"]:
    return state["content_type"]

# 构建图
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 语句，因为它本质上就是。

---

Demo 3：将 AI agent crews 集成到 Flows 中

这就是 CrewAI 真正强大的地方。Flows 不只是串联 LLM 调用 - 它们还可以编排完整的、自主的 agent Crews。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):
        """一个完整的 agent Crew 负责调研。"""
        researcher = Agent(
            role="高级研究分析师",
            goal=f"为以下主题产出全面调研：{self.state.topic}",
            backstory="你是一位经验丰富的分析师，以详尽且有据可查的研究报告著称。",
            llm="gpt-4o"
        )

        research_task = Task(
            description=f"彻底研究 '{self.state.topic}'。涵盖关键趋势、数据点和专家观点。",
            expected_output="一份带来源的详细研究简报。",
            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):
        """另一个 Crew 负责写作。"""
        writer = Agent(
            role="技术写作者",
            goal="基于提供的调研撰写一篇引人入胜的文章。",
            backstory="你擅长把复杂研究转化为清晰、易读的文字。",
            llm="gpt-4o"
        )

        editor = Agent(
            role="资深编辑",
            goal="审阅并润色文章，使其达到发布质量。",
            backstory="在顶级科技媒体拥有 20 年编辑经验。",
            llm="gpt-4o"
        )

        write_task = Task(
            description=f"根据以下调研撰写文章：\n{self.state.research}",
            expected_output="一篇结构良好的草稿文章。",
            agent=writer
        )

        edit_task = Task(
            description="审阅、核查并润色草稿文章。",
            expected_output="一篇可发布的文章。",
            agent=editor
        )

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

# 运行完整流水线
flow = ArticleFlow()
flow.state.topic = "边缘 AI 的未来"
flow.kickoff()
print(flow.state.final_article)

这就是关键洞察：Flows 提供编排层，Crews 提供智能层。 Flow 中的每一步都可以启动一个完整的协作 agent 团队，每个团队都有自己的角色、目标和工具。你同时获得了结构化、可预测的控制流和自主的 agent 协作 - 两全其美。

在 LangGraph 中，要实现类似能力，就得在 node 函数里手工实现 agent 通信协议、工具调用循环和委托逻辑。它当然可以做到，但你每次都得从头搭这些管线。

---

Demo 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):
        # 你的 agentic 或确定性代码
        pass

    @listen(start_method)
    def run_tech_analysis(self):
        # 你的 agentic 或确定性代码
        pass

    @listen(start_method)
    def gather_competitor_intel(self):
        # 你的 agentic 或确定性代码
        pass

    @listen(and_(gather_market_data, run_tech_analysis, gather_competitor_intel))
    def synthesize_report(self):
        # 你的 agentic 或确定性代码
        pass

flow = ParallelAnalysisFlow()
flow.state.topic = "AI 驱动的开发者工具"
flow.kickoff()

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

在 LangGraph 中，你需要自己搭建 fan-out / fan-in 模式、一个同步节点，以及仔细的状态合并 - 全都要通过显式边来连接。

---

为什么生产环境更适合 CrewAI Flows

除了语法更简洁，Flows 还带来了几个对生产很关键的优势：

内置状态持久化。 Flow 状态由 LanceDB 提供支持，这意味着工作流可以在崩溃后恢复、继续执行，并跨运行积累知识。LangGraph 需要你单独配置 checkpointer。

类型安全的状态管理。 Pydantic 模型默认就提供验证、序列化和 IDE 支持。LangGraph 的 TypedDict 状态不会在运行时做验证。

一等公民的 agent 编排。 Crews 是原生的基本单元。你定义带角色、目标、背景故事和工具的 agent，它们会在 Flow 的结构化边界内自主协作。无需重新发明多 agent 协调机制。

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

CLI 集成。 用 crewai run 运行 flow。没有单独的编译步骤，也没有图序列化。你的 Flow 就是一个 Python 类，而且会像 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. 思考 Crews 的位置。 任何包含复杂多步骤 agent 逻辑的节点，都是提取成 Crew 的候选。这往往能带来最大的质量提升。

---

开始动手

安装 CrewAI 并脚手架一个新的 Flow 项目：

pip install crewai
crewai create flow my_first_flow
cd my_first_flow

这会生成一个项目结构，其中包含一个可立即编辑的 Flow 类、配置文件，以及一个已经设置好 type = "flow" 的 pyproject.toml。然后运行：

crewai run

从这里开始，添加你的 agent，连接监听器，然后发布它。

---

最后想法

LangGraph 让整个生态认识到 AI 工作流需要结构。这是一个重要教训。但 CrewAI Flows 把这个教训做成了更快编写、更易阅读、并且在生产环境中更强大的形式 - 尤其适合涉及多个协作 agent 的工作流。

如果你正在构建任何超出单个 agent 链条的东西，值得认真看看 Flows。装饰器驱动的模型、原生 Crew 集成和内置状态管理，意味着你会花更少时间处理管线细节，把更多时间用在真正重要的问题上。

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