ReAct 循环的 50 行 Go 实现，逐行拆解

张

张建站

2026/5/23 6:57:33

10分钟阅读

ReAct 循环的 50 行 Go 实现逐行拆解系列「企业级 AI Agent 实现拆解」第三篇。上一篇讲了 Session 聚合根和状态机——状态怎么迁移、事件怎么发、终态怎么判。但状态机本身是静态的谁在驱动这些迁移答案是RunTurnHandler.Handle()——每一轮用户消息进来它负责加载会话、组装上下文、把 ReAct 循环跑起来、然后落库收尾。先说一个架构决策ReAct 循环本身LLM 推理 → 工具调用 → 观察 → 继续推理我们没自己写交给 Eino 的react.NewAgent()跑。Handle()只做框架解决不了的部分Session 状态迁移、多租户 context 注入、长期记忆召回、事务落库、异步记忆提炼。下面是去掉错误处理样板代码后的主干大约 50 行func(h*RunTurnHandler)Handle(ctx context.Context,in RunTurnInput)error{// ① 加载会话校验状态sess,_:h.repo.Load(ctx,in.SessionID)ifsess.IsTerminal(){returnerrors.New(session already terminal)}// ② context 注入ctxcontext.WithValue(ctx,port.ContextKeyAgentTenantID{},sess.TenantID())ctxcontext.WithValue(ctx,port.ContextKeyAgentUserID{},sess.UserID())ctxcontext.WithValue(ctx,port.ContextKeyAgentSessionID{},string(sess.ID()))// ③ 前置 hook 加载配置h.invokeHook(func(hr port.HookRunner){hr.BeforeSession(ctx,sess.ID())})cfg,_:h.configs.Load(ctx,sess.TenantID(),sess.AgentConfig())// ④ 判断路径新 turn 还是 HITL 恢复isResume:in.UserTextrunInput:port.RunStreamInput{IsResume:isResume}varcapturedHistory[]model.MessageifisResume{// ⑤a HITL 恢复状态迁移 WAITING→RUNNING先落库再跑sess.Resume(*in.Decision)h.persist(ctx,sess)}else{// ⑤b 新 turn拉历史、写 user 消息、召回记忆、拼 system prompthistory,_:h.repo.ListMessages(ctx,sess.ID())sess.IncTurn()userMsg:model.NewUserMessage(sess.ID(),sess.TurnCount(),in.UserText)h.repo.AppendMessage(ctx,userMsg)historyappend(history,userMsg)capturedHistoryhistory memCtx:h.recallMemory(ctx,sess.TenantID(),sess.UserID(),in.UserText)runInput.SystemContentexpandSystemPrompt(cfg.SystemPrompt,sess)memCtx runInput.Historyhistory}// ⑥ 交给 EinoReAct 循环、工具调用、Hook callback 全在里面tr,interrupt,err:h.runnableFac.StreamTurn(ctx,cfg,sess.TenantID(),runInput,h.hooks,h.stream,sess.ID(),)iferr!nil{returnh.failSession(ctx,sess,...)}// ⑦ 处理 HITL 中断ifinterrupt!nil{returnh.handleEinoInterrupt(ctx,sess,interrupt)}defertr.Close()// ⑧ 消费流token → SSE turn.delta 收集 finalContentfinalContent,_:h.consumeEinoStream(ctx,sess.ID(),tr)// ⑨ 落库触发终态事件异步记忆提炼asstMsg:model.NewAssistantMessage(sess.ID(),sess.TurnCount(),finalContent,nil)sess.Complete()h.commitFinalAnswer(ctx,sess,asstMsg)ifh.memory!nil!isResumelen(capturedHistory)0{h.scheduleMemoryExtraction(sess,capturedHistory,finalContent)}returnnil}接下来逐块说每一段在做什么以及为什么这样设计。① 加载会话校验状态sess,err:h.repo.Load(ctx,in.SessionID)iferr!nil{returnh.onError(ctx,in.SessionID,err)}ifsess.IsTerminal(){returnerrors.New(session already terminal)}终态COMPLETED / FAILED / CANCELLED的 Session 不可以再跑新 Turn这是状态机的硬约束。在入口做这个检查比在循环里做更干净。② context 注入为什么用context.WithValue而不是 struct 字段因为 Runnable 按 AgentConfig 维度缓存跨 session 复用——如果 sessionID 绑到 struct 字段上缓存完第一个 session 后就错了。从 context 动态读缓存才能安全共享。ctxcontext.WithValue(ctx,port.ContextKeyAgentTenantID{},sess.TenantID())ctxcontext.WithValue(ctx,port.ContextKeyAgentUserID{},sess.UserID())ctxcontext.WithValue(ctx,port.ContextKeyAgentSessionID{},string(sess.ID()))这三个值随 context 一路传进 Eino 的chatModelAdapter和toolBrokerAdapter后者从 context 取出来做租户隔离和工具路由。③ 前置 hook 加载配置h.invokeHook(func(hr port.HookRunner){_hr.BeforeSession(ctx,sess.ID())})cfg,err:h.configs.Load(ctx,sess.TenantID(),sess.AgentConfig())BeforeSession是 Hook 链的第一个阶段典型用途是认证验证、配额预检、tracing 初始化。invokeHook做了 nil 检查没有注入 HookRunner 时静默跳过。AgentConfig包含 system prompt 模板、LLM profile、工具白名单、MaxTurns、HITL 开关等每次 Turn 都重新加载配置变更即时生效。④ 判断路径isResume:in.UserText一个布尔值区分两条路径UserText 表示 HITL 恢复从 Eino checkpoint 继续否则是新 Turn。两条路径的准备工作不同但最终都走同一个StreamTurn调用。⑤a HITL 恢复路径ifsess.State()!model.StateWaiting{returnfmt.Errorf(resume called but session not waiting, state%s,sess.State())}dec:model.InterruptDecision{Action:model.DecisionApprove}ifin.Decision!nil{dec*in.Decision}iferr:sess.Resume(dec);err!nil{_h.onError(ctx,sess.ID(),err)returnerr}iferr:h.persist(ctx,sess);err!nil{returnerr}先落库再跑。sess.Resume(dec)把状态从 WAITING 迁移到 RUNNING立刻写库。这样进程崩溃重启时session 已经是 RUNNING重试逻辑可以正确处理。如果先跑再落库崩溃后 session 还是 WAITING重试会重新走恢复路径可能重复执行。isResumetrue时runInput.History和runInput.SystemContent保持零值——Eino 收到nilinputMsgs 时自动从 CheckpointStore 加载上次暂停时的图状态。⑤b 新 turn 路径history,_:h.repo.ListMessages(ctx,sess.ID())sess.IncTurn()userMsg:model.NewUserMessage(sess.ID(),sess.TurnCount(),in.UserText)h.repo.AppendMessage(ctx,userMsg)historyappend(history,userMsg)capturedHistoryhistory memCtx:h.recallMemory(ctx,sess.TenantID(),sess.UserID(),in.UserText)runInput.SystemContentexpandSystemPrompt(cfg.SystemPrompt,sess)memCtx runInput.Historyhistory五件事按顺序做拉历史 → 增轮次计数 → 写 user 消息 → 召回长期记忆 → 组装 system prompt。recallMemory用本轮用户文本检索相关记忆片段追加到 system prompt 末尾失败时静默返回空字符串。expandSystemPrompt展开{{user_id}}、{{tenant_id}}、{{time}}三个占位符。capturedHistory留着会话结束后给记忆提炼用。⑥ 交给 Einotr,interrupt,err:h.runnableFac.StreamTurn(ctx,cfg,sess.TenantID(),runInput,h.hooks,h.stream,sess.ID(),)这一行是整个Handle的边界线左边是 DDD 领域逻辑右边是 Eino 框架内部。为什么切在这里而不是更早比如让 Eino 接管整个 Handle或更晚手写 ReAct for 循环因为状态加载、context 注入、记忆召回这些事和 LLM 编排无关框架不该管而 LLM 调用、工具分发、流式输出这些事已经有成熟实现没必要重写。这条边界让两边各自演进——换 Eino 版本不影响状态机改记忆策略不影响 Runnable 缓存。StreamTurn内部查 Runnable 缓存TTL 35 分钟→ 未命中则react.NewAgent()构建并编译 → 设置 compose 选项checkpoint ID、callbacks、force-new-run→runnable.Stream()→ 检测 interrupt 信号。应用层完全不感知这些。返回三种情况含义清晰返回值含义(TokenReader, nil, nil)正常流消费到 Final Answer(nil, *AgentInterruptInfo, nil)Eino 图级 HITL 中断(nil, nil, err)执行错误⑦ 处理 HITL 中断ifinterrupt!nil{it:model.NewPreToolInterrupt(hitl: tool call requires approval,model.ToolCall{},h.interruptTTL)sess.Pause(it)h.stream.Emit(ctx,sess.ID(),port.StreamEvent{Type:interrupt,Payload:map[string]any{before_nodes:info.BeforeNodes},})returnh.persist(ctx,sess)}中断信号来自 Eino 图级的WithInterruptBeforeNodes([tools])不是 Hook 返回值。Hook 的BeforeToolUse在 Eino 的 Callback 层执行应用层看到的是 Eino 已经决定了要中断只需要把 Session 状态迁移到 WAITING 并推 SSE 通知前端。⑧ 消费流varsb strings.Builderfor{content,err:tr.Recv()iferrors.Is(err,io.EOF){break}iferr!nil{returnsb.String(),err}ifcontent!{sb.WriteString(content)h.stream.Emit(ctx,sid,port.StreamEvent{Type:turn.delta,Payload:content})}}port.TokenReader是框架无关接口Recv() (string, error)。Eino 的schema.StreamReader在einoadapter包里被包装成这个接口这里看不到任何 Eino 类型。LLM 吐一个 tokenSSE 就推一帧没有额外 goroutine没有缓冲区。⑨ 落库、触发终态、异步记忆提炼asstMsg:model.NewAssistantMessage(sess.ID(),sess.TurnCount(),finalContent,nil)sess.Complete()h.commitFinalAnswer(ctx,sess,asstMsg)ifh.memory!nil!isResumelen(capturedHistory)0{h.scheduleMemoryExtraction(sess,capturedHistory,finalContent)}三件事commitFinalAnswer优先走事务路径消息写入和 session 状态更新在同一个数据库事务里。写完后emitTerminalEvents触发AfterSessionhook 和 SSEdone帧。注意persist()和emitTerminalEvents()是两个独立函数——persist只写库发事件SSEdone由emitTerminalEvents单独控制避免事务和非事务两条路径重复触发。scheduleMemoryExtraction起一个 goroutine30 秒超时把本轮 user/assistant 消息异步发给 memory BC 提炼记忆。失败只触发OnErrorhook不阻塞响应返回。HITL 恢复路径isResumetrue跳过避免同一会话重复提炼。流程图Handle(in) │ ├─ Load IsTerminal 检查 ├─ context 注入tenantID/userID/sessionID ├─ BeforeSession hook Load AgentConfig │ ├─── isResumetrue ──────────────────────────────────────┐ │ StateWaiting 校验 │ │ sess.Resume → persist先落库 │ │ runInput: IsResumetrue, Historynil │ │ │ └─── isResumefalse ─────────────────────────────────────┤ ListMessages IncTurn AppendMessage │ recallMemory expandSystemPrompt │ runInput: SystemContent History │ ↓ runnableFac.StreamTurn(...) ┌── Eino 内部 ──────────────┐ │ react.NewAgent ReAct 循环│ │ 工具调用 Hook callback │ └──────────────────────────┘ │ ┌─────────────┼──────────────┐ interrupt TokenReader error │ │ │ sess.Pause() consumeEinoStream failSession SSE interrupt SSE turn.delta persist 收集 finalContent │ sess.Complete() commitFinalAnswer事务 emitTerminalEvents → SSE done scheduleMemoryExtractiongoroutine小结回头看这 50 行主干Handle()的核心职责不是跑 ReAct 循环——那件事 Eino 做了。它的真正工作是做好框架不管的事Session 状态迁移的原子性、多租户 context 的透传、记忆的召回和异步提炼、事务落库和终态事件解耦。这些事情没有框架能替你做因为它们和具体业务域强绑定。改 Eino 版本不影响状态机改记忆策略不影响 Runnable 缓存改 Hook 不影响事务落库——这条边界线的价值就在这里。下一篇看 HITL 完整路径从 Eino 图级中断到人工审批再到 checkpoint 恢复执行。下一篇HITL 中断 —— 让人类随时叫停 AI 的正确姿势