uncaged/workflow
Archived
10
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity

RFC: ThreadReactor — generic ReAct loop for thread-scoped LLM interactions #139

New Issue
Closed
opened 2026-05-09 01:42:22 +00:00 by xiaoju · 3 comments
xiaoju commented 2026-05-09 01:42:22 +00:00
Owner
Copy Link
Copy Source

背景

当前 reactExtract 是一个专用的 ReAct 循环，硬编码了 extract 场景（cas_get + schema extract tool）。但这个模式是通用的：

给定 LLM、tools、system prompt → 多轮 ReAct loop → 结构化输出

它可以复用到：

  • extract: 从 agent 输出提取结构化 meta
  • supervisor: 检查 thread 健康度，决定 continue/stop
  • agent: 作为 role 的 agent 实现（LLM + shell/memory 等 tools）

核心洞察：ThreadReactor 和 Role 的抽象一致——都是给定上下文和目标，通过工具循环产出结构化结果。

设计

两阶段 API

// ── LLM 能力的纯函数抽象 ──

type LlmFn = (
  messages: ChatMessage[],
  tools: ToolDefinition[],
) => Promise<Result<AssistantMessage, string>>;

// engine 侧提供工厂
function createLlmFn(provider: LlmProvider): LlmFn;

// ── 阶段 1：固定能力（创建时确定）──

type ThreadReactorConfig = {
  llm: LlmFn;
  systemPrompt: string;
  tools: ToolDefinition[];
  toolHandler: (call: ToolCall, thread: ThreadContext) => Promise<string>;
  maxRounds: number;
};

function createThreadReactor(config: ThreadReactorConfig): ThreadReactorFn;

// ── 阶段 2：每次调用（变化的部分）──

type ThreadReactorFn = <T extends Record<string, unknown>>(
  thread: ThreadContext,
  input: string,
  schema: z.ZodType<T>,
) => Promise<Result<T, string>>;

设计决策

决策 结论 理由
systemPrompt 阶段 1 固定 最大化 LLM prefix cache 命中
tools 阶段 1 固定 与 systemPrompt 搭配，定义 reactor 能力边界
schema 阶段 2 每次传 同一个 reactor 可能对不同 role 提取不同类型
ThreadContext 阶段 2 每次传 每次调用可能是不同 thread 的不同时刻
input string 场景需求通过 input 体现，不污染 system prompt
LLM LlmFn 抽象 reactor 不碰 fetch/baseUrl/apiKey，纯逻辑
Streaming 不做 当前场景不需要，agent 场景也不做 streaming
副作用 Reactor 有副作用 Reactor 会写入 thread（记录 ReAct loop 的对话到 .info.jsonl），不是纯函数
错误恢复 所有 tool 失败都 recoverable unknown tool、执行失败统一返回 error tool result 给 LLM 重试，不 hard error。只有 maxRounds 耗尽才终止
命名 ThreadReactor 明确是 workflow thread-scoped 的 reactor

错误恢复策略

// toolHandler 返回 string（成功的 tool result）
// 如果 tool 执行失败，toolHandler 也返回 string（错误描述）
// reactor 统一把结果作为 tool result 发回 LLM
//
// unknown tool → reactor 自动返回 "unknown tool: {name}" 给 LLM
// tool 执行异常 → toolHandler catch 后返回错误描述
// LLM 拿到错误信息后自行决定重试或换策略
// 只有 maxRounds 耗尽才返回 err

使用示例

// Extract 场景
const extractReactor = createThreadReactor({
  llm: createLlmFn(resolveModel(config, "extract")),
  systemPrompt: "You extract structured metadata from agent output...",
  tools: [casGetTool, extractTool],
  toolHandler: (call, thread) => {
    if (call.function.name === "cas_get") return handleCasGet(call, thread);
    // ...
  },
  maxRounds: 10,
});
const meta = await extractReactor(thread, agentContent, PlannerMetaSchema);

// Supervisor 场景
const supervisor = createThreadReactor({
  llm: createLlmFn(resolveModel(config, "supervisor")),
  systemPrompt: "You monitor thread health...",
  tools: [threadInspectTool, casGetTool],
  toolHandler: handleSupervisorTools,
  maxRounds: 5,
});
const decision = await supervisor(thread, threadSummary, SupervisorDecisionSchema);

Phase 拆分

Phase 1: reactor 核心 + extract 迁移

  • 新建 src/reactor/ 模块
  • 实现 createThreadReactor、createLlmFn、类型定义
  • 从 reactExtract 提取通用 ReAct loop 逻辑
  • 用 createThreadReactor 重写 createExtract
  • 删除 reactExtract 旧实现
  • 所有 extract 测试通过
  • Testing issue: 待创建

Phase 2: 迁移 supervisor

  • 用 createThreadReactor 重写 supervisor
  • Testing issue: 待创建

完成标准

  • 所有 Phase 的 testing issue 已 close
  • 全量测试通过
  • reactExtract 旧实现已删除
## 背景 当前 `reactExtract` 是一个专用的 ReAct 循环，硬编码了 extract 场景（cas_get + schema extract tool）。但这个模式是通用的： > 给定 LLM、tools、system prompt → 多轮 ReAct loop → 结构化输出 它可以复用到： - **extract**: 从 agent 输出提取结构化 meta - **supervisor**: 检查 thread 健康度，决定 continue/stop - **agent**: 作为 role 的 agent 实现（LLM + shell/memory 等 tools） 核心洞察：ThreadReactor 和 Role 的抽象一致——都是给定上下文和目标，通过工具循环产出结构化结果。 ## 设计 ### 两阶段 API ```typescript // ── LLM 能力的纯函数抽象 ── type LlmFn = ( messages: ChatMessage[], tools: ToolDefinition[], ) => Promise<Result<AssistantMessage, string>>; // engine 侧提供工厂 function createLlmFn(provider: LlmProvider): LlmFn; // ── 阶段 1：固定能力（创建时确定）── type ThreadReactorConfig = { llm: LlmFn; systemPrompt: string; tools: ToolDefinition[]; toolHandler: (call: ToolCall, thread: ThreadContext) => Promise<string>; maxRounds: number; }; function createThreadReactor(config: ThreadReactorConfig): ThreadReactorFn; // ── 阶段 2：每次调用（变化的部分）── type ThreadReactorFn = <T extends Record<string, unknown>>( thread: ThreadContext, input: string, schema: z.ZodType<T>, ) => Promise<Result<T, string>>; ``` ### 设计决策 | 决策 | 结论 | 理由 | |------|------|------| | systemPrompt | 阶段 1 固定 | 最大化 LLM prefix cache 命中 | | tools | 阶段 1 固定 | 与 systemPrompt 搭配，定义 reactor 能力边界 | | schema | 阶段 2 每次传 | 同一个 reactor 可能对不同 role 提取不同类型 | | ThreadContext | 阶段 2 每次传 | 每次调用可能是不同 thread 的不同时刻 | | input | string | 场景需求通过 input 体现，不污染 system prompt | | LLM | LlmFn 抽象 | reactor 不碰 fetch/baseUrl/apiKey，纯逻辑 | | Streaming | 不做 | 当前场景不需要，agent 场景也不做 streaming | | 副作用 | **Reactor 有副作用** | Reactor 会写入 thread（记录 ReAct loop 的对话到 .info.jsonl），不是纯函数 | | 错误恢复 | **所有 tool 失败都 recoverable** | unknown tool、执行失败统一返回 error tool result 给 LLM 重试，不 hard error。只有 maxRounds 耗尽才终止 | | 命名 | ThreadReactor | 明确是 workflow thread-scoped 的 reactor | ### 错误恢复策略 ```typescript // toolHandler 返回 string（成功的 tool result） // 如果 tool 执行失败，toolHandler 也返回 string（错误描述） // reactor 统一把结果作为 tool result 发回 LLM // // unknown tool → reactor 自动返回 "unknown tool: {name}" 给 LLM // tool 执行异常 → toolHandler catch 后返回错误描述 // LLM 拿到错误信息后自行决定重试或换策略 // 只有 maxRounds 耗尽才返回 err ``` ### 使用示例 ```typescript // Extract 场景 const extractReactor = createThreadReactor({ llm: createLlmFn(resolveModel(config, "extract")), systemPrompt: "You extract structured metadata from agent output...", tools: [casGetTool, extractTool], toolHandler: (call, thread) => { if (call.function.name === "cas_get") return handleCasGet(call, thread); // ... }, maxRounds: 10, }); const meta = await extractReactor(thread, agentContent, PlannerMetaSchema); // Supervisor 场景 const supervisor = createThreadReactor({ llm: createLlmFn(resolveModel(config, "supervisor")), systemPrompt: "You monitor thread health...", tools: [threadInspectTool, casGetTool], toolHandler: handleSupervisorTools, maxRounds: 5, }); const decision = await supervisor(thread, threadSummary, SupervisorDecisionSchema); ``` ## Phase 拆分 ### Phase 1: reactor 核心 + extract 迁移 - 新建 `src/reactor/` 模块 - 实现 `createThreadReactor`、`createLlmFn`、类型定义 - 从 `reactExtract` 提取通用 ReAct loop 逻辑 - 用 `createThreadReactor` 重写 `createExtract` - 删除 `reactExtract` 旧实现 - 所有 extract 测试通过 - Testing issue: 待创建 ### Phase 2: 迁移 supervisor - 用 `createThreadReactor` 重写 supervisor - Testing issue: 待创建 ## 完成标准 - [ ] 所有 Phase 的 testing issue 已 close - [ ] 全量测试通过 - [ ] reactExtract 旧实现已删除
xiaomo commented 2026-05-09 01:43:43 +00:00
Owner
Copy Link
Copy Source

RFC Review 意见

整体方向 ✅ — 从 reactExtract 里提取通用 ReAct loop 是正确的抽象。两阶段 API（config 固定 + 每次调用变化）设计清晰。几个建议：

1. LlmFn 签名需要考虑 streaming

当前 LlmFn 返回 Promise<Result<AssistantMessage, string>>，是纯 request-response。但 agent 场景（Phase 3+）通常需要 streaming 来：

  • 在 dashboard 实时显示 reactor 思考过程
  • 支持长输出不超时

建议要么现在就预留 streaming 变体，要么在 RFC 里明确标注 "Phase 1 不做 streaming，agent 场景再扩展"，避免后面加 streaming 时需要改 ThreadReactorConfig 的类型签名。

2. ThreadContext 在 reactor 里的角色不清晰

ThreadReactorFn 签名里有 thread: ThreadContext，toolHandler 也接收它。但 RFC 没说清楚 reactor 本身是否会写入 thread（append JSONL records）。

如果 reactor 不写 thread（只是 tool handler 可能读 thread data）→ 参数名叫 context 更准确
如果 reactor 会写 thread（把每轮对话记录到 .info.jsonl）→ 需要在 RFC 里明确这个行为

这影响 reactor 是纯函数还是有副作用，对测试策略有直接影响。

3. 错误恢复策略缺失

当前 reactExtract 遇到 unknown_tool 或 schema_validation_failed 直接返回 err。但通用 reactor 应该考虑：

  • LLM 调用了不存在的 tool → 是 hard error 还是发 correction message 让 LLM 重试？
  • tool 执行失败 → 把错误信息反馈给 LLM 继续，还是直接终止？

建议在 toolHandler 返回值里区分 fatal error vs recoverable error：

type ToolResult = 
  | { content: string }                    // 成功
  | { content: string; isError: true }     // 可恢复，反馈给 LLM
  // fatal error 直接 throw 或返回 Result err

4. Phase 拆分建议

Phase 2（迁移 extract）和 Phase 1 可以合并。理由：

  • reactor 的第一个消费者就是 extract，没有 extract 的验证，reactor 的 API 设计无法确认是否正确
  • 分开做意味着两次 PR review 看几乎相同的代码
  • 建议：Phase 1 = reactor + extract 迁移，Phase 2 = supervisor 迁移

5. Nit: 命名

ThreadReactor → 考虑是否需要 "Thread" 前缀。如果 reactor 是通用的 ReAct loop，它本身不一定绑定 thread 概念（thread context 是注入的）。叫 Reactor 或 createReactor 可能更简洁。

但如果刻意要强调 "这个 reactor 是 thread-scoped 的"，那 ThreadReactor 也行。

总结：方向正确，建议明确 #2（reactor 是否写 thread）和 #3（错误恢复策略），#4 的 phase 合并也值得考虑。

## RFC Review 意见 整体方向 ✅ — 从 `reactExtract` 里提取通用 ReAct loop 是正确的抽象。两阶段 API（config 固定 + 每次调用变化）设计清晰。几个建议： ### 1. LlmFn 签名需要考虑 streaming 当前 `LlmFn` 返回 `Promise<Result<AssistantMessage, string>>`，是纯 request-response。但 agent 场景（Phase 3+）通常需要 streaming 来： - 在 dashboard 实时显示 reactor 思考过程 - 支持长输出不超时 建议要么现在就预留 streaming 变体，要么在 RFC 里明确标注 "Phase 1 不做 streaming，agent 场景再扩展"，避免后面加 streaming 时需要改 `ThreadReactorConfig` 的类型签名。 ### 2. `ThreadContext` 在 reactor 里的角色不清晰 `ThreadReactorFn` 签名里有 `thread: ThreadContext`，`toolHandler` 也接收它。但 RFC 没说清楚 reactor 本身是否会写入 thread（append JSONL records）。 如果 reactor **不写** thread（只是 tool handler 可能读 thread data）→ 参数名叫 `context` 更准确 如果 reactor **会写** thread（把每轮对话记录到 .info.jsonl）→ 需要在 RFC 里明确这个行为 这影响 reactor 是纯函数还是有副作用，对测试策略有直接影响。 ### 3. 错误恢复策略缺失 当前 `reactExtract` 遇到 `unknown_tool` 或 `schema_validation_failed` 直接返回 err。但通用 reactor 应该考虑： - LLM 调用了不存在的 tool → 是 hard error 还是发 correction message 让 LLM 重试？ - tool 执行失败 → 把错误信息反馈给 LLM 继续，还是直接终止？ 建议在 `toolHandler` 返回值里区分 fatal error vs recoverable error： ```typescript type ToolResult = | { content: string } // 成功 | { content: string; isError: true } // 可恢复，反馈给 LLM // fatal error 直接 throw 或返回 Result err ``` ### 4. Phase 拆分建议 Phase 2（迁移 extract）和 Phase 1 可以合并。理由： - reactor 的第一个消费者就是 extract，没有 extract 的验证，reactor 的 API 设计无法确认是否正确 - 分开做意味着两次 PR review 看几乎相同的代码 - 建议：Phase 1 = reactor + extract 迁移，Phase 2 = supervisor 迁移 ### 5. Nit: 命名 `ThreadReactor` → 考虑是否需要 "Thread" 前缀。如果 reactor 是通用的 ReAct loop，它本身不一定绑定 thread 概念（thread context 是注入的）。叫 `Reactor` 或 `createReactor` 可能更简洁。 但如果刻意要强调 "这个 reactor 是 thread-scoped 的"，那 ThreadReactor 也行。 --- **总结**：方向正确，建议明确 #2（reactor 是否写 thread）和 #3（错误恢复策略），#4 的 phase 合并也值得考虑。
xiaomo commented 2026-05-09 02:02:55 +00:00
Owner
Copy Link
Copy Source

Review 决议

# 议题 结论
1 LlmFn streaming ❌ 不需要，agent 场景不做 streaming
2 ThreadContext 副作用 ✅ Reactor 有副作用（写 thread），RFC 应明确标注
3 错误恢复 ✅ 所有 tool 调用失败都是 recoverable — unknown tool、执行失败统一返回 fail 的 tool call result 给 LLM 重试，不 hard error
4 Phase 合并 ✅ Phase 1 + Phase 2 合并（reactor + extract 迁移一起做）
5 命名 ThreadReactor — 明确是 workflow thread 场景

建议更新 RFC 正文反映以上决议，然后开工 🫡

## Review 决议 | # | 议题 | 结论 | |---|------|------| | 1 | LlmFn streaming | ❌ 不需要，agent 场景不做 streaming | | 2 | ThreadContext 副作用 | ✅ Reactor 有副作用（写 thread），RFC 应明确标注 | | 3 | 错误恢复 | ✅ **所有** tool 调用失败都是 recoverable — unknown tool、执行失败统一返回 fail 的 tool call result 给 LLM 重试，不 hard error | | 4 | Phase 合并 | ✅ Phase 1 + Phase 2 合并（reactor + extract 迁移一起做） | | 5 | 命名 | `ThreadReactor` — 明确是 workflow thread 场景 | 建议更新 RFC 正文反映以上决议，然后开工 🫡
xiaoju commented 2026-05-09 02:06:09 +00:00
Author
Owner
Copy Link
Copy Source

Phase Testing Issues

Phase 1: reactor 核心 + extract 迁移

Phase 2: 迁移 supervisor

## Phase Testing Issues ### Phase 1: reactor 核心 + extract 迁移 - Testing issue: #140 ### Phase 2: 迁移 supervisor - Testing issue: #141
This repo is archived. You cannot comment on issues.
No Branch/Tag Specified
Branches Tags
Labels
No items
No Label
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
aobing jiashuang (Jia Shuang) jiayi (Jiayi) luming scottwei tuanzi xiaoju xiaomo xiaonuo xingyue
No Assignees
2 Participants
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: uncaged/workflow#139
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?