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

如果你已经基于 LangGraph 构建过项目，学习如何快速将它们迁移到 CrewAI

你已经用 LangGraph 构建过 agent。你折腾过 StateGraph，连接过条件边，并在凌晨 2 点调试过状态字典。它确实能用——但在这个过程中，你大概也开始思考：有没有一条更适合走向生产环境的路。

有。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 / persistence 内置 memory（基于 LanceDB）

下面来看看它在实践中是什么样子。

示例 1：简单的顺序流水线

假设你要构建一个流水线：输入一个主题，先调研，再写摘要，最后格式化输出。下面是两个框架各自的实现方式。

LangGraph 的方式

  1. from typing import TypedDict
  2. from langgraph.graph import StateGraph, START, END
  4. class ResearchState(TypedDict):
  5.     topic: str
  6.     raw_research: str
  7.     summary: str
  8.     formatted_output: str
  10. def research_topic(state: ResearchState) -> dict:
  11.     # 调用 LLM 或搜索 API
  12.     result = llm.invoke(f"Research the topic: {state['topic']}")
  13.     return {"raw_research": result}
  15. def write_summary(state: ResearchState) -> dict:
  16.     result = llm.invoke(
  17.         f"Summarize this research:\n{state['raw_research']}"
  18.     )
  19.     return {"summary": result}
  21. def format_output(state: ResearchState) -> dict:
  22.     result = llm.invoke(
  23.         f"Format this summary as a polished article section:\n{state['summary']}"
  24.     )
  25.     return {"formatted_output": result}
  27. # 构建图
  28. graph = StateGraph(ResearchState)
  29. graph.add_node("research", research_topic)
  30. graph.add_node("summarize", write_summary)
  31. graph.add_node("format", format_output)
  33. graph.add_edge(START, "research")
  34. graph.add_edge("research", "summarize")
  35. graph.add_edge("summarize", "format")
  36. graph.add_edge("format", END)
  38. # 编译并运行
  39. app = graph.compile()
  40. result = app.invoke({"topic": "quantum computing advances in 2026"})
  41. print(result["formatted_output"])

你需要先定义函数，把它们注册为节点，再手动连接每一个状态转换。对于这样一个简单顺序流程来说，这套仪式感还是有点重。

CrewAI Flows 的方式

  1. from crewai import LLM, Agent, Crew, Process, Task
  2. from crewai.flow.flow import Flow, listen, start
  3. from pydantic import BaseModel
  5. llm = LLM(model="openai/gpt-5.2")
  7. class ResearchState(BaseModel):
  8.     topic: str = ""
  9.     raw_research: str = ""
  10.     summary: str = ""
  11.     formatted_output: str = ""
  13. class ResearchFlow(Flow[ResearchState]):
  14.     @start()
  15.     def research_topic(self):
  16.         # 方式 1：直接调用 LLM
  17.         result = llm.call(f"Research the topic: {self.state.topic}")
  18.         self.state.raw_research = result
  19.         return result
  21.     @listen(research_topic)
  22.     def write_summary(self, research_output):
  23.         # 方式 2：单个 agent
  24.         summarizer = Agent(
  25.             role="Research Summarizer",
  26.             goal="Produce concise, accurate summaries of research content",
  27.             backstory="You are an expert at distilling complex research into clear, "
  28.             "digestible summaries.",
  29.             llm=llm,
  30.             verbose=True,
  31.         )
  32.         result = summarizer.kickoff(
  33.             f"Summarize this research:\n{self.state.raw_research}"
  34.         )
  35.         self.state.summary = str(result)
  36.         return self.state.summary
  38.     @listen(write_summary)
  39.     def format_output(self, summary_output):
  40.         # 方式 3：完整的 crew（包含一个或多个 agent）
  41.         formatter = Agent(
  42.             role="Content Formatter",
  43.             goal="Transform research summaries into polished, publication-ready article sections",
  44.             backstory="You are a skilled editor with expertise in structuring and "
  45.             "presenting technical content for a general audience.",
  46.             llm=llm,
  47.             verbose=True,
  48.         )
  49.         format_task = Task(
  50.             description=f"Format this summary as a polished article section:\n{self.state.summary}",
  51.             expected_output="A well-structured, polished article section ready for publication.",
  52.             agent=formatter,
  53.         )
  54.         crew = Crew(
  55.             agents=[formatter],
  56.             tasks=[format_task],
  57.             process=Process.sequential,
  58.             verbose=True,
  59.         )
  60.         result = crew.kickoff()
  61.         self.state.formatted_output = str(result)
  62.         return self.state.formatted_output
  64. # 运行 flow
  65. flow = ResearchFlow()
  66. flow.state.topic = "quantum computing advances in 2026"
  67. result = flow.kickoff()
  68. print(flow.state.formatted_output)

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

示例 2：条件路由

这部分就更有意思了。假设你正在构建一个内容流水线，需要根据识别出的内容类型，将内容路由到不同的处理路径。

LangGraph 的方式

  1. from typing import TypedDict, Literal
  2. from langgraph.graph import StateGraph, START, END
  4. class ContentState(TypedDict):
  5.     input_text: str
  6.     content_type: str
  7.     result: str
  9. def classify_content(state: ContentState) -> dict:
  10.     content_type = llm.invoke(
  11.         f"Classify this content as 'technical', 'creative', or 'business':\n{state['input_text']}"
  12.     )
  13.     return {"content_type": content_type.strip().lower()}
  15. def process_technical(state: ContentState) -> dict:
  16.     result = llm.invoke(f"Process as technical doc:\n{state['input_text']}")
  17.     return {"result": result}
  19. def process_creative(state: ContentState) -> dict:
  20.     result = llm.invoke(f"Process as creative writing:\n{state['input_text']}")
  21.     return {"result": result}
  23. def process_business(state: ContentState) -> dict:
  24.     result = llm.invoke(f"Process as business content:\n{state['input_text']}")
  25.     return {"result": result}
  27. # 路由函数
  28. def route_content(state: ContentState) -> Literal["technical", "creative", "business"]:
  29.     return state["content_type"]
  31. # 构建图
  32. graph = StateGraph(ContentState)
  33. graph.add_node("classify", classify_content)
  34. graph.add_node("technical", process_technical)
  35. graph.add_node("creative", process_creative)
  36. graph.add_node("business", process_business)
  38. graph.add_edge(START, "classify")
  39. graph.add_conditional_edges(
  40.     "classify",
  41.     route_content,
  42.     {
  43.         "technical": "technical",
  44.         "creative": "creative",
  45.         "business": "business",
  46.     }
  47. )
  48. graph.add_edge("technical", END)
  49. graph.add_edge("creative", END)
  50. graph.add_edge("business", END)
  52. app = graph.compile()
  53. result = app.invoke({"input_text": "Explain how TCP handshakes work"})

你需要单独写一个路由函数、显式的条件边映射，以及为每条分支都写终止边。路由逻辑和产生路由决策的节点还是分离的。

CrewAI Flows 的方式

  1. from crewai import LLM, Agent
  2. from crewai.flow.flow import Flow, listen, router, start
  3. from pydantic import BaseModel
  5. llm = LLM(model="openai/gpt-5.2")
  7. class ContentState(BaseModel):
  8.     input_text: str = ""
  9.     content_type: str = ""
  10.     result: str = ""
  12. class ContentFlow(Flow[ContentState]):
  13.     @start()
  14.     def classify_content(self):
  15.         self.state.content_type = (
  16.             llm.call(
  17.                 f"Classify this content as 'technical', 'creative', or 'business':\n"
  18.                 f"{self.state.input_text}"
  19.             )
  20.             .strip()
  21.             .lower()
  22.         )
  23.         return self.state.content_type
  25.     @router(classify_content)
  26.     def route_content(self, classification):
  27.         if classification == "technical":
  28.             return "process_technical"
  29.         elif classification == "creative":
  30.             return "process_creative"
  31.         else:
  32.             return "process_business"
  34.     @listen("process_technical")
  35.     def handle_technical(self):
  36.         agent = Agent(
  37.             role="Technical Writer",
  38.             goal="Produce clear, accurate technical documentation",
  39.             backstory="You are an expert technical writer who specializes in "
  40.             "explaining complex technical concepts precisely.",
  41.             llm=llm,
  42.             verbose=True,
  43.         )
  44.         self.state.result = str(
  45.             agent.kickoff(f"Process as technical doc:\n{self.state.input_text}")
  46.         )
  48.     @listen("process_creative")
  49.     def handle_creative(self):
  50.         agent = Agent(
  51.             role="Creative Writer",
  52.             goal="Craft engaging and imaginative creative content",
  53.             backstory="You are a talented creative writer with a flair for "
  54.             "compelling storytelling and vivid expression.",
  55.             llm=llm,
  56.             verbose=True,
  57.         )
  58.         self.state.result = str(
  59.             agent.kickoff(f"Process as creative writing:\n{self.state.input_text}")
  60.         )
  62.     @listen("process_business")
  63.     def handle_business(self):
  64.         agent = Agent(
  65.             role="Business Writer",
  66.             goal="Produce professional, results-oriented business content",
  67.             backstory="You are an experienced business writer who communicates "
  68.             "strategy and value clearly to professional audiences.",
  69.             llm=llm,
  70.             verbose=True,
  71.         )
  72.         self.state.result = str(
  73.             agent.kickoff(f"Process as business content:\n{self.state.input_text}")
  74.         )
  76. flow = ContentFlow()
  77. flow.state.input_text = "Explain how TCP handshakes work"
  78. flow.kickoff()
  79. print(flow.state.result)

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

示例 3：在 Flows 中集成 AI Agent Crews

这里才是 CrewAI 真正发力的地方。Flows 不只是用来串联 LLM 调用——它们还可以编排完整的 Crews，也就是自治 agent 团队。而这一点是 LangGraph 原生并不具备的。

  1. from crewai import Agent, Task, Crew
  2. from crewai.flow.flow import Flow, listen, start
  3. from pydantic import BaseModel
  5. class ArticleState(BaseModel):
  6.     topic: str = ""
  7.     research: str = ""
  8.     draft: str = ""
  9.     final_article: str = ""
  11. class ArticleFlow(Flow[ArticleState]):
  13.     @start()
  14.     def run_research_crew(self):
  15.         """由完整的 Crew 负责研究。"""
  16.         researcher = Agent(
  17.             role="Senior Research Analyst",
  18.             goal=f"Produce comprehensive research on: {self.state.topic}",
  19.             backstory="You're a veteran analyst known for thorough, "
  20.                        "well-sourced research reports.",
  21.             llm="gpt-4o"
  22.         )
  24.         research_task = Task(
  25.             description=f"Research '{self.state.topic}' thoroughly. "
  26.                         "Cover key trends, data points, and expert opinions.",
  27.             expected_output="A detailed research brief with sources.",
  28.             agent=researcher
  29.         )
  31.         crew = Crew(agents=[researcher], tasks=[research_task])
  32.         result = crew.kickoff()
  33.         self.state.research = result.raw
  34.         return result.raw
  36.     @listen(run_research_crew)
  37.     def run_writing_crew(self, research_output):
  38.         """由另一个 Crew 负责写作。"""
  39.         writer = Agent(
  40.             role="Technical Writer",
  41.             goal="Write a compelling article based on provided research.",
  42.             backstory="You turn complex research into engaging, clear prose.",
  43.             llm="gpt-4o"
  44.         )
  46.         editor = Agent(
  47.             role="Senior Editor",
  48.             goal="Review and polish articles for publication quality.",
  49.             backstory="20 years of editorial experience at top tech publications.",
  50.             llm="gpt-4o"
  51.         )
  53.         write_task = Task(
  54.             description=f"Write an article based on this research:\n{self.state.research}",
  55.             expected_output="A well-structured draft article.",
  56.             agent=writer
  57.         )
  59.         edit_task = Task(
  60.             description="Review, fact-check, and polish the draft article.",
  61.             expected_output="A publication-ready article.",
  62.             agent=editor
  63.         )
  65.         crew = Crew(agents=[writer, editor], tasks=[write_task, edit_task])
  66.         result = crew.kickoff()
  67.         self.state.final_article = result.raw
  68.         return result.raw
  70. # 运行完整流水线
  71. flow = ArticleFlow()
  72. flow.state.topic = "The Future of Edge AI"
  73. flow.kickoff()
  74. print(flow.state.final_article)

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

而在 LangGraph 中，如果你想做出类似效果，通常需要在节点函数内部手动实现 agent 通信协议、工具调用循环和委派逻辑。并不是做不到，但这些底层管道几乎每次都得自己搭。

示例 4：并行执行与同步

真实世界的流水线经常需要把任务分叉出去，再把结果汇总回来。CrewAI Flows 通过 and_ 和 or_ 操作符优雅地处理了这一点。

  1. from crewai import LLM
  2. from crewai.flow.flow import Flow, and_, listen, start
  3. from pydantic import BaseModel
  5. llm = LLM(model="openai/gpt-5.2")
  7. class AnalysisState(BaseModel):
  8.     topic: str = ""
  9.     market_data: str = ""
  10.     tech_analysis: str = ""
  11.     competitor_intel: str = ""
  12.     final_report: str = ""
  14. class ParallelAnalysisFlow(Flow[AnalysisState]):
  15.     @start()
  16.     def start_method(self):
  17.         pass
  19.     @listen(start_method)
  20.     def gather_market_data(self):
  21.         # 你的 agent 式或确定性代码
  22.         pass
  24.     @listen(start_method)
  25.     def run_tech_analysis(self):
  26.         # 你的 agent 式或确定性代码
  27.         pass
  29.     @listen(start_method)
  30.     def gather_competitor_intel(self):
  31.         # 你的 agent 式或确定性代码
  32.         pass
  34.     @listen(and_(gather_market_data, run_tech_analysis, gather_competitor_intel))
  35.     def synthesize_report(self):
  36.         # 你的 agent 式或确定性代码
  37.         pass
  39. flow = ParallelAnalysisFlow()
  40. flow.state.topic = "AI-powered developer tools"
  41. flow.kickoff()

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

在 LangGraph 中，你则需要构建一个扇出 / 扇入模式，包含并行分支、同步节点，以及谨慎的状态合并——而这些都要通过显式连边完成。

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

除了语法更简洁之外，Flows 还带来了几个对生产环境非常关键的优势：

内置状态持久化。 Flow 状态由 LanceDB 支持，这意味着你的工作流可以在崩溃后恢复、支持续跑，并且能在多次运行中积累知识。LangGraph 则需要你单独配置 checkpointer。

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

一等公民级别的 agent 编排。 Crews 是原生能力。你可以定义具有角色、目标、背景故事和工具的 agents，并让它们在 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 项目：

  1. pip install crewai
  2. crewai create flow my_first_flow
  3. cd my_first_flow

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

  1. crewai run

接下来，你就可以添加自己的 agents，连接 listeners，然后发布上线。

最后的一点想法

LangGraph 让整个生态意识到，AI 工作流需要结构。这是一个很重要的启发。但 CrewAI Flows 把这个启发落到了更易写、更易读、而且在生产环境中更强大的形式上——尤其当你的工作流涉及多个协作 agent 时更是如此。

如果你构建的已经不只是单 agent 链路，那就应该认真看看 Flows。装饰器驱动的模型、原生的 Crew 集成，以及内置状态管理，意味着你可以少花时间搭底层管道，多花时间解决真正重要的问题。

从 crewai create flow 开始。你大概不会想再回头了。