plaita.core.flow

流程模型定义与解析入口。包含 Flow、parse()、parse_and_run()。

flow

plaita.core.flow — Flow model definition and entry-point helpers.

Contains the canonical Flow class, parse(), and parse_and_run().

Flow

Bases: BaseModel

流程定义

flow_id is the canonical identifier. The legacy id accessor still works but emits a DeprecationWarning.

id property

id: Optional[str]

Deprecated — use flow_id instead.

start_node property

start_node

流程入口节点。

2026-07 行为变更: 入度 0 "启发式推断" 已删除——历史上无显式 Start 时 本属性会扫描 nodes 数组找"未被任何 next/branch 引用的节点"作为入口, 但当存在多个孤儿节点时返回结果依赖数组顺序, 让可视化编排工具导出 顺序的细微差异就能改变流程行为。现在统一要求显式 Start 节点: 没有 Start 就抛 FlowStartMissingError, 让问题在解析期暴露。

input_property property

input_property

历史别名（第 3/4 个名字）：等价 input_type。勿在新代码使用—— 同一概念现有 inputType（JSON）/ input_type（Python 字段）两个规范名。

output_property property

output_property

历史别名：等价 output_type。见 :attr:input_property 说明。

parse_flow classmethod

parse_flow(data: Dict) -> Dict

Parse flow definition and handle field name compatibility.

源代码位于： plaita/core/flow.py

@model_validator(mode="before")
@classmethod
def parse_flow(cls, data: Dict) -> Dict:
    """Parse flow definition and handle field name compatibility."""
    if not data:
        raise ValueError("Flow content cannot be empty")

    normalized = data.copy()

    # Normalize id / flowId → flow_id (flow_id takes priority)
    if "flow_id" not in normalized or normalized.get("flow_id") is None:
        for legacy_key in ("flowId", "id"):
            if legacy_key in normalized and normalized[legacy_key] is not None:
                normalized["flow_id"] = normalized.pop(legacy_key)
                break
    # Remove legacy keys so Pydantic doesn't choke on unknown fields
    normalized.pop("flowId", None)
    normalized.pop("id", None)

    camel_mappings = {
        "inputType": "input_type",
        "outputType": "output_type",
        "globalContext": "global_context",
    }
    for old_key, new_key in camel_mappings.items():
        if old_key in normalized:
            value = normalized.pop(old_key)
            if value is not None:
                normalized[new_key] = value

    for key in ["input_type", "output_type"]:
        if key in normalized:
            normalized[key] = Property.from_json(normalized[key])

    # 节点解析已移除——历史上此处调 ``get_default_registry().parse_node``
    # 把每个 node dict 解析成对应 ``Node`` 子类。这让 Pydantic validator
    # 产生副作用并隐式耦合到模块级默认 registry 的当前状态 (import 期注册了
    # 坏节点会让全进程解析任何 flow 都坏)。现在节点保持为原始 dict，由
    # ``Flow.resolve_nodes`` 显式解析，``model_validate`` 默认自动调一次。
    return normalized

model_validate classmethod

model_validate(data, *, registry=None, **kwargs)

Parse + validate a Flow, then resolve node dicts into Node 子类。

默认用 get_default_registry() 解析节点，保持常规用法开箱即用；传 registry= 可注入自定义注册表，避免隐式依赖进程级单例。

源代码位于： plaita/core/flow.py

@classmethod
def model_validate(cls, data, *, registry=None, **kwargs):
    """Parse + validate a Flow, then resolve node dicts into ``Node`` 子类。

    默认用 ``get_default_registry()`` 解析节点，保持常规用法开箱即用；传
    ``registry=`` 可注入自定义注册表，避免隐式依赖进程级单例。
    """
    flow = super().model_validate(data, **kwargs)
    flow.resolve_nodes(registry)
    flow._warn_uncovered_env_refs()
    return flow

model_validate_json classmethod

model_validate_json(json_data, *, registry=None, **kwargs)

JSON 解析 + 校验 + 节点解析（见 model_validate 说明）。

源代码位于： plaita/core/flow.py

@classmethod
def model_validate_json(cls, json_data, *, registry=None, **kwargs):
    """JSON 解析 + 校验 + 节点解析（见 ``model_validate`` 说明）。"""
    flow = super().model_validate_json(json_data, **kwargs)
    flow.resolve_nodes(registry)
    flow._warn_uncovered_env_refs()
    return flow

resolve_nodes

resolve_nodes(registry=None) -> None

把 self.nodes 里 dict 形态的节点解析成对应 Node 子类实例。

幂等：已经是 Node 实例的节点原样保留，重复调用安全。这是节点解析 的显式入口——不再藏在 Pydantic validator 里，调用方可随时用自定义 registry 重新解析。model_validate / model_validate_json / from_string / from_file / parse 默认会用 get_default_registry() 自动调一次。

源代码位于： plaita/core/flow.py

def resolve_nodes(self, registry=None) -> None:
    """把 ``self.nodes`` 里 dict 形态的节点解析成对应 ``Node`` 子类实例。

    幂等：已经是 ``Node`` 实例的节点原样保留，重复调用安全。这是节点解析
    的**显式**入口——不再藏在 Pydantic validator 里，调用方可随时用自定义
    ``registry`` 重新解析。``model_validate`` / ``model_validate_json`` /
    ``from_string`` / ``from_file`` / ``parse`` 默认会用
    ``get_default_registry()`` 自动调一次。
    """
    from plaita.node import Node, get_default_registry

    if not self.nodes:
        return
    reg = registry if registry is not None else get_default_registry()
    resolved: List[Any] = []
    changed = False
    for n in self.nodes:
        if isinstance(n, Node):
            resolved.append(n)
        elif isinstance(n, dict):
            resolved.append(reg.parse_node(n))
            changed = True
        else:
            resolved.append(n)
    if changed:
        self.nodes = resolved
        # 节点集合变了, 失效索引指纹
        self._node_index_sig = ()
    # 构建期结构检查（2026-09 LLM 作者模拟 P0-2）：id 重复 last-wins、
    # next 悬空、if 缺分支目标、switch 缺 default——ir_validate 的这些检查
    # 历史上只在 builder/codeflow/sexpr 路径生效，README 主入口
    # Flow.from_string（JSON，可视化导出格式、LLM 最常输出的形态）完全不
    # 经过。降级为 warning 保持兼容，把"静默成功"变成可见。
    if any(isinstance(n, Node) for n in resolved):
        try:
            from plaita.dsl.ir_validate import validate_flow_ir

            # model_dump() 不含类级 node_type（它是 ClassVar，不是字段），
            # 直接 dump 会让 validator 找不到 type=start，对所有 flow 误报。
            # 这里显式补回 type，保证校验的是真实结构而非序列化伪影。
            validate_flow_ir(
                {"nodes": [
                    {**n.model_dump(), "type": type(n).node_type}
                    for n in resolved if isinstance(n, Node)
                ]},
                recursive=True,
            )
        except Exception as e:
            logger.warning(
                "flow %s structural check (non-fatal, flow still runs): %s",
                self.flow_id, _scrub_url_text(e),
            )

    # 环检测（R6 fuzz B3；plaita#105 修正判据）：next/branches 成环且无 flow
    # 级 timeout 时 Normal 模式永久挂起，解析期应有提示。判据是**回边**（当前
    # DFS 路径上的祖先），不是「被访问过」——后者把任何汇聚点（合法菱形 DAG、
    # 两分支共用同一 End）都误报成死循环。仅告警（显式 loop/while 有自己的
    # 迭代上限），且只在**没有 flow 级 timeout** 时告警：设了超时就不再是
    # 「will loop forever」，这正是文案给出的处置办法。
    # 可达性口径不变：只检测 start 可达的子图（不可达子图的真环仍不报，见
    # plaita#105 的「边界」——那属于行为扩张，由维护者定夺）。
    if any(isinstance(n, Node) for n in resolved):
        start = self.start_node
        id_to_node = {n.id: n for n in resolved if isinstance(n, Node) and n.id}
        adjacency: Dict[str, List[str]] = {}
        for node_id, node_obj in id_to_node.items():
            targets: List[str] = []
            if node_obj.next and isinstance(node_obj.next, str):
                targets.append(node_obj.next)
            for b in getattr(node_obj, "branches", None) or []:
                nxt = getattr(b, "next", None)
                if isinstance(nxt, str):
                    targets.append(nxt)
            adjacency[node_id] = [t for t in targets if t in id_to_node]
        cycle = (
            _resolve_cycle(adjacency, start.id)
            if start is not None and start.id in adjacency else []
        )
        if cycle and not self.timeout:
            logger.warning(
                "flow %s contains a cycle in next/branches: %s; without a "
                "flow-level timeout a Normal run will loop forever. If "
                "intentional, set flow timeout or use loop/while nodes with "
                "iteration limits.",
                self.flow_id, " -> ".join(cycle),
            )

    # 构建期校验：JSON 直载路径历史上从不执行 node.validate()，builder 路径
    # 却会（如 Switch 无 branches 直接抛错）——同一类配置错误一条路响一条路哑。
    # 这里以 warning 降级执行（存量 JSON 流程可能带有 builder 才会拦的配置），
    # 致命的运行期问题仍由调度层兜底（如分支未命中硬失败）。
    self.validate_nodes(resolved)

validate_nodes

validate_nodes(nodes=None) -> None

对已解析的节点跑构建期 node.validate()（warning 降级，不阻断）。

128：resolve_nodes() 是唯一执行该校验的地方，而**直接构造 Flow(...)

从不调用它（只有 model_validate / model_validate_json 会显式调一次）。 于是"传 Node 实例"这条主路径（builder / @flow 前端 / 大量既有测试） 坏配置连 warning 都没有**，比传 dict 那条路更哑——与 resolve_nodes 注释里声称要修的"一条路响一条路哑"恰好同构。

抽出本方法，使两条构造路径可以共用同一实现（见 _validate_node_instances）。

源代码位于： plaita/core/flow.py

def validate_nodes(self, nodes=None) -> None:
    """对已解析的节点跑构建期 ``node.validate()``（warning 降级，不阻断）。

    #128：`resolve_nodes()` 是唯一执行该校验的地方，而**直接构造 ``Flow(...)``
    从不调用它**（只有 ``model_validate`` / ``model_validate_json`` 会显式调一次）。
    于是"传 ``Node`` 实例"这条主路径（builder / @flow 前端 / 大量既有测试）
    坏配置**连 warning 都没有**，比传 dict 那条路更哑——与 ``resolve_nodes``
    注释里声称要修的"一条路响一条路哑"恰好同构。

    抽出本方法，使两条构造路径可以共用同一实现（见 ``_validate_node_instances``）。
    """
    for node in (self.nodes if nodes is None else nodes) or []:
        if isinstance(node, Node):
            try:
                node.validate()
            except Exception as e:
                logger.warning(
                    "flow %s node %s (%s) failed construction-time validation "
                    "(non-fatal, the flow still runs): %s",
                    self.flow_id, getattr(node, "id", "?"), getattr(node, "node_type", "?"), _scrub_url_text(e),
                )

from_string classmethod

from_string(content: str, *, registry: Optional['NodeRegistry'] = None) -> 'Flow'

从 JSON 或 YAML 字符串解析 Flow。

JSON 走 model_validate_json；YAML（及无法按 JSON 解析的内容） 走 plaita.io_format.loads 再 model_validate。 JSON 解析失败时把原始异常作为 cause 保留，避免真正的报错被 YAML fallback 的次级报错淹没。

registry= 可注入自定义 NodeRegistry 解析节点（与 model_validate/resolve_nodes 对齐）；不传时用 get_default_registry()。

源代码位于： plaita/core/flow.py

@classmethod
def from_string(cls, content: str, *, registry: Optional["NodeRegistry"] = None) -> "Flow":
    """从 JSON 或 YAML 字符串解析 Flow。

    JSON 走 ``model_validate_json``；YAML（及无法按 JSON 解析的内容）
    走 ``plaita.io_format.loads`` 再 ``model_validate``。
    JSON 解析失败时把原始异常作为 cause 保留，避免真正的报错被
    YAML fallback 的次级报错淹没。

    ``registry=`` 可注入自定义 ``NodeRegistry`` 解析节点（与
    ``model_validate``/``resolve_nodes`` 对齐）；不传时用
    ``get_default_registry()``。
    """
    from plaita.io_format import loads

    if not content or not content.strip():
        raise ValueError("Flow content cannot be empty")
    # 优先 JSON，保持历史行为与错误信息
    if content.lstrip()[:1] in ("{", "["):
        try:
            flow = cls.model_validate_json(content, registry=registry)
        except ValueError as json_err:
            # 不要静默吞掉 JSON 报错——若 YAML fallback 也失败，
            # 把原始 JSON 异常作为 cause 一并抛出，方便定位真凶。
            try:
                data = loads(content)
            except Exception:
                raise json_err
            # YAML fallback 成功本身要大声说：JSON 语法错（尾逗号等）被
            # YAML 静默吸收，且 YAML 会把裸 no/on 翻译成 False/True 篡改值
            # （2026-09 LLM 作者模拟 D8）。
            logging.getLogger("plaita.core.flow").warning(
                "content failed JSON parsing (%s) but parsed as YAML — "
                "double-check the syntax; YAML also coerces bare no/on to "
                "False/True", _scrub_url_text(json_err),
            )
            flow = cls.model_validate(data, registry=registry)
        return flow
    data = loads(content)
    return cls.model_validate(data, registry=registry)

from_file classmethod

from_file(path: str, *, registry: Optional['NodeRegistry'] = None) -> 'Flow'

从文件加载 Flow，按后缀（.json/.yaml/.yml）选择解析器。

registry= 语义同 :meth:from_string。

源代码位于： plaita/core/flow.py

@classmethod
def from_file(cls, path: str, *, registry: Optional["NodeRegistry"] = None) -> "Flow":
    """从文件加载 Flow，按后缀（.json/.yaml/.yml）选择解析器。

    ``registry=`` 语义同 :meth:`from_string`。
    """
    from plaita.io_format import load_file

    data = load_file(path)
    flow = cls.model_validate(data, registry=registry)
    return flow

rebuild_node_index

rebuild_node_index() -> None

显式重建节点索引。直接修改 flow.nodes 内节点 id 后调用一次即可, 不必新建 Flow。

源代码位于： plaita/core/flow.py

def rebuild_node_index(self) -> None:
    """显式重建节点索引。直接修改 ``flow.nodes`` 内节点 id 后调用一次即可,
    不必新建 Flow。"""
    self._node_index_sig = ()
    self._ensure_index()

is_end_node

is_end_node(node) -> bool

True if node is an End node. Lives on Flow so callers (strategies, helpers) need not import plaita.node.End and re-trigger the core → node circular import.

源代码位于： plaita/core/flow.py

def is_end_node(self, node) -> bool:
    """True if *node* is an End node. Lives on Flow so callers (strategies,
    helpers) need not import ``plaita.node.End`` and re-trigger the
    core → node circular import."""
    return node is not None and node.node_type == End.node_type

run

run(*args, **params)

同步执行。输入走 run({...})（单个 dict/scalar）或 run(key=value, ...)。

契约：每个关键字都成为 $INPUT 的一个字段（README 的 flow.run(name="kongjie")），包括在兄弟入口里是执行控制的 mode/params/timeout 等名字——它们在这里不被拒绝。

只有 :data:_RESERVED_RUN_KEYWORDS（历史输入包装别名 inputs_data/input_data，以及 run_compatible 的形参 lazy）按名拒绝：它们到不了调用方以为的位置，静默接受只会 变成"看起来正常"的错误结果（plaita#137）。要传同名输入字段， 用 dict 形式：flow.run({"inputs_data": ...})。

源代码位于： plaita/core/flow.py

def run(self, *args, **params):
    """同步执行。输入走 ``run({...})``（单个 dict/scalar）或 ``run(key=value, ...)``。

    契约：**每个关键字都成为 ``$INPUT`` 的一个字段**（README 的
    ``flow.run(name="kongjie")``），包括在兄弟入口里是执行控制的
    ``mode``/``params``/``timeout`` 等名字——它们在这里**不**被拒绝。

    只有 :data:`_RESERVED_RUN_KEYWORDS`（历史输入包装别名
    ``inputs_data``/``input_data``，以及 ``run_compatible`` 的形参
    ``lazy``）按名拒绝：它们到不了调用方以为的位置，静默接受只会
    变成"看起来正常"的错误结果（plaita#137）。要传同名**输入字段**，
    用 dict 形式：``flow.run({"inputs_data": ...})``。
    """
    _enforce_dict_input(args)
    _reject_reserved_run_keywords(params)
    return FlowExecution().run_compatible(self, False, *args, **params)

arun async

arun(*args, **params)

异步 :meth:run；输入与保留关键字契约同 :meth:run。

源代码位于： plaita/core/flow.py

async def arun(self, *args, **params):
    """异步 :meth:`run`；输入与保留关键字契约同 :meth:`run`。"""
    _enforce_dict_input(args)
    _reject_reserved_run_keywords(params)
    return await FlowExecution().arun_compatible(self, False, *args, **params)

debug

debug(*args, **params)

generator 模式的 :meth:run；输入与保留关键字契约同 :meth:run。

源代码位于： plaita/core/flow.py

def debug(self, *args, **params):
    """generator 模式的 :meth:`run`；输入与保留关键字契约同 :meth:`run`。"""
    _enforce_dict_input(args)
    _reject_reserved_run_keywords(params)
    return FlowExecution().run_compatible(self, True, *args, **params)

parse

parse(content: Union[str, dict]) -> Optional[Flow]

Parse flow definition from JSON/YAML string or dict.

字符串内容自动识别 JSON 或 YAML（YAML 需安装 plaita[yaml]）； dict 直接走模型校验。所有解析最终都经过 Flow.parse_flow 这个 model_validator，行为与历史完全一致。

与 Flow.from_string 的空输入约定不同：本函数对空内容返回 None （历史契约，签名即 Optional[Flow]），而 from_string("") 抛 ValueError。

源代码位于： plaita/core/flow.py

def parse(content: Union[str, dict]) -> Optional[Flow]:
    """Parse flow definition from JSON/YAML string or dict.

    字符串内容自动识别 JSON 或 YAML（YAML 需安装 ``plaita[yaml]``）；
    dict 直接走模型校验。所有解析最终都经过 ``Flow.parse_flow`` 这个
    model_validator，行为与历史完全一致。

    与 ``Flow.from_string`` 的空输入约定不同：本函数对空内容返回 ``None``
    （历史契约，签名即 ``Optional[Flow]``），而 ``from_string("")`` 抛
    ``ValueError``。
    """
    if not content:
        return None
    if isinstance(content, str):
        from plaita.io_format import loads

        data = loads(content)
    else:
        data = content
    runtime = data.get("runtime", "python")
    if runtime != "python":
        raise ValueError(
            f"Unsupported runtime: {runtime!r}. plaita only executes 'python' flows; "
            "use a matching engine for other runtimes."
        )
    return Flow.model_validate(data)

parse_and_run

parse_and_run(content: str, *args, **kwargs)

Parse and execute a flow definition.

支持 JSON 与 YAML 两种文本格式。

源代码位于： plaita/core/flow.py

def parse_and_run(content: str, *args, **kwargs):
    """Parse and execute a flow definition.

    支持 JSON 与 YAML 两种文本格式。
    """
    from plaita.io_format import loads

    _enforce_dict_input(args)
    _reject_reserved_run_keywords(kwargs)
    data = loads(content)
    flow = Flow.model_validate(data)
    return FlowExecution().run_compatible(flow, False, *args, **kwargs)