掌握 Flow 状态管理

一份关于如何在 CrewAI Flows 中管理、持久化并利用状态来构建健壮 AI 应用的综合指南。

理解 Flows 中状态的力量

状态管理是任何复杂 AI 工作流的核心支柱。在 CrewAI Flows 中，状态系统使你能够维护上下文、在步骤之间共享数据，并构建复杂的应用逻辑。掌握状态管理，对于创建可靠、可维护且强大的 AI 应用至关重要。

本指南将带你系统了解在 CrewAI Flows 中管理状态所需掌握的一切内容，从基础概念到高级技巧，并配有实用代码示例。

为什么状态管理很重要

有效的状态管理可以帮助你：

  1. 在执行步骤之间保持上下文 —— 在工作流的不同阶段之间无缝传递信息
  2. 构建复杂的条件逻辑 —— 基于累积数据做出决策
  3. 创建可持久化的应用 —— 保存并恢复工作流进度
  4. 优雅地处理错误 —— 实现恢复模式，让应用更健壮
  5. 扩展你的应用 —— 通过合理的数据组织支持复杂工作流
  6. 支持对话式应用 —— 存储并访问对话历史，以支持具备上下文感知能力的 AI 交互

接下来，我们将深入探讨如何高效利用这些能力。

状态管理基础

Flow 状态生命周期

在 CrewAI Flows 中，状态遵循一个可预测的生命周期：

  1. 初始化 —— 当一个 flow 被创建时，它的状态会被初始化（可以是一个空字典，也可以是一个 Pydantic 模型实例）
  2. 修改 —— Flow 方法在执行过程中访问并修改状态
  3. 传递 —— 状态会在 flow 方法之间自动传递
  4. 持久化（可选）—— 状态可以被保存到存储中，并在之后重新获取
  5. 完成 —— 最终状态会反映所有已执行方法所累积产生的变更

理解这个生命周期，对于设计高效 flow 至关重要。

两种状态管理方式

CrewAI 提供了两种管理 flow 状态的方式：

  1. 非结构化状态 —— 使用类字典对象，强调灵活性
  2. 结构化状态 —— 使用 Pydantic 模型，强调类型安全和校验

接下来我们会详细介绍这两种方式。

非结构化状态管理

非结构化状态采用类似字典的方式，适用于简单直接的应用，具备较高的灵活性。

它是如何工作的

使用非结构化状态时：

  • 你通过 self.state 访问状态，它的行为类似于字典
  • 你可以在任意时刻自由添加、修改或删除键
  • 所有状态都会自动对所有 flow 方法可用

基础示例

下面是一个简单的非结构化状态管理示例：

  1. from crewai.flow.flow import Flow, listen, start
  3. class UnstructuredStateFlow(Flow):
  4.     @start()
  5.     def initialize_data(self):
  6.         print("Initializing flow data")
  7.         # Add key-value pairs to state
  8.         self.state["user_name"] = "Alex"
  9.         self.state["preferences"] = {
  10.             "theme": "dark",
  11.             "language": "English"
  12.         }
  13.         self.state["items"] = []
  15.         # The flow state automatically gets a unique ID
  16.         print(f"Flow ID: {self.state['id']}")
  18.         return "Initialized"
  20.     @listen(initialize_data)
  21.     def process_data(self, previous_result):
  22.         print(f"Previous step returned: {previous_result}")
  24.         # Access and modify state
  25.         user = self.state["user_name"]
  26.         print(f"Processing data for {user}")
  28.         # Add items to a list in state
  29.         self.state["items"].append("item1")
  30.         self.state["items"].append("item2")
  32.         # Add a new key-value pair
  33.         self.state["processed"] = True
  35.         return "Processed"
  37.     @listen(process_data)
  38.     def generate_summary(self, previous_result):
  39.         # Access multiple state values
  40.         user = self.state["user_name"]
  41.         theme = self.state["preferences"]["theme"]
  42.         items = self.state["items"]
  43.         processed = self.state.get("processed", False)
  45.         summary = f"User {user} has {len(items)} items with {theme} theme. "
  46.         summary += "Data is processed." if processed else "Data is not processed."
  48.         return summary
  50. # Run the flow
  51. flow = UnstructuredStateFlow()
  52. result = flow.kickoff()
  53. print(f"Final result: {result}")
  54. print(f"Final state: {flow.state}")

何时使用非结构化状态

非结构化状态适用于：

  • 快速原型开发和简单 flow
  • 状态需求会动态变化的场景
  • 事先无法明确状态结构的情况
  • 状态要求较简单的 flows

虽然它非常灵活，但非结构化状态缺乏类型检查和模式校验，在复杂应用中更容易引发错误。

结构化状态管理

结构化状态通过 Pydantic 模型定义 flow 状态的模式，从而提供类型安全、校验能力以及更好的开发体验。

它是如何工作的

使用结构化状态时：

  • 你需要定义一个表示状态结构的 Pydantic 模型
  • 你将该模型类型作为类型参数传递给 Flow 类
  • 你通过 self.state 访问状态，它的行为类似于一个 Pydantic 模型实例
  • 所有字段都会根据定义的类型自动校验
  • 你可以获得 IDE 自动补全和类型检查支持

基础示例

下面是如何实现结构化状态管理的示例：

  1. from crewai.flow.flow import Flow, listen, start
  2. from pydantic import BaseModel, Field
  3. from typing import List, Dict, Optional
  5. # Define your state model
  6. class UserPreferences(BaseModel):
  7.     theme: str = "light"
  8.     language: str = "English"
  10. class AppState(BaseModel):
  11.     user_name: str = ""
  12.     preferences: UserPreferences = UserPreferences()
  13.     items: List[str] = []
  14.     processed: bool = False
  15.     completion_percentage: float = 0.0
  17. # Create a flow with typed state
  18. class StructuredStateFlow(Flow[AppState]):
  19.     @start()
  20.     def initialize_data(self):
  21.         print("Initializing flow data")
  22.         # Set state values (type-checked)
  23.         self.state.user_name = "Taylor"
  24.         self.state.preferences.theme = "dark"
  26.         # The ID field is automatically available
  27.         print(f"Flow ID: {self.state.id}")
  29.         return "Initialized"
  31.     @listen(initialize_data)
  32.     def process_data(self, previous_result):
  33.         print(f"Processing data for {self.state.user_name}")
  35.         # Modify state (with type checking)
  36.         self.state.items.append("item1")
  37.         self.state.items.append("item2")
  38.         self.state.processed = True
  39.         self.state.completion_percentage = 50.0
  41.         return "Processed"
  43.     @listen(process_data)
  44.     def generate_summary(self, previous_result):
  45.         # Access state (with autocompletion)
  46.         summary = f"User {self.state.user_name} has {len(self.state.items)} items "
  47.         summary += f"with {self.state.preferences.theme} theme. "
  48.         summary += "Data is processed." if self.state.processed else "Data is not processed."
  49.         summary += f" Completion: {self.state.completion_percentage}%"
  51.         return summary
  53. # Run the flow
  54. flow = StructuredStateFlow()
  55. result = flow.kickoff()
  56. print(f"Final result: {result}")
  57. print(f"Final state: {flow.state}")

结构化状态的优势

使用结构化状态有以下几个优点：

  1. 类型安全 —— 在开发阶段就能发现类型错误
  2. 自说明性 —— 状态模型能清晰说明可用的数据有哪些
  3. 校验能力 —— 自动校验数据类型和约束条件
  4. IDE 支持 —— 可获得自动补全和内联文档
  5. 默认值 —— 可以轻松为缺失数据定义回退值

何时使用结构化状态

结构化状态推荐用于：

  • 拥有明确数据模式的复杂 flows
  • 多位开发者共同维护同一代码的团队项目
  • 对数据校验要求较高的应用
  • 需要强制执行特定数据类型和约束的 flows

自动状态 ID

无论是非结构化状态还是结构化状态，都会自动获得一个唯一标识符（ UUID ），以便追踪和管理状态实例。

它是如何工作的

  • 对于非结构化状态，ID 可通过 self.state["id"] 访问
  • 对于结构化状态，ID 可通过 self.state.id 访问
  • 这个 ID 会在 flow 创建时自动生成
  • 在 flow 的整个生命周期中，这个 ID 保持不变
  • 这个 ID 可用于追踪、日志记录以及检索已持久化的状态

在实现持久化或追踪多个 flow 执行时，这个 UUID 尤其有价值。

动态状态更新

无论你使用结构化状态还是非结构化状态，都可以在 flow 执行过程中动态更新状态。

在步骤之间传递数据

Flow 方法可以返回值，这些返回值会作为参数传递给监听它们的方法：

  1. from crewai.flow.flow import Flow, listen, start
  3. class DataPassingFlow(Flow):
  4.     @start()
  5.     def generate_data(self):
  6.         # This return value will be passed to listening methods
  7.         return "Generated data"
  9.     @listen(generate_data)
  10.     def process_data(self, data_from_previous_step):
  11.         print(f"Received: {data_from_previous_step}")
  12.         # You can modify the data and pass it along
  13.         processed_data = f"{data_from_previous_step} - processed"
  14.         # Also update state
  15.         self.state["last_processed"] = processed_data
  16.         return processed_data
  18.     @listen(process_data)
  19.     def finalize_data(self, processed_data):
  20.         print(f"Received processed data: {processed_data}")
  21.         # Access both the passed data and state
  22.         last_processed = self.state.get("last_processed", "")
  23.         return f"Final: {processed_data} (from state: {last_processed})"

这种模式可以将直接数据传递与状态更新结合起来，带来最大的灵活性。

持久化 Flow 状态

CrewAI 最强大的特性之一，就是能够在多次执行之间持久化 flow 状态。这使得工作流可以被暂停、恢复，甚至在失败后继续执行。

@persist() 装饰器

@persist() 装饰器可以自动完成状态持久化，在执行过程中的关键节点保存 flow 状态。

类级别持久化

当 @persist() 应用在类级别时，它会在每个方法执行后保存状态：

  1. from crewai.flow.flow import Flow, listen, start
  2. from crewai.flow.persistence import persist
  3. from pydantic import BaseModel
  5. class CounterState(BaseModel):
  6.     value: int = 0
  8. @persist()  # Apply to the entire flow class
  9. class PersistentCounterFlow(Flow[CounterState]):
  10.     @start()
  11.     def increment(self):
  12.         self.state.value += 1
  13.         print(f"Incremented to {self.state.value}")
  14.         return self.state.value
  16.     @listen(increment)
  17.     def double(self, value):
  18.         self.state.value = value * 2
  19.         print(f"Doubled to {self.state.value}")
  20.         return self.state.value
  22. # First run
  23. flow1 = PersistentCounterFlow()
  24. result1 = flow1.kickoff()
  25. print(f"First run result: {result1}")
  27. # Second run - state is automatically loaded
  28. flow2 = PersistentCounterFlow()
  29. result2 = flow2.kickoff()
  30. print(f"Second run result: {result2}")  # Will be higher due to persisted state

方法级别持久化

如果你需要更细粒度的控制，可以将 @persist() 应用到特定方法上：

  1. from crewai.flow.flow import Flow, listen, start
  2. from crewai.flow.persistence import persist
  4. class SelectivePersistFlow(Flow):
  5.     @start()
  6.     def first_step(self):
  7.         self.state["count"] = 1
  8.         return "First step"
  10.     @persist()  # Only persist after this method
  11.     @listen(first_step)
  12.     def important_step(self, prev_result):
  13.         self.state["count"] += 1
  14.         self.state["important_data"] = "This will be persisted"
  15.         return "Important step completed"
  17.     @listen(important_step)
  18.     def final_step(self, prev_result):
  19.         self.state["count"] += 1
  20.         return f"Complete with count {self.state['count']}"

高级状态模式

条件启动与可恢复执行

Flows 支持条件 @start() 和可恢复执行，这对于 HITL / 循环场景非常有用：

  1. from crewai.flow.flow import Flow, start, listen, and_, or_
  3. class ResumableFlow(Flow):
  4.     @start()  # unconditional start
  5.     def init(self):
  6.         ...
  8.     # Conditional start: run after "init" or external trigger name
  9.     @start("init")
  10.     def maybe_begin(self):
  11.         ...
  13.     @listen(and_(init, maybe_begin))
  14.     def proceed(self):
  15.         ...
  • 条件 @start() 可以接受方法名、router 标签或可调用条件
  • 在恢复执行期间，监听器会从之前的检查点继续；循环 / router 分支也会遵循恢复标记

基于状态的条件逻辑

你可以利用状态在 flow 中实现复杂条件逻辑：

  1. from crewai.flow.flow import Flow, listen, router, start
  2. from pydantic import BaseModel
  4. class PaymentState(BaseModel):
  5.     amount: float = 0.0
  6.     is_approved: bool = False
  7.     retry_count: int = 0
  9. class PaymentFlow(Flow[PaymentState]):
  10.     @start()
  11.     def process_payment(self):
  12.         # Simulate payment processing
  13.         self.state.amount = 100.0
  14.         self.state.is_approved = self.state.amount < 1000
  15.         return "Payment processed"
  17.     @router(process_payment)
  18.     def check_approval(self, previous_result):
  19.         if self.state.is_approved:
  20.             return "approved"
  21.         elif self.state.retry_count < 3:
  22.             return "retry"
  23.         else:
  24.             return "rejected"
  26.     @listen("approved")
  27.     def handle_approval(self):
  28.         return f"Payment of ${self.state.amount} approved!"
  30.     @listen("retry")
  31.     def handle_retry(self):
  32.         self.state.retry_count += 1
  33.         print(f"Retrying payment (attempt {self.state.retry_count})...")
  34.         # Could implement retry logic here
  35.         return "Retry initiated"
  37.     @listen("rejected")
  38.     def handle_rejection(self):
  39.         return f"Payment of ${self.state.amount} rejected after {self.state.retry_count} retries."

处理复杂状态转换

对于复杂状态转换，你可以创建专门的方法来处理：

  1. from crewai.flow.flow import Flow, listen, start
  2. from pydantic import BaseModel
  3. from typing import List, Dict
  5. class UserData(BaseModel):
  6.     name: str
  7.     active: bool = True
  8.     login_count: int = 0
  10. class ComplexState(BaseModel):
  11.     users: Dict[str, UserData] = {}
  12.     active_user_count: int = 0
  14. class TransformationFlow(Flow[ComplexState]):
  15.     @start()
  16.     def initialize(self):
  17.         # Add some users
  18.         self.add_user("alice", "Alice")
  19.         self.add_user("bob", "Bob")
  20.         self.add_user("charlie", "Charlie")
  21.         return "Initialized"
  23.     @listen(initialize)
  24.     def process_users(self, _):
  25.         # Increment login counts
  26.         for user_id in self.state.users:
  27.             self.increment_login(user_id)
  29.         # Deactivate one user
  30.         self.deactivate_user("bob")
  32.         # Update active count
  33.         self.update_active_count()
  35.         return f"Processed {len(self.state.users)} users"
  37.     # Helper methods for state transformations
  38.     def add_user(self, user_id: str, name: str):
  39.         self.state.users[user_id] = UserData(name=name)
  40.         self.update_active_count()
  42.     def increment_login(self, user_id: str):
  43.         if user_id in self.state.users:
  44.             self.state.users[user_id].login_count += 1
  46.     def deactivate_user(self, user_id: str):
  47.         if user_id in self.state.users:
  48.             self.state.users[user_id].active = False
  49.             self.update_active_count()
  51.     def update_active_count(self):
  52.         self.state.active_user_count = sum(
  53.             1 for user in self.state.users.values() if user.active
  54.         )

通过创建辅助方法来处理状态转换，可以让 flow 方法本身保持简洁，同时支持复杂状态操作。

将状态管理与 Crews 结合使用

在 CrewAI 中，最强大的模式之一，就是将 flow 状态管理与 crew 执行结合起来。

将状态传递给 Crews

你可以使用 flow 状态来为 crews 提供参数：

  1. from crewai.flow.flow import Flow, listen, start
  2. from crewai import Agent, Crew, Process, Task
  3. from pydantic import BaseModel
  5. class ResearchState(BaseModel):
  6.     topic: str = ""
  7.     depth: str = "medium"
  8.     results: str = ""
  10. class ResearchFlow(Flow[ResearchState]):
  11.     @start()
  12.     def get_parameters(self):
  13.         # In a real app, this might come from user input
  14.         self.state.topic = "Artificial Intelligence Ethics"
  15.         self.state.depth = "deep"
  16.         return "Parameters set"
  18.     @listen(get_parameters)
  19.     def execute_research(self, _):
  20.         # Create agents
  21.         researcher = Agent(
  22.             role="Research Specialist",
  23.             goal=f"Research {self.state.topic} in {self.state.depth} detail",
  24.             backstory="You are an expert researcher with a talent for finding accurate information."
  25.         )
  27.         writer = Agent(
  28.             role="Content Writer",
  29.             goal="Transform research into clear, engaging content",
  30.             backstory="You excel at communicating complex ideas clearly and concisely."
  31.         )
  33.         # Create tasks
  34.         research_task = Task(
  35.             description=f"Research {self.state.topic} with {self.state.depth} analysis",
  36.             expected_output="Comprehensive research notes in markdown format",
  37.             agent=researcher
  38.         )
  40.         writing_task = Task(
  41.             description=f"Create a summary on {self.state.topic} based on the research",
  42.             expected_output="Well-written article in markdown format",
  43.             agent=writer,
  44.             context=[research_task]
  45.         )
  47.         # Create and run crew
  48.         research_crew = Crew(
  49.             agents=[researcher, writer],
  50.             tasks=[research_task, writing_task],
  51.             process=Process.sequential,
  52.             verbose=True
  53.         )
  55.         # Run crew and store result in state
  56.         result = research_crew.kickoff()
  57.         self.state.results = result.raw
  59.         return "Research completed"
  61.     @listen(execute_research)
  62.     def summarize_results(self, _):
  63.         # Access the stored results
  64.         result_length = len(self.state.results)
  65.         return f"Research on {self.state.topic} completed with {result_length} characters of results."

在状态中处理 Crew 输出

当一个 crew 执行完成后，你可以处理其输出，并将结果保存到 flow 状态中：

  1. @listen(execute_crew)
  2. def process_crew_results(self, _):
  3.     # Parse the raw results (assuming JSON output)
  4.     import json
  5.     try:
  6.         results_dict = json.loads(self.state.raw_results)
  7.         self.state.processed_results = {
  8.             "title": results_dict.get("title", ""),
  9.             "main_points": results_dict.get("main_points", []),
  10.             "conclusion": results_dict.get("conclusion", "")
  11.         }
  12.         return "Results processed successfully"
  13.     except json.JSONDecodeError:
  14.         self.state.error = "Failed to parse crew results as JSON"
  15.         return "Error processing results"

状态管理最佳实践

1. 让状态保持聚焦

设计状态时，只保留必要内容：

  1. # Too broad
  2. class BloatedState(BaseModel):
  3.     user_data: Dict = {}
  4.     system_settings: Dict = {}
  5.     temporary_calculations: List = []
  6.     debug_info: Dict = {}
  7.     # ...many more fields
  9. # Better: Focused state
  10. class FocusedState(BaseModel):
  11.     user_id: str
  12.     preferences: Dict[str, str]
  13.     completion_status: Dict[str, bool]

2. 在复杂 Flows 中使用结构化状态

随着 flow 复杂度增加，结构化状态的价值会越来越明显：

  1. # Simple flow can use unstructured state
  2. class SimpleGreetingFlow(Flow):
  3.     @start()
  4.     def greet(self):
  5.         self.state["name"] = "World"
  6.         return f"Hello, {self.state['name']}!"
  8. # Complex flow benefits from structured state
  9. class UserRegistrationState(BaseModel):
  10.     username: str
  11.     email: str
  12.     verification_status: bool = False
  13.     registration_date: datetime = Field(default_factory=datetime.now)
  14.     last_login: Optional[datetime] = None
  16. class RegistrationFlow(Flow[UserRegistrationState]):
  17.     # Methods with strongly-typed state access

3. 为状态转换写文档

对于复杂 flows，建议记录状态如何在执行过程中发生变化：

  1. @start()
  2. def initialize_order(self):
  3.     """
  4.     Initialize order state with empty values.
  6.     State before: {}
  7.     State after: {order_id: str, items: [], status: 'new'}
  8.     """
  9.     self.state.order_id = str(uuid.uuid4())
  10.     self.state.items = []
  11.     self.state.status = "new"
  12.     return "Order initialized"

4. 优雅地处理状态错误

为状态访问实现错误处理：

  1. @listen(previous_step)
  2. def process_data(self, _):
  3.     try:
  4.         # Try to access a value that might not exist
  5.         user_preference = self.state.preferences.get("theme", "default")
  6.     except (AttributeError, KeyError):
  7.         # Handle the error gracefully
  8.         self.state.errors = self.state.get("errors", [])
  9.         self.state.errors.append("Failed to access preferences")
  10.         user_preference = "default"
  12.     return f"Used preference: {user_preference}"

5. 使用状态追踪进度

在长时间运行的 flows 中，可以利用状态追踪执行进度：

  1. class ProgressTrackingFlow(Flow):
  2.     @start()
  3.     def initialize(self):
  4.         self.state["total_steps"] = 3
  5.         self.state["current_step"] = 0
  6.         self.state["progress"] = 0.0
  7.         self.update_progress()
  8.         return "Initialized"
  10.     def update_progress(self):
  11.         """Helper method to calculate and update progress"""
  12.         if self.state.get("total_steps", 0) > 0:
  13.             self.state["progress"] = (self.state.get("current_step", 0) /
  14.                                     self.state["total_steps"]) * 100
  15.             print(f"Progress: {self.state['progress']:.1f}%")
  17.     @listen(initialize)
  18.     def step_one(self, _):
  19.         # Do work...
  20.         self.state["current_step"] = 1
  21.         self.update_progress()
  22.         return "Step 1 complete"
  24.     # Additional steps...

6. 尽可能使用不可变操作

尤其是在结构化状态中，优先考虑不可变操作可以提高代码清晰度：

  1. # Instead of modifying lists in place:
  2. self.state.items.append(new_item)  # Mutable operation
  4. # Consider creating new state:
  5. from pydantic import BaseModel
  6. from typing import List
  8. class ItemState(BaseModel):
  9.     items: List[str] = []
  11. class ImmutableFlow(Flow[ItemState]):
  12.     @start()
  13.     def add_item(self):
  14.         # Create new list with the added item
  15.         self.state.items = [*self.state.items, "new item"]
  16.         return "Item added"

调试 Flow 状态

记录状态变化

在开发过程中，可以添加日志来追踪状态变化：

  1. import logging
  2. logging.basicConfig(level=logging.INFO)
  4. class LoggingFlow(Flow):
  5.     def log_state(self, step_name):
  6.         logging.info(f"State after {step_name}: {self.state}")
  8.     @start()
  9.     def initialize(self):
  10.         self.state["counter"] = 0
  11.         self.log_state("initialize")
  12.         return "Initialized"
  14.     @listen(initialize)
  15.     def increment(self, _):
  16.         self.state["counter"] += 1
  17.         self.log_state("increment")
  18.         return f"Incremented to {self.state['counter']}"

状态可视化

你还可以添加方法来可视化状态，以便调试：

  1. def visualize_state(self):
  2.     """Create a simple visualization of the current state"""
  3.     import json
  4.     from rich.console import Console
  5.     from rich.panel import Panel
  7.     console = Console()
  9.     if hasattr(self.state, "model_dump"):
  10.         # Pydantic v2
  11.         state_dict = self.state.model_dump()
  12.     elif hasattr(self.state, "dict"):
  13.         # Pydantic v1
  14.         state_dict = self.state.dict()
  15.     else:
  16.         # Unstructured state
  17.         state_dict = dict(self.state)
  19.     # Remove id for cleaner output
  20.     if "id" in state_dict:
  21.         state_dict.pop("id")
  23.     state_json = json.dumps(state_dict, indent=2, default=str)
  24.     console.print(Panel(state_json, title="Current Flow State"))

结论

掌握 CrewAI Flows 中的状态管理，将赋予你构建复杂、健壮 AI 应用的能力，让这些应用能够维护上下文、做出复杂决策，并输出一致可靠的结果。

无论你选择非结构化状态还是结构化状态，只要实施良好的状态管理实践，就能帮助你创建更易维护、更易扩展，并且能真正解决现实问题的 flows。

当你开发越来越复杂的 flows 时，请记住，良好的状态管理关键在于找到灵活性与结构性之间的平衡，让你的代码既强大又易于理解。

现在，你已经掌握了 CrewAI Flows 中状态管理的核心概念与实践方法！有了这些知识，你就可以构建健壮的 AI 工作流，有效维护上下文、在步骤之间共享数据，并建立复杂的应用逻辑。

下一步

  • 在你的 flows 中同时尝试结构化状态与非结构化状态
  • 尝试为长时间运行的工作流实现状态持久化
  • 探索 构建你的第一个 Crew，看看 crews 与 flows 如何协同工作
  • 查看 Flow 参考文档了解更多高级特性