Kimi-cli Flow 系统设计与实现分析

概述

本文深入分析 Kimi Code CLI 中的 Flow 系统 —— 一个用于编排和执行业务流程的核心组件。它允许通过声明式流程图（D2/Mermaid）定义复杂的交互模式，实现 AI 代理的自动化工作流。

---

一、核心数据结构设计

1.1 节点类型 (FlowNodeKind)

FlowNodeKind = Literal["begin", "end", "task", "decision"]

类型	说明	出边数
`begin`	流程起点	1
`end`	流程终点	0
`task`	普通任务节点	1
`decision`	决策节点	≥2

1.2 核心类定义

@dataclass(frozen=True, slots=True)
class FlowNode:
    id: str                          # 节点唯一标识
    label: str | list[ContentPart]   # 节点标签（可作为 LLM prompt）
    kind: FlowNodeKind               # 节点类型

@dataclass(frozen=True, slots=True)
class FlowEdge:
    src: str        # 源节点 ID
    dst: str        # 目标节点 ID  
    label: str|None # 边标签（decision 分支标识）

@dataclass(slots=True)
class Flow:
    nodes: dict[str, FlowNode]          # 节点集合
    outgoing: dict[str, list[FlowEdge]] # 出边邻接表
    begin_id: str
    end_id: str

---

二、流程验证机制

Flow 系统通过 validate_flow() 函数确保流程图的完整性：

1. 单一起止点检查 —— 必须有且只有一个 begin 和一个 end 2. 连通性检查 —— 通过 BFS 确保从 begin 可达所有节点 3. 决策节点约束 —— 多条出边必须有标签，且标签不能重复 4. 终点可达性 —— end 必须可从 begin 到达

---

三、双解析器支持

3.1 D2 解析器

支持 D2 声明式图表语法：

BEGIN -> R1 -> R2
R1: "执行任务内容"
R2: "检查是否完成？"
R2 -> R2: "CONTINUE"
R2 -> END: "STOP"

特性：

支持 |md Markdown 块标签
支持多行引号字符串
自动推断 decision 节点（出边数>1）

3.2 Mermaid 解析器

支持 Mermaid flowchart 语法：

flowchart TD
    BEGIN --> R1[执行任务]
    R1 --> R2{检查完成？}
    R2 -->|CONTINUE| R2
    R2 -->|STOP| END

特性：

支持 [文本]、(文本)、{文本} 三种形状
支持 |标签| 和 --标签--> 边标签语法

---

四、流程执行引擎 (FlowRunner)

4.1 执行循环逻辑

async def run(self, soul: KimiSoul, args: str) -> None:
    current_id = self._flow.begin_id
    while True:
        node = self._flow.nodes[current_id]
        edges = self._flow.outgoing.get(current_id, [])
        
        if node.kind == "end":
            return  # 流程结束
            
        next_id, steps = await self._execute_flow_node(soul, node, edges)
        current_id = next_id

4.2 节点执行策略

节点类型	执行方式
`task`	执行 `label` 作为 prompt，完成后自动走唯一出边
`decision`	执行 `label` 作为 prompt，等待 LLM 返回 `标签` 选择分支

4.3 分支选择解析

_CHOICE_RE = re.compile(r"<choice>([^<]*)</choice>")

def parse_choice(text: str) -> str | None:
    matches = _CHOICE_RE.findall(text or "")
    return matches[-1].strip() if matches else None

---

五、Ralph Loop - 自动化循环模式

Ralph Loop 是一种特殊的自动化流程，用于无人值守任务：

流程: BEGIN → R1(task) → R2(decision) --CONTINUE--> R2
                                    \--STOP--> END

使用场景：

批量代码修复
多轮迭代优化
自动化测试

实现原理：R2 节点通过决策判断任务是否完成，未完成则循环回自身继续执行。

---

六、与 Skill 系统集成

Flow 通过 Skill 系统暴露给用户：

# skill.yaml
name: my-flow
type: flow
description: 我的自动化流程
flow:
  format: d2  # 或 mermaid
  content: |
    BEGIN -> task1 -> decision1
    task1: "执行任务1"
    decision1: "继续吗？"
    decision1 -> task1: "YES"
    decision1 -> END: "NO"

注册后可通过 /flow:my-flow 命令调用。

---

七、设计亮点总结

1. 声明式定义 —— 使用图表语法描述流程，直观易懂 2. 双解析器支持 —— D2 适合复杂布局，Mermaid 适合快速原型 3. 与 LLM 深度集成 —— decision 节点通过自然语言选择分支 4. 自我修正 —— 无效选择时自动提示重新选择 5. Ralph Loop —— 支持无人值守的自动化任务执行

---

源码位置

src/kimi_cli/skill/flow/__init__.py    # 核心数据结构
src/kimi_cli/skill/flow/d2.py          # D2 解析器
src/kimi_cli/skill/flow/mermaid.py     # Mermaid 解析器
src/kimi_cli/soul/kimisoul.py          # FlowRunner 执行引擎

---

*分析基于 kimi-cli 最新源码，使用 sg_sag.php RAG 记忆管理系统辅助检索*