先看单个方法的定义（flow/flow_definition.py:643）：

# flow/flow_definition.py:643
class FlowMethodDefinition(BaseModel):
    """Static definition of one Flow method and its execution roles."""
    description: str | None = None                    # 人读的说明
    do: FlowActionDefinition = Field(...)             # 这方法执行什么动作（code/crew/agent/...）
    start: bool | FlowDefinitionCondition | None = None  # 是否起点 / 起点条件
    listen: FlowDefinitionCondition | None = None     # 监听条件（等谁）
    router: bool = False                              # 是否路由方法（输出当事件名）
    emit: list[str] | None = None                     # 声明可能发出的路由事件
    human_feedback: FlowHumanFeedbackDefinition | None = None  # 人在环步骤
    persist: FlowPersistenceDefinition | None = None  # 方法级持久化覆盖

它还有一个巧妙的规范化校验器（flow/flow_definition.py:689）：

# flow/flow_definition.py:689
@model_validator(mode="after")
def _canonicalize_human_feedback_routing(self) -> FlowMethodDefinition:
    # 一个 human_feedback 声明了 emit 结果的方法，就按 router 对待
    if self.human_feedback is not None and self.human_feedback.emit:
        self.router = True
        self.emit = None
    return self

把所有方法定义装进来的顶层契约（flow/flow_definition.py:710）：

# flow/flow_definition.py:710
class FlowDefinition(BaseModel):
    schema_: Literal["crewai.flow/v1"] = Field(default="crewai.flow/v1", alias="schema")
    name: str = Field(...)                            # 流程名，日志/事件/trace 里用
    description: str | None = None
    state: FlowStateDefinition | None = None          # 状态契约（dict / pydantic / json_schema）
    config: FlowConfigDefinition = Field(default_factory=FlowConfigDefinition)
    persist: FlowPersistenceDefinition | None = None  # 流程级持久化
    conversational: FlowConversationalDefinition | None = None  # 对话式配置（D47）
    methods: dict[str, FlowMethodDefinition] = Field(default_factory=dict)  # ★方法名 → 定义

契约是 Pydantic 模型，构建后自动跑三个校验器（flow/flow_definition.py:767）：

# flow/flow_definition.py:767
@model_validator(mode="after")
def _validate_method_names(self) -> FlowDefinition:
    for method_name in self.methods:
        _validate_step_name(method_name, field="Flow method names")   # ① 名字合法
    return self

@model_validator(mode="after")
def _validate_trigger_namespace(self) -> FlowDefinition:
    for method_name, method in self.methods.items():
        if _condition_references(method.listen, method_name):         # ② 不能监听自己
            raise ValueError(f"methods.{method_name}.listen must not reference itself")
    return self

@model_validator(mode="after")
def _validate_cel_expressions(self) -> FlowDefinition:
    for method_name, method in self.methods.items():
        _validate_action_cel(method.do, path=f"methods.{method_name}.do",   # ③ 表达式合法
                             allowed_roots=_BASE_CEL_ROOTS)
    return self

装饰器把信息挂在方法上，_build_flow_definition_from_class 负责把它们收集成完整契约（flow/dsl/_utils.py:435）：

# flow/dsl/_utils.py:435
def _build_flow_definition_from_class(flow_class, namespace=None) -> FlowDefinition:
    methods: dict[str, FlowMethodDefinition] = {}
    flow_methods = _iter_flow_methods(flow_class)          # ① 扫出所有被装饰的方法
    ...
    for method_name, method in flow_methods.items():
        methods[method_name] = _build_method_definition(   # ② 每个方法 → 一份 MethodDef
            method, f"methods.{method_name}")

    definition = FlowDefinition(                           # ③ 组装顶层契约
        name=getattr(flow_class, "__name__", "Flow"),
        description=(flow_class.__doc__ or "").strip() or None,
        state=_build_state_definition(flow_class),         # 从 Flow[T] 的 T 推状态契约
        config=_build_config_definition(flow_class),
        persist=_build_persistence_definition(flow_class),
        conversational=_build_conversational_definition(flow_class),
        methods=methods,
    )
    log_flow_definition_issues(definition)                 # ④ 记录（非致命的）诊断
    return definition

而 _build_method_definition（flow/dsl/_utils.py:361）把装饰器挂的碎片和 do 动作合成一份：

# flow/dsl/_utils.py:361
def _build_method_definition(method, path) -> FlowMethodDefinition:
    fragment = _get_flow_method_definition(method)         # 取装饰器挂的元数据
    if fragment is None:
        method_definition = FlowMethodDefinition(do=_method_action(method))
    else:
        method_definition = fragment.model_copy(deep=True, update={"do": _method_action(method)})
    human_feedback = _build_human_feedback_definition(method, ...)  # 合入人在环
    if human_feedback is not None:
        method_definition.human_feedback = human_feedback
    method_definition.persist = _build_persistence_definition(method)  # 合入持久化
    return method_definition

流程级执行配置（flow/flow_definition.py:191）：

# flow/flow_definition.py:191
class FlowConfigDefinition(BaseModel):
    tracing: bool | None = None
    stream: bool = False                    # 是否流式发事件
    memory: dict[str, Any] | None = None
    input_provider: str | None = None
    suppress_flow_events: bool = False      # 关掉事件（内部 Flow 用）
    max_method_calls: int = 100             # ★一次 kickoff 最多执行多少次方法（防跑飞）
    defer_trace_finalization: bool = False
    checkpoint: bool | dict[str, Any] | None = None

另有一个全局单例 flow_config（flow/flow_config.py:17），是完全不同的东西：

# flow/flow_config.py:17
class FlowConfig:
    """Global configuration for Flow execution."""
    def __init__(self) -> None:
        self._hitl_provider: HumanFeedbackProvider | None = None   # 全局人在环 provider
        self._input_provider: InputProvider | None = None          # 全局 ask() 输入源
    ...
flow_config = FlowConfig()   # 进程级单例，部署时在启动处覆盖

契约既然是纯数据，就能反过来"从数据造 Flow"（flow/flow_definition.py:809）：

# flow/flow_definition.py:809
@classmethod
def from_declaration(cls, *, contents=None, path=None) -> FlowDefinition:
    if isinstance(contents, cls):
        return contents
    source_path = None
    if contents is None:
        if path is None:
            raise ValueError("Provide contents or path")
        source_path = Path(path)
        contents = source_path.expanduser().read_text(encoding="utf-8")  # 读 YAML 文件
    if isinstance(contents, dict):
        return cls._load_mapping(contents)
    ...
    loaded = yaml.safe_load(contents)                # 解析 YAML → dict
    if not isinstance(loaded, dict):
        raise ValueError("Flow declaration must contain a mapping")
    return cls._load_mapping(loaded, source_path=source_path)  # dict → 校验 → FlowDefinition

运行时的 Flow.from_declaration 再把契约变成可跑的实例（runtime/__init__.py:478）：

# runtime/__init__.py:478
@classmethod
def from_declaration(cls, *, contents=None, path=None, **kwargs) -> Flow[Any]:
    definition = FlowDefinition.from_declaration(contents=contents, path=path)
    return cls.model_validate(
        {**definition.config.model_dump(), **kwargs},
        context={"flow_definition": definition})    # 把契约注入实例

schema: crewai.flow/v1
name: ResearchFlow
state:
  type: dict
  default: {topic: "AI agents"}
methods:
  seed:
    start: true
    do: {call: expression, expr: state.topic}
  greet:
    listen: seed
    do: {call: expression, expr: "'研究: ' + state.topic"}