模板解析把一个字符串拆成交替的文本/表达式片段（flow/expressions.py:83）：

# flow/expressions.py:83
@lru_cache(maxsize=256)                     # 同一模板只解析一次
def _parse_template_segments(value: str) -> tuple[str | _ExpressionSegment, ...]:
    segments = []
    index = 0
    while (start := value.find("${", index)) != -1:   # 找下一个 ${
        if start > index:
            segments.append(value[index:start])        # ${ 之前的纯文本
        end = _marker_end(value, start + 2)             # 找配对的 }
        source = value[start + 2 : end].strip()         # 抠出表达式源码
        if not source:
            raise ExpressionError(f"empty CEL expression in {value!r}")
        segments.append(_ExpressionSegment(source))     # 包成表达式段
        index = end + 1
    if index < len(value) or not segments:
        segments.append(value[index:])                  # 收尾剩余文本
    return tuple(segments)

Expression 是对外的门面（flow/expressions.py:197）：

# flow/expressions.py:197
class Expression:
    def __init__(self, value, *, context=None) -> None:
        self.value = value; self.context = context

    @classmethod
    def from_flow(cls, value, flow, *, local_context=None) -> Expression:
        return cls(value, context=cls._flow_context(flow, local_context=local_context))  # 带上 Flow 上下文

    def evaluate(self, context=None) -> Any:            # 整个值当一个完整 CEL 表达式求值
        resolved = self.context if context is None else context
        return self._evaluate_cel(self._require_cel_source(self.value), resolved or {})

    def render_template(self, context=None) -> Any:     # 渲染 ${...} 模板
        resolved = self.context if context is None else context
        return self._render_template_value(self.value, resolved or {})

模板渲染最精妙的一条规则（flow/expressions.py:349）：

# flow/expressions.py:349
@staticmethod
def _render_template_string(value: str, context) -> Any:
    segments = _parse_template_segments(value)
    expressions = [s for s in segments if isinstance(s, _ExpressionSegment)]
    if not expressions:
        return value                          # 没有 ${...} → 原样返回文本
    literals = [s for s in segments if isinstance(s, str)]
    if len(expressions) == 1 and all(not lit.strip() for lit in literals):
        return Expression._evaluate_cel(expressions[0].source, context)  # ★整串就一个表达式 → 保留原类型
    rendered = []                             # 否则逐段渲染、拼成字符串
    for segment in segments:
        if isinstance(segment, str):
            rendered.append(segment); continue
        result = Expression._evaluate_cel(segment.source, context)
        rendered.append("" if result is None else _stringify_cel_value(result))
    return "".join(rendered)

真正求值不是 CrewAI 手写的，而是委托 celpy 库（flow/expressions.py:369）：

# flow/expressions.py:369
@staticmethod
def _evaluate_cel(expression: str, context: dict[str, Any]) -> Any:
    try:
        from celpy import Environment
        from celpy.adapter import CELJSONEncoder, json_to_cel
        from celpy.evaluation import Context

        environment = Environment()
        program = environment.program(
            Expression._compile_cel(expression, environment=environment),
            functions=_EXPRESSION_FUNCTIONS,          # 注入自定义函数 text()
        )
        result = program.evaluate(cast(Context, json_to_cel(context)))  # 在受限上下文里算
        return json.loads(json.dumps(result, cls=CELJSONEncoder))       # CEL 类型 → 普通 Python
    except Exception as e:
        raise ExpressionError(f"failed to evaluate CEL expression {expression!r}: {e}") from e

Flow 给表达式准备的上下文（flow/expressions.py:310）：

# flow/expressions.py:310
@staticmethod
def _flow_context(flow, local_context=None) -> dict[str, Any]:
    from crewai.flow.runtime._outputs import outputs_by_name
    local_outputs = local_context.get("outputs") if local_context else None
    outputs = outputs_by_name(flow._method_outputs, local_outputs=local_outputs, serialize=True)
    context = {
        "state": flow._copy_and_serialize_state(),   # ★整个 state（序列化后的副本）
        "outputs": outputs,                          # ★按方法名索引的历史输出
    }
    if local_context:                                # 可选：额外的局部变量
        context.update({k: to_serializable(v, max_depth=0)
                        for k, v in local_context.items() if k not in {"outputs", "state"}})
    return context

D44 的校验器调的就是 validate_expression，它检查表达式只用了允许的根（flow/expressions.py:217）：

# flow/expressions.py:217
def validate_expression(self, *, allowed_roots, source="CEL expression") -> None:
    allowed = frozenset(allowed_roots)                       # 通常 {"state","outputs"}
    expression = self._require_cel_source(self.value, source=source)
    roots = self._collect_root_identifiers(self._compile_cel(expression, source=source))
    unknown = sorted(r for r in roots if r not in allowed)   # 用了不认识的根？
    if unknown:
        raise ExpressionError(
            f"unknown CEL root at {source}: ...; allowed roots: {', '.join(sorted(allowed))}. "
            "Reference flow data through one of those roots, for example state.field ...")

还有一个自定义函数 text() 专门兜"值可能缺失/为 null"（flow/expressions.py:27）：

# flow/expressions.py:27
def _handle_text_custom_expression(root, path, default="") -> StringType:
    fallback = StringType("" if default is None else str(default))
    current = root
    for part in str(path).split("."):        # 沿 "a.b.c" 逐层下钻
        if current is None:
            return fallback                  # 中途断了 → 返回默认
        try:
            current = current[int(part)] if isinstance(current, list) else current[StringType(part)]
        except (KeyError, IndexError, TypeError, ValueError):
            return fallback                  # 取不到 → 返回默认，不抛异常
    return StringType(...) if current is not None else fallback