🔄 Ralph Loop —— 自动化任务循环的艺术
"有些事情不需要问，有些事情不需要说。只要循环还在转动，答案终会出现。"
一、问题的提出：Agent 需要自动化
想象这样一个场景：
用户："帮我检查这个项目中所有的 Python 文件，给每个函数加上类型注解"
Agent："好的，让我先列出所有 Python 文件..."
│
▼ 执行 Glob
Agent："找到了 50 个文件。我先处理第一个..."
│
▼ 处理 file1.py
Agent："file1.py 完成。接下来处理 file2.py..."
│
▼ 处理 file2.py
Agent："file2.py 完成。接下来处理 file3.py..."
│
...（重复 47 次）
│
▼ 处理 file50.py
Agent："全部完成！"
这种交互式批量处理存在明显问题：
- 效率低下 —— 每次处理完一个文件都要等待用户确认
- 上下文开销 —— 每次迭代都要重新加载上下文
- 易出错 —— 长时间的重复操作容易遗漏或出错
- 用户体验差 —— 用户需要盯着屏幕看 50 轮对话
我们需要一种机制，让 Agent 能够自动循环执行任务，直到完成。
这就是 Ralph Loop 诞生的原因。
二、Ralph Loop 是什么？
Ralph Loop 是 Kimi Code CLI 中的自动化任务循环机制。它允许 Agent 在无需用户交互的情况下，自动重复执行同一任务，直到满足停止条件。
名字的由来
"Ralph" 这个名字据说来自项目早期的一个内部代号（也可能是某位开发者的宠物名）。无论起源如何，这个名字现在已经成为了自动化循环的代名词。
核心特性
| 特性 | 说明 |
|---|
| **自动迭代** | 无需用户输入，自动重复执行任务 |
| **智能判断** | LLM 自主决定何时停止循环 |
| **安全限制** | 可配置最大迭代次数，防止无限循环 |
| **状态保持** | 每次迭代都能看到之前的执行结果 |
三、Ralph Loop 的架构
1. 配置参数
Ralph Loop 通过配置控制：
class LoopControl(BaseModel):
max_ralph_iterations: int = Field(default=0, ge=-1)
"""
Ralph 模式的额外迭代次数。
0 = 禁用 Ralph Loop（默认）
-1 = 无限循环（直到 LLM 决定停止）
N = 最多 N 次额外迭代
"""
使用方式：
# 启用 Ralph Loop，最多 10 次迭代
kimi --max-ralph-iterations 10
# 启用无限循环（慎用！）
kimi --max-ralph-iterations -1
2. 入口判断
在 KimiSoul.run() 中判断是否进入 Ralph Loop：
async def run(self, user_input: str | list[ContentPart]):
# ... 处理 slash 命令 ...
elif self._loop_control.max_ralph_iterations != 0:
# 进入 Ralph Loop 模式！
runner = FlowRunner.ralph_loop(
user_message,
self._loop_control.max_ralph_iterations,
)
await runner.run(self, "")
else:
# 普通模式
await self._turn(user_message)
四、Ralph Loop 的实现：Flow 的力量
Ralph Loop 的巧妙之处在于：它不是独立的循环机制，而是基于 Flow（流程）系统构建的。
Flow 是什么？
Flow 是一个状态机执行引擎，用于执行预定义的工作流：
@dataclass
class Flow:
nodes: dict[str, FlowNode] # 节点（状态）
outgoing: dict[str, list[FlowEdge]] # 边（转移）
begin_id: str # 开始节点
end_id: str # 结束节点
节点类型：
begin —— 起点end —— 终点task —— 执行任务decision —— 决策点（多分支）
Ralph Loop 的 Flow 构造
这是 Ralph Loop 的核心代码：
@staticmethod
def ralph_loop(user_message: Message, max_ralph_iterations: int) -> FlowRunner:
prompt_content = list(user_message.content)
prompt_text = Message(role="user", content=prompt_content).extract_text(" ").strip()
# 计算总运行次数
total_runs = max_ralph_iterations + 1
if max_ralph_iterations < 0:
total_runs = 1000000000000000 # 实际上无限
# 构建 Flow 节点
nodes = {
"BEGIN": FlowNode(id="BEGIN", label="BEGIN", kind="begin"),
"END": FlowNode(id="END", label="END", kind="end"),
"R1": FlowNode(id="R1", label=prompt_content, kind="task"),
"R2": FlowNode(
id="R2",
label=(
f"{prompt_text}. (You are running in an automated loop..."
"Only choose STOP when the task is fully complete. "
"If you are not 100% sure, choose CONTINUE.)"
),
kind="decision",
),
}
# 构建转移边
outgoing = {
"BEGIN": [FlowEdge(src="BEGIN", dst="R1", label=None)],
"R1": [FlowEdge(src="R1", dst="R2", label=None)],
"R2": [
FlowEdge(src="R2", dst="R2", label="CONTINUE"), # 自循环！
FlowEdge(src="R2", dst="END", label="STOP"),
],
"END": [],
}
flow = Flow(nodes=nodes, outgoing=outgoing, begin_id="BEGIN", end_id="END")
return FlowRunner(flow, max_moves=total_runs)
Flow 图的可视化
┌─────────┐ ┌─────────┐ ┌─────────┐
│ BEGIN │────▶│ R1 │────▶│ R2 │
└─────────┘ │ (Task) │ │(Decision│
└─────────┘ └────┬────┘
│
┌───────────────┼───────────────┐
│ │ │
▼ │ ▼
┌─────────┐ │ ┌─────────┐
│ R2 │◀─────────┘ │ END │
│(Continue│ │ (Stop) │
└─────────┘ └─────────┘
R2 节点的自循环是 Ralph Loop 的核心！
五、执行机制：LLM 自主决策
一轮循环的执行
async def _execute_flow_node(self, soul: KimiSoul, node: FlowNode, edges: list[FlowEdge]):
# 构建提示词
base_prompt = self._build_flow_prompt(node, edges)
steps_used = 0
while True:
# 调用 LLM
result = await self._flow_turn(soul, prompt)
steps_used += result.step_count
# 决策节点：解析 LLM 的选择
if node.kind == "decision":
choice = parse_choice(result.final_message.extract_text(" "))
next_id = self._match_flow_edge(edges, choice)
if next_id is not None:
return next_id, steps_used
# 无效选择，重新提示
prompt = base_prompt + "\\n\\n请使用 <choice>...</choice> 格式回复。"
决策提示词
R2 节点的提示词设计非常精妙：
{原始用户输入}
(You are running in an automated loop where the same
prompt is fed repeatedly. Only choose STOP when the task
is fully complete. Including it will stop further iterations.
If you are not 100% sure, choose CONTINUE.)
关键点：
- 重复原始输入 —— 让 LLM 知道任务是什么
- 说明循环机制 —— 告知 LLM 它在自动循环中
- 明确停止条件 —— "Only choose STOP when the task is fully complete"
- 保守策略 —— "If you are not 100% sure, choose CONTINUE"
选择解析
LLM 需要通过 <choice>...</choice> 标签明确表达选择：
_CHOICE_RE = re.compile(r"<choice>([^<]*)</choice>")
def parse_choice(text: str) -> str | None:
matches = _CHOICE_RE.findall(text or "")
if not matches:
return None
return matches[-1].strip() # 取最后一个匹配
示例对话：
第 1 轮：
用户任务: "给所有函数加类型注解"
LLM: "我将开始处理..."
<choice>CONTINUE</choice>
第 2 轮：
用户任务: "给所有函数加类型注解" (自动重复)
LLM: "已处理 10/50 个文件..."
<choice>CONTINUE</choice>
...
第 50 轮：
用户任务: "给所有函数加类型注解" (自动重复)
LLM: "所有文件处理完成！"
<choice>STOP</choice>
六、Ralph Loop vs 普通模式
普通模式
# 用户输入 -> Agent 处理 -> 等待用户输入 -> ...
async def normal_mode():
while True:
user_input = await get_user_input()
if user_input == "/done":
break
result = await agent.process(user_input)
print(result)
特点：
Ralph Loop 模式
# 用户输入 -> Agent 自动循环 -> 自动停止
async def ralph_mode():
initial_input = await get_user_input()
runner = FlowRunner.ralph_loop(initial_input, max_iterations=10)
await runner.run(agent, "")
特点：
- 初始输入后自动执行
- LLM 自主控制停止
- 适合确定性批量任务
对比图
【普通模式】
用户: "处理第1个文件"
│
▼
Agent: "完成"
│
▼
用户: "处理第2个文件" ← 用户必须参与每一轮
│
▼
Agent: "完成"
│
...
【Ralph Loop】
用户: "处理所有文件" ← 只说一次
│
▼
Agent: "处理第1个..."
│ (自动循环)
Agent: "处理第2个..."
│ (自动循环)
Agent: "处理第3个..."
│
...直到完成
│
Agent: "全部完成！"
七、实际应用场景
场景 1：批量代码重构
$ kimi --max-ralph-iterations 20
> 将所有 print 语句替换为 logger 调用
【Ralph Loop 自动执行】
- 找到所有 print 语句
- 逐个文件替换
- 验证每个替换
- 完成后停止
场景 2：多步骤数据分析
$ kimi --max-ralph-iterations -1
> 分析 logs/ 目录下的所有日志文件，提取错误模式
【Ralph Loop 自动执行】
- 读取 log_001.txt
- 分析错误模式
- 读取 log_002.txt
- 累积分析结果
- ...直到所有文件处理完毕
- 生成汇总报告
场景 3：迭代式调试
$ kimi --max-ralph-iterations 10
> 运行测试，修复失败的测试，重复直到所有测试通过
【Ralph Loop 自动执行】
- 运行测试
- 分析失败原因
- 修复代码
- 重新运行测试
- 如果还有失败，继续
- 如果全部通过，停止
八、安全机制
最大迭代限制
total_runs = max_ralph_iterations + 1
if max_ralph_iterations < 0:
total_runs = 1000000000000000 # 实际上是无限，但有上限
# 在 FlowRunner 中检查
if moves >= self._max_moves:
raise MaxStepsReached(total_steps)
即使设置为 -1（无限），也有一个极大的上限（10^15），防止真正的无限循环。
步骤数限制
Ralph Loop 本身也受 max_steps_per_turn 限制：
if step_no > self._loop_control.max_steps_per_turn:
raise MaxStepsReached(self._loop_control.max_steps_per_turn)
用户中断
用户可以随时发送取消信号（如 Ctrl+C），Ralph Loop 会优雅地停止：
async def run_soul(..., cancel_event: asyncio.Event):
# 监听取消事件
cancel_event_task = asyncio.create_task(cancel_event.wait())
await asyncio.wait(
[soul_task, cancel_event_task],
return_when=asyncio.FIRST_COMPLETED,
)
九、设计哲学
1. 信任但验证
Ralph Loop 信任 LLM 能够正确判断何时停止，但提供了多重安全网：
2. 渐进式自动化
Ralph Loop 不是全有或全无：
0 = 完全手动（默认）10 = 适度自动化-1 = 完全信任 LLM
用户可以根据任务复杂度和信任程度选择合适的级别。
3. 基于 Flow 的通用性
Ralph Loop 基于 Flow 系统构建，这意味着：
- 代码复用 —— 不需要独立实现循环逻辑
- 扩展性 —— 可以利用 Flow 的所有特性
- 一致性 —— 与
/flow:* 命令使用相同的执行引擎
十、局限与改进
当前局限
- 单轮限制 —— Ralph Loop 只能在一次对话中循环，不能跨会话
- 无持久化 —— 如果中断，无法恢复进度
- 简单决策 —— 目前只有 CONTINUE/STOP 两种选择
可能的改进
# 更丰富的决策选项
class RalphDecision(Enum):
CONTINUE = "continue" # 继续
STOP = "stop" # 停止
PAUSE = "pause" # 暂停，等待用户输入
ADJUST = "adjust" # 调整策略后继续
ROLLBACK = "rollback" # 回退到上一轮
# 带进度持久化的 Ralph Loop
class PersistentRalphLoop:
async def run(self):
progress = await self.load_progress()
for i in range(progress.last_iteration, self.max_iterations):
await self.save_progress(i)
# ... 执行迭代
十一、结语
Ralph Loop 是 Kimi Code CLI 中实现高效自动化的关键机制。它让 Agent 能够：
- 自主执行 —— 无需用户干预的批量任务
- 智能判断 —— LLM 决定何时完成任务
- 安全可控 —— 多重限制防止失控
通过巧妙地将循环建模为 Flow 的自循环边，Ralph Loop 以最简洁的方式实现了强大的自动化能力。
正如它的名字所暗示的那样：Ralph Loop 不仅仅是一个技术机制，它是 Agent 从"被动工具"进化为"主动助手"的重要一步。
"在循环的尽头，答案已经等待多时。"
文章完成时间：2026-02-23
作者：爪爪