《12-Factor Agents》8. 掌控你的控制流

如果你能掌控自己的控制流，你就可以实现许多有趣的功能。

构建适合你特定用例的自定义控制结构。具体来说，某些类型的工具调用可能是跳出循环、等待人类响应或等待另一个长时间运行任务 (例如训练管道) 的理由。你可能还想整合以下功能的自定义实现：

• 工具调用结果的摘要或缓存
• 针对结构化输出的 LLM-as-judge (大语言模型即评判者)
• 上下文窗口压缩或其他 内存管理
• 日志记录、追踪和指标
• 客户端速率限制
• 持久化休眠 / 暂停 / “等待事件”

下面的例子展示了三种可能的控制流模式：

• request_clarification: 模型请求更多信息，中断循环并等待人类的响应
• fetch_git_tags: 模型请求 git 标签列表，获取标签，将其附加到上下文窗口，然后直接传回给模型
• deploy_backend: 模型请求部署后端，这是一件高风险的事情，因此中断循环并等待人类批准

def handle_next_step(thread: Thread):
while True:
next_step = await determine_next_step(thread_to_prompt(thread))
# 为清晰起见，此处为内联代码 - 实际上你可以把它
# 放在一个方法里，使用异常来控制流程，或任何你喜欢的方式
if next_step.intent == 'request_clarification':
thread.events.append({
type: 'request_clarification',
data: nextStep,
})
await send_message_to_human(next_step)
await db.save_thread(thread)
# 异步步骤 - 中断循环，我们稍后会收到一个 webhook
break
elif next_step.intent == 'fetch_open_issues':
thread.events.append({
type: 'fetch_open_issues',
data: next_step,
})
issues = await linear_client.issues()
thread.events.append({
type: 'fetch_open_issues_result',
data: issues,
})
# 同步步骤 - 将新的上下文传递给 LLM 以确定下一步的行动
continue
elif next_step.intent == 'create_issue':
thread.events.append({
type: 'create_issue',
data: next_step,
})
await request_human_approval(next_step)
await db.save_thread(thread)
# 异步步骤 - 中断循环，我们稍后会收到一个 webhook
break

这种模式允许你根据需要中断和恢复智能体的流程，从而创建更自然的对话和工作流。

示例 – 对于市面上的每一个 AI 框架，我最希望看到的功能就是能够中断一个正在工作的智能体并在稍后恢复它，尤其是在工具 选择 和工具 调用 这两个时刻之间。

如果没有这种级别的可恢复性/粒度，就无法在工具调用运行前对其进行审查/批准，这意味着你不得不选择以下几种方式之一：

1. 在等待长时间运行的任务完成时，将任务暂停在内存中 (就像 while...sleep)，如果进程被中断，则必须从头重新启动
2. 将智能体的权限限制在仅能进行研究和摘要等低风险、低重要性的调用
3. 赋予智能体执行更重大、更有用任务的权限，然后凭运气 (yolo) 指望它别出差错

你可能会注意到，这与 第 5 要素 – 统一执行状态与业务状态 和 第 6 要素 – 通过简单的 API 启动/暂停/恢复 密切相关，但可以独立实现。