AuraBoot
免费开始
English简体中文

ConversationTurnService

ConversationTurnService 是执行一次对话回合的 唯一 入口 —— 无论触发方是 SSE chat 端点、IM @AI 提及、群聊 agent 回复，还是 Aura Connect Protocol(ACP)运行。根据红线 §12，所有 chat 实现 必须 调用 runTurn 或 resumeTurn。禁止旁路写 SSE/WebSocket frame、临时持久化或手写 broadcast 循环;该 chokepoint 的存在是为了让 turn 审计、outbox 发布、成本计量、approval gate 与 resume 语义保持统一。

概念

"一次 turn" 指 1 条用户消息 → 1 次 agent 响应的完整循环，期间可能包含工具调用和中断。该 Service 负责：

  1. 在 turn 事务内持久化用户消息。
  2. 解析 conversation、agent、model 以及工具列表。
  3. 执行成本上限、并发守护、approval gate。
  4. 驱动 model + tool 循环，将增量 streaming 到 ResponseSink。
  5. 把 turn-lifecycle 事件发布到 outbox。
  6. 中断时(approval 待批 / 工具错误 / 取消)，持久化状态供 resumeTurn 使用。

streaming 出口被 ResponseSink 抽象 —— HTTP SSE 用 SseResponseSink，IM/群聊用 BroadcastResponseSink。实现自定义 chat 入口的插件应提供新的 sink，绝不要写新的 turn driver。

接口签名

package com.auraboot.framework.conversation;

public interface ConversationTurnService {

    /** 启动一次新 turn。返回 sealed 结果 TurnOutcome
     *  (Success / Interrupted / PendingConfirmation / Failed)。 */
    TurnOutcome runTurn(TurnRequest request, ResponseSink sink);

    /** 恢复一次因 confirm 挂起的 turn(approval 批复 / 拒绝 / 取消后)。 */
    TurnOutcome resumeTurn(String pendingTurnId, ConfirmDecision decision, ResponseSink sink);

    enum ConfirmDecision { APPROVED, DENIED, CANCELLED }
}

TurnRequest 是一个 record，携带 tenantId、userId、humanMemberId、channel、agentCode、conversationId、clientMsgId、userMessage、pageContext(Map)、options(Map)、inboundMode 等字段。resumeTurn 通过 pendingTurnId 找回挂起的 turn 状态，并按 ConfirmDecision 决定批复 / 拒绝 / 取消;它 不会 新建 inbound message。

实现示例

一个自定义 HTTP 端点，把 tenant 内部 chat 面暴露给第三方 UI：

package com.acme.chat.controller;

import com.auraboot.framework.conversation.*;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.web.bind.annotation.*;
import org.springframework.http.MediaType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

@RestController
@RequestMapping("/api/acme/chat")
public class AcmeChatController {

    @Autowired private ConversationTurnService turns;
    @Autowired private ObjectMapper objectMapper;

    @PostMapping(path = "/stream",
                 produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter stream(@RequestBody TurnRequest req) {
        SseEmitter emitter = new SseEmitter(0L);
        // SseResponseSink(SseEmitter, ObjectMapper) —— 平台提供的 SSE 适配器
        ResponseSink sink  = new SseResponseSink(emitter, objectMapper);
        // runTurn 驱动循环;turn 完成或因 confirm 挂起时
        // 同步返回 TurnOutcome。
        turns.runTurn(req, sink);
        return emitter;
    }

    @PostMapping(path = "/{pendingTurnId}/resume",
                 produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter resume(@PathVariable String pendingTurnId,
                             @RequestParam ConversationTurnService.ConfirmDecision decision) {
        SseEmitter emitter = new SseEmitter(0L);
        ResponseSink sink  = new SseResponseSink(emitter, objectMapper);
        // resumeTurn 按 pendingTurnId 找回挂起状态，并按 ConfirmDecision 续跑
        turns.resumeTurn(pendingTurnId, decision, sink);
        return emitter;
    }
}

该实现 不会 做(你也不应做)以下事情：

  • 直接持久化消息到 ab_im_message。
  • 在 runTurn 返回后再写 SSE frame。
  • 重发 outbox 事件。
  • 触碰成本计量、approval 表或并发信号量。

这些都在 runTurn / resumeTurn 内部完成。

注册

宿主托管的单例 bean —— @Autowired 即可。这里没有 SPI;如需扩展行为，应扩展 agent(工具、prompt、model)或 sink(传输)，而非 turn driver。

常见陷阱

  • 为"简单"场景旁路。 即使是单步、无工具的 agent 也必须经 runTurn —— outbox 事件与 approval gate 对审计是必要的。
  • 在 runTurn 返回后写 SSE frame。 sink 拥有连接;双写会破坏流并泄漏状态。
  • 手写 resume。 resumeTurn 携带完整的中断状态 —— 切勿从 conversation log 重建。
  • 自定义工具里把工具错误吞掉。 应当 throw —— runTurn 会把它转换为一个模型可观测并可重试的 tool-error step。

相关