从零实现一个 ReAct Agent Loop——可中断、可流式、多模型支持

前言

现在 ReAct loop 的主要实现已经不再是秘密，网上可以看到各种模板。

但要把它变成能反复使用的 SDK 还有可在生产环境中跑，光这个循环远远不够。还需要考虑很多问题，比如：

• 使用者痛点一：使用者中途点取消。怎么干净的停下来 - 不让 tool_use 变成没有 tool_result 的孤儿，有一些LLM厂商直接报错。

• 使用者痛点二：如何定义流式结构。让前端更为方便的接入渲染逻辑，更快的呈现在使用者面前。

• 使用者痛点三：在流式基础上。如何优雅的处理模型同时调了多个工具执行，增加客户端体验。

• 使用者痛点四：想同时支持 Anthropic 和 OpenAI 兼容协议，需要屏蔽两边在 tool_use、流式事件、消息角色上的差异。

当然除了上述这些还有很多需要做的，我会在接下来的系列文章中再带大家一起调整。不过，这篇我们先打好地基。从零实现一个可中断、可流式、多模型可插拔的 ReAct Agent Loop。

文章每一 Feature 都有一个独立的小例子可以直接在线跑。链接放在每节末尾——一边读一边动手，比闷头读代码顺得多。

Part ：主要概念 + 最简实现

ReAct的循环很短，一张图能讲清楚：

flowchart LR
Start --> A
A --> B{有 tool_calls?话说回来，}
B -- 否 --> End
B -- 是 --> C
C --> D
D --> A

为了让代码能直接跑。我们用 DeepSeek 作为示例。老实说，先准备一个简单工具根据城市名返回假的天气。

// OpenAI 接受的工具定义结构
const tools =。},},},]，// 工具的具体实现
const toolImpls: Record Promise = {
get_wear: async => {
await new Promise => setTimeout);return `${city} 今天 °C 晴`;},}，

主循环由于我们是以 SDK 的角度去实现。而且后续代码会越来越多，所以采用 Class 来实现，方便维护。

export class Agent {
// LLM 实例
private client: OpenAI;// 内部维护消息列表，这一步为了让LLM有记忆，感知到上下文。话说回来，private messages: any =;constructor {
// 创建实例
this.client = new OpenAI({
apiKey。baseURL: "https://api.deepseek.com/v1",});}
async run {
console.log;this.messages.push;// 创建 while agent loop，内部判断无工具调用时再退出
while {
console.log;const resp = await this.client.chat.completions.create({
model这方面。"deepseek-chat",messages: this.messages,tools,});const assistant = resp.choices.message;this.messages.push;// 调出循环的唯一条件，没有工具调用。也就是LLM用纯文本回复
if {
console.log;return，}
// 打印工具调用的日志
const summary = assistant.tool_calls
.map => `${c.function.name}`)
.join;console.log，// 批量调用工具
const toolMsgs = await Promise.all(
assistant.tool_calls.map => {
const args = JSON.parse;const result = await toolImpls;console.log，return { role: "tool" as const。tool_call_id: call.id,content: result };})
)，this.messages.push;}
}
}

上面就是最简化的 ReAct Agent Loop 版本。可以直接使用该类运行一下：

const agent = new Agent;await agent.run;

Part ：支持使用者中途取消

User Pain Point：模型已经生成到一半。使用者嫌它跑偏，点了停止按钮，却导致孤儿 tool_call 报错。难道不该干净退出吗，老实说，

最自然的方式是 Web 标准的 AbortController**——把 signal 丢给所有需要感知取消的异步操作。调一次 controller.abort,所有挂在这个 signal 上的操作都会收到通知。

OpenAI / Anthropic 的官方 SDK 都支持给请求传 signal. 我们要做的是：

1. 主循环每一步开始时检查 alert.signal.aborted
2. L​LM 请求带上 signal
3. 工具调用也要感知 signal
4. 不要直接 throw AbortError**——最终一条 tool_call 未完成就抛，会导致下次 resume 时出现孤儿错误。我们要做的是“干净退出”。

#1 工具层面支持取消**:

const toolImpls: Record Promise = {
get_wear: async => {
await new Promise => setTimeout);if throw signal.reason;return `${city} 今天 °C 晴`;},}，

#2 在 Agent 类里加入 abort 方法并在循环里检查**:

export class Agent {
private _abortController: AbortController | null = null;abort {
this._abortController?.abort，}
async run {
this._abortController = new AbortController;const signal = this._abortController.signal;try {
while {
if {
console.log);return，不过，}
console.log;const resp = await this.client.chat.completions.create(
{ model:"deepseek-chat"。messages:this.messages,tools },{ signal }
);const assistant = resp.choices.message;this.messages.push;if {
console.log;return，其实，}
console.log;if { console.log;return，}
const toolMsgs = await Promise.all(
assistant.tool_calls.map=>{
const args = JSON.parse;const result = await toolImpls;console.log: ${result}`);return { role:"tool" as const。
tool_call_id:call.id,content:result };})
)，this.messages.push;}
} catch {
if {
console.log;this._finalizeOnAbort);return，其实，}
throw err;} finally{
this._abortController=null;}
}
/** 补齐未完成的 tool_result，使得协议自洽 */
private _finalizeOnAbort{
const last=this.messages.findLast=>m.role==="assistant");if return，const done=new Set=>m.role==="tool").map);for{
if) continue;老实说，this.messages.push({
从role来看，"tool"。tool_call_id:call.id,content:` ${reason}`,});console.log，}
}
}

#3 调用方示例****:

const agent = new Agent;console.log，setTimeout => agent.abort"),5000);try{
await agent.run;} finally{
// 清理计时器等资源
}

• #4 小细节提醒**：
  • *abort 检查点放在两个地方*：每次新 step 开头、tool 调用前。按理说，这样即使模型已经返回了带有 tool_calls 的 assistant 消息。也不会因为后续 abort 而丢失已推送到 history 的信息。
  • *messages 状态保持自洽*：若 abort 前已经 push 助手消息但未 push 对应 tool_result。需要通过 `补齐占位，以免下次 resume 报错。怎么说呢，

Part ：流式事件结构——前端要的是渐进式可见

User Pain Point： 没有流式输出时前端只能等全部响应返回才渲染。“卡顿”体验极差，而大多数使用者已习惯“字一个个蹦出来”。我们必须把 LLM 的 token 流实时推送给前端，同时保持协议完整性。

L​LM API 原生支持流式：只要在请求里加 { stream:true }。SDK 会返回一个 AsyncIterable，每个 chunk 都是增量 delta。但这些 delta 天然碎散，例如：

• { content:"北" }
• { tool_calls: }
• { …}

再看*问题1*。Tool 使用时参数 JSON 在流途中会出现半成品，如 {"city":"上";怎么说呢，若直接解析会炸。从*问题2*来看，不同消费场景对流数据需求不同——前端想逐 token 渲染；CLI 想最终答案，持久化只想保存定稿。

说到设计目标**，统一两类信号。让消费方不必处理碎片**:

• `partial`事件**：当前完整快照，每次 yield 替换旧状态。消费方只需 `setMessage` 即可渲染。
• `message`事件**：定稿——assistant 完整回复或某个 `tool_result` 全部收齐后发送，可用于持久化或关闭流。话说回来，

Agent 内部累积逻辑 & 对外接口**:

type AgentEvent =
| { type:"partial";message:any }
| { type:"message";message:any };class Agent{
// 与 Part 相同字段 + abort + _finalizeOnAbort
/** 流式入口 */
async *stream: AsyncGenerator{
/* 同 Part 基础结构，只是把 “await create” 换成逐块消费 */
/* 每读取一个 chunk 累积后 yield partial */
/* 完整助手消息生成后 yield message */
/* 工具结果一样走 message */
}
/** CLI / 持久化快捷方式 */
async invoke: Promise{
const collected:any=;for await){
if collected.push;}
return collected;}
}

主要 `_streamAssistant` 实现****:

private async *_streamAssistant: AsyncGenerator{
const stream=await this.client.chat.completions.create(
{model这方面。"deepseek-chat",messages:this.messages,tools,stream:true},{signal}
);let text=""，说起来，const toolCalls:{id:string;name:string，argsJson:string}=;let snapshot:any=null;for await{
const delta=chunk.choices?.delta，if continue;
 if text+=delta.content;for{
const idx=tc.index;toolCalls，?话说回来，={id:"",name:"",argsJson:""};if toolCalls.id=tc.id;if toolCalls.name=tc.function.name;if toolCalls.argsJson+=tc.function.arguments;}
snapshot={
至于role，"assistant"。content:text||null,tool_calls:
toolCalls.length?toolCalls.map(tc=>({
从id来看，tc.id,type:"function"。function:{name:tc.name,arguments:tc.argsJson}
再看}))，undefined,};yield {type:"partial",message:snapshot};

}

return snapshot;// 最终完整 Assistant 消息交给外层循环
}


使用示例****:

// 前端逐 token 渲染
for await){
if setMessage;if commitMessage;}


// CLI / 持久化一次性获取全部消息
const msgs=await agent.invoke;说起来，console.log;

Part ：Promise.race - 多工具并行支持流式。提高客户端体验 

*User Pain Point*：当前使用 Pormise.all 会等所有工具都结束才返回，如果其中一个慢工具耗时几秒，会导致整个回合卡住“又卡住了”。我们希望能够 **边完成边推送** 已完成的结果，让 UI 能即时刷新。


ts

private async *_act:AsyncGenerator{

// 将每个 Tool 包装为立即执行函数并存入 pending 数组
















🧪 在线实验室

从尝试以下示例来看。


npm install ...



📚 Reference

...


(https的观点是，//img.shields.io/github/stars/lhz960904/c…

🗒️ 注释与待办事项

🌐 官方文档


💡 想法/建议/反馈