AgentScope-Java 入门：用 SSE 展示评审任务进度

系列说明

本文是「AgentScope-Java 2.0.0-RC4 项目式入门」系列第 2 篇。

本篇对应：

学习目标：

• 理解为什么 Agent 应用需要事件流。
• 定义 ReviewEvent 和事件类型。
• 用 WebFlux 输出 text/event-stream。
• 用 Vue EventSource 连接后端。
• 让创建任务、执行任务、任务完成形成可观察链路。

为什么需要 SSE

代码评审不是一个瞬间完成的动作。

它至少包含这些阶段：

 复制代码创建任务
读取 Git diff
读取文件上下文
规则检查
模型评审
生成 findings
生成报告
任务完成

如果前端只发一个 HTTP 请求，然后等待最终结果，用户会遇到两个问题：

• 不知道后端是否还在运行。
• 不知道卡在哪个阶段。

所以我们需要事件流。

对这个项目来说，SSE 比 WebSocket 更合适：

定义 ReviewEvent

先定义事件对象：

 复制代码// ReviewEvent.java
public record ReviewEvent(
        String jobId,
        ReviewEventType type,
        String message,
        Map<String, Object> payload,
        Instant timestamp) {
}

事件类型：

 复制代码// ReviewEventType.java
public enum ReviewEventType {
    JOB_CREATED,
    DIFF_LOADED,
    DIFF_SUMMARIZED,
    FILE_CONTEXT_LOADED,
    RULE_CHECK_DONE,
    MODEL_REVIEWING,
    FINDING_GENERATED,
    REPORT_READY,
    JOB_COMPLETED,
    JOB_FAILED
}

这里的事件不是 AgentScope 的底层 AgentEvent，而是我们业务层的 ReviewEvent。

其中 DIFF_SUMMARIZED 在第 2 篇的模拟流程中不会出现，它是为第 3 篇预留的：当 Agent 对 diff 生成摘要后发布。提前在枚举里声明，避免后续跨篇加字段时编译不过。

用 Reactor Sinks 管理事件流

后端用 ReviewEventPublisher 维护每个任务的事件流。

核心思路：

 复制代码// ReviewEventPublisher.java
private final Map<String, Sinks.Many<ReviewEvent>> sinks = new ConcurrentHashMap<>();public void publish(ReviewEvent event) {
    sink(event.jobId()).tryEmitNext(event);
}public Flux<ReviewEvent> stream(String jobId) {
    return sink(jobId).asFlux();
}

每个 jobId 对应一个事件管道。

当评审执行到某一步，就调用：

 复制代码eventPublisher.publish(ReviewEvent.of(
        job.id(),
        ReviewEventType.DIFF_LOADED,
        "Git diff 已读取。",
        Map.of("changedFiles", diff.changedFiles())));

前端只订阅当前任务的事件，不需要关心其他任务。

提供 SSE API

Controller 增加接口：

 复制代码// ReviewController.java
@GetMapping(value = "/api/reviews/{id}/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ReviewEvent> streamEvents(@PathVariable String id) {
    return reviewService.streamEvents(id);
}

返回类型是 Flux<ReviewEvent>。

WebFlux 会把它编码成 SSE 数据。

浏览器端接收到的是一条条事件：

 复制代码{
  "jobId": "review-20260706-001",
  "type": "DIFF_LOADED",
  "message": "Git diff 已读取。",
  "payload": {
    "changedFiles": ["src/main/java/demo/DemoService.java"]
  },
  "timestamp": "2026-07-06T12:00:00Z"
}

让 ReviewService 发布阶段事件

任务创建后，不应该阻塞 HTTP 请求等待完整评审结束。

所以 createReview 做两件事：

1. 保存 ReviewJob。
2. 异步启动评审流程。

核心代码：

 复制代码// ReviewService.java
public Mono<ReviewJob> createReview(ReviewRequest request) {
    return Mono.fromCallable(() -> {
        ReviewJob job = ReviewJob.created(request, repoRoot.toString());
        jobStore.save(job);        eventPublisher.publish(ReviewEvent.of(
                job.id(),
                ReviewEventType.JOB_CREATED,
                "评审任务已创建。",
                Map.of("repoPath", job.repoPath())));        Mono.fromRunnable(() -> runReview(job.id()))
                .subscribeOn(Schedulers.boundedElastic())
                .subscribe();        return job;
    });
}

为什么这里用 boundedElastic？

因为后续会有文件读取、Git 命令、报告写入等阻塞操作。这类操作不应该占用 WebFlux 的事件循环线程。

前端封装 EventSource

前端 API 层：

 复制代码// reviewApi.ts
export function connectReviewEvents(id: string, handlers: ReviewEventHandlers) {
  const source = new EventSource(`/api/reviews/${id}/events`)  source.onmessage = (message) => {
    handlers.onEvent(JSON.parse(message.data))
  }  source.onerror = () => {
    handlers.onError?.()
  }  return source
}

Pinia store 里维护事件列表：

 复制代码// reviewStore.ts
function connectEvents(id: string) {
  eventSource?.close()
  eventSource = connectReviewEvents(id, {
    onEvent(event) {
      events.value.push(event)
      if (event.type === 'JOB_COMPLETED' || event.type === 'JOB_FAILED') {
        void load(id)
        eventSource?.close()
        eventSource = null
      }
    },
    onError() {
      error.value = 'SSE 连接已中断。请重新加载任务以刷新状态。'
    },
  })
}

这个 store 后续会继续扩展：

• 保存当前任务。
• 保存 findings。
• 保存 Markdown 报告。
• 复制报告。

进度页面

ReviewProgressView 做三件事：

• 加载当前 ReviewJob。
• 建立 SSE 连接。
• 展示事件时间线。

页面结构可以很简单：

 复制代码<template>
  <main>
    <h1>评审进度</h1>
    <ReviewStatusBadge :status="store.currentJob?.status" />
    <ReviewEventTimeline :events="store.events" />
  </main>
</template>

事件时间线不要设计得太复杂。

第一版只需要：

验证

启动后端和前端：

 复制代码cd backend
mvn spring-boot:run

 复制代码cd frontend
npm run dev

创建评审任务：

 复制代码curl -X POST  ^
  -H "Content-Type: application/json" ^
  -d "{"repoPath":"D:workspacedemo-project","diffMode":"WORKING_TREE","sessionId":"demo"}"

订阅事件流：

 复制代码curl -N {id}/events

预期能看到阶段事件不断输出。

前端访问：

 复制代码{id}/progress

第 2 篇的 runReview 还是模拟流程，目的是先验证事件流，不代表真实评审已经完成。预期页面显示类似：

• 评审任务已创建。
• 评审流水线已启动。
• 模拟阶段事件持续出现。
• 模拟任务完成。

从第 3 篇开始，这些模拟事件会逐步替换成真实的 Git diff、文件上下文和规则检查事件。第 4 篇才会生成真实 Markdown 报告。

设计思考：业务事件和 Agent 事件要分层

AgentScope-Java 本身有更细粒度的事件系统。

比如：

• ModelCallStartEvent
• ModelCallEndEvent
• TextBlockDeltaEvent
• ToolCallStartEvent
• ToolResultEndEvent

这些事件适合调试 Agent 内部执行。

但前端业务页面不一定需要直接展示它们。

对用户来说，“源码上下文已读取”比“某个 ToolResultBlock 结束”更容易理解。

所以本项目采用两层事件：

这也是做 Agent 产品时很重要的一点：不要把框架内部复杂度直接推给用户。

常见问题

SSE 连接中断怎么办

前端应该提示用户刷新状态。

本项目的状态会通过 GET /api/reviews/{id} 查询，所以 SSE 中断不应该导致任务丢失。

为什么不是 WebSocket

当前场景是服务端单向推送进度，SSE 更轻量。

等后续需要用户实时确认、暂停、继续时，再考虑 WebSocket 或 AgentScope HITL 事件。

为什么任务创建后立即返回

因为评审可能耗时较长。

POST /api/reviews 应该只负责创建任务，具体执行放到后台。

本篇小结

这一篇完成了：

• ReviewEvent。
• ReviewEventType。
• ReviewEventPublisher。
• GET /api/reviews/{id}/events。
• 后端异步评审流程。
• 前端 EventSource 连接。
• 进度页面和事件时间线。

下一篇，我们把“模拟流程”变成“真实评审”：读取本地 Git diff、读取文件上下文，并生成第一批规则化 findings。

系列导航