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

## 概述 本文深入分析 **Kimi Code CLI** 中的 **Flow 系统** —— 一个用于编排和执行业务流程的核心组件。它允许通过声明式流程图（D2/Mermaid）定义复杂的交互模式，实现 AI 代理的自动化工作流。 --- ## 一、核心数据结构设计 ### 1.1 节点类型 (FlowNodeKind) ```python FlowNodeKind = Literal["begin", "end", "task", "decision"] ``` | 类型 | 说明 | 出边数 | |------|------|--------| | `begin` | 流程起点 | 1 | | `end` | 流程终点 | 0 | | `task` | 普通任务节点 | 1 | | `decision` | 决策节点 | ≥2 | ### 1.2 核心类定义 ```python @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 声明式图表语法： ```d2 BEGIN -> R1 -> R2 R1: "执行任务内容" R2: "检查是否完成？" R2 -> R2: "CONTINUE" R2 -> END: "STOP" ``` **特性**： - 支持 `|md` Markdown 块标签 - 支持多行引号字符串 - 自动推断 decision 节点（出边数>1） ### 3.2 Mermaid 解析器 支持 Mermaid flowchart 语法： ```mermaid flowchart TD BEGIN --> R1[执行任务] R1 --> R2{检查完成？} R2 -->|CONTINUE| R2 R2 -->|STOP| END ``` **特性**： - 支持 `[文本]`、`(文本)`、`{文本}` 三种形状 - 支持 `|标签|` 和 `--标签-->` 边标签语法 --- ## 四、流程执行引擎 (FlowRunner) ### 4.1 执行循环逻辑 ```python 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 返回 `<choice>标签</choice>` 选择分支 | ### 4.3 分支选择解析 ```python _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 系统暴露给用户： ```yaml # 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 记忆管理系统辅助检索*