在SpringBoot中实现流式AI对话功能

@postmapping("/chat")
public chatresponse chat(@requestbody chatrequest request) {
    string answer = chatclient.prompt()
            .user(request.message())
            .call()
            .content();

    return new chatresponse(answer);
}

浏览器提交问题
      ↓
spring boot 接收请求
      ↓
调用大模型
      ↓
模型生成完整答案
      ↓
后端一次性返回
      ↓
浏览器展示内容

第 0.3 秒：你好
第 0.6 秒：，这个问题
第 0.9 秒：可以从
第 1.2 秒：三个方面理解
...

用户提问
   ↓
后端建立流式连接
   ↓
模型生成一个片段
   ↓
后端推送一个片段
   ↓
前端追加到页面
   ↓
直到生成完成

客户端发送一次问题
服务端持续返回生成片段

flux<string>
  ├─ "你好"
  ├─ "，"
  ├─ "可以"
  ├─ "这样"
  └─ "理解..."

text/event-stream

event: message
data: 你好

event: message
data: ，这是一个流式回答

event: done
data: [done]

flux<string> stream = flux.just("a", "b", "c");

flux<serversentevent<string>>

flux<string> output = chatclient.prompt()
        .user("tell me a joke")
        .stream()
        .content();

flux<chatresponse> response = chatclient.prompt()
        .user("tell me a joke")
        .stream()
        .chatresponse();

浏览器
  │
  │ post /api/ai/chat/stream
  ▼
spring boot api
  ├─ 参数校验
  ├─ 鉴权
  ├─ 会话加载
  ├─ prompt 组装
  ├─ spring ai 流式调用
  ├─ sse 事件封装
  ├─ 错误处理
  └─ 消息落库
  │
  ▼
大模型服务

{
  "conversationid": "c_1001",
  "messageid": "m_2001",
  "content": "你好"
}

flux<string>
  ↓
doonnext 追加到 stringbuilder
  ↓
dooncomplete 保存 assistant 消息

<dependencies>
    <dependency>
        <groupid>org.springframework.boot</groupid>
        <artifactid>spring-boot-starter-webflux</artifactid>
    </dependency>
    <dependency>
        <groupid>org.springframework.ai</groupid>
        <artifactid>spring-ai-starter-model-openai</artifactid>
    </dependency>
</dependencies>

spring:
  ai:
    openai:
      api-key: ${openai_api_key}
      chat:
        options:
          model: ${ai_chat_model:gpt-4.1-mini}
          temperature: 0.7

public record streamchatrequest(
        string conversationid,
        string message
) {
}

public record streamchatevent(
        string conversationid,
        string messageid,
        string type,
        string content
) {
    public static streamchatevent start(string conversationid, string messageid) {
        return new streamchatevent(conversationid, messageid, "start", "");
    }

    public static streamchatevent delta(string conversationid, string messageid, string content) {
        return new streamchatevent(conversationid, messageid, "delta", content);
    }

    public static streamchatevent done(string conversationid, string messageid) {
        return new streamchatevent(conversationid, messageid, "done", "");
    }

    public static streamchatevent error(string conversationid, string messageid, string content) {
        return new streamchatevent(conversationid, messageid, "error", content);
    }
}

@configuration
public class aiconfig {

    @bean
    chatclient chatclient(chatclient.builder builder) {
        return builder
                .defaultsystem("""
                        你是一个 java 后端技术助手。
                        回答要准确、简洁，并优先给出可落地的工程建议。
                        如果用户的问题缺少关键上下文，先提出澄清问题。
                        """)
                .build();
    }
}

@service
public class aistreamservice {

    private final chatclient chatclient;

    public aistreamservice(chatclient chatclient) {
        this.chatclient = chatclient;
    }

    public flux<streamchatevent> stream(streamchatrequest request) {
        string conversationid = request.conversationid();
        string assistantmessageid = uuid.randomuuid().tostring();
        stringbuilder fullcontent = new stringbuilder();

        flux<streamchatevent> start = flux.just(
                streamchatevent.start(conversationid, assistantmessageid)
        );

        flux<streamchatevent> content = chatclient.prompt()
                .user(request.message())
                .stream()
                .content()
                .map(chunk -> {
                    fullcontent.append(chunk);
                    return streamchatevent.delta(conversationid, assistantmessageid, chunk);
                })
                .dooncomplete(() -> {
                    string answer = fullcontent.tostring();
                    // todo 保存 assistant 消息
                    // messagerepository.saveassistantmessage(conversationid, assistantmessageid, answer);
                });

        flux<streamchatevent> done = flux.just(
                streamchatevent.done(conversationid, assistantmessageid)
        );

        return flux.concat(start, content, done)
                .onerrorresume(ex -> flux.just(
                        streamchatevent.error(conversationid, assistantmessageid, "生成失败，请稍后重试")
                ));
    }
}

@restcontroller
@requestmapping("/api/ai")
public class aistreamcontroller {

    private final aistreamservice aistreamservice;

    public aistreamcontroller(aistreamservice aistreamservice) {
        this.aistreamservice = aistreamservice;
    }

    @postmapping(
            value = "/chat/stream",
            produces = mediatype.text_event_stream_value
    )
    public flux<serversentevent<streamchatevent>> stream(
            @requestbody streamchatrequest request
    ) {
        return aistreamservice.stream(request)
                .map(event -> serversentevent.<streamchatevent>builder()
                        .event(event.type())
                        .data(event)
                        .build());
    }
}

async function streamchat(message) {
  const response = await fetch("/api/ai/chat/stream", {
    method: "post",
    headers: {
      "content-type": "application/json"
    },
    body: json.stringify({
      conversationid: "c_1001",
      message
    })
  });

  const reader = response.body.getreader();
  const decoder = new textdecoder("utf-8");
  let buffer = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });

    const events = buffer.split("\n\n");
    buffer = events.pop() || "";

    for (const rawevent of events) {
      const dataline = rawevent
        .split("\n")
        .find(line => line.startswith("data:"));

      if (!dataline) continue;

      const json = dataline.substring(5).trim();
      const event = json.parse(json);

      if (event.type === "delta") {
        appendmessagetext(event.content);
      }

      if (event.type === "done") {
        markmessagedone();
      }

      if (event.type === "error") {
        showerror(event.content);
      }
    }
  }
}

const controller = new abortcontroller();
fetch("/api/ai/chat/stream", {
  method: "post",
  signal: controller.signal,
  headers: {
    "content-type": "application/json"
  },
  body: json.stringify({ conversationid, message })
});
function stopgeneration() {
  controller.abort();
}

return aistreamservice.stream(request)
        .dooncancel(() -> {
            // todo 标记本轮消息被用户中断
        })
        .map(event -> serversentevent.<streamchatevent>builder()
                .event(event.type())
                .data(event)
                .build());

user      用户输入
assistant 模型回答

create table ai_message (
    id bigint primary key,
    conversation_id bigint not null,
    role varchar(32) not null,
    content text not null,
    status varchar(32) not null,
    model varchar(128),
    created_at timestamp not null
);

1. 收到请求后先保存 user 消息
2. 创建 assistant 消息，状态为 generating
3. 流式返回 delta
4. 完成后更新 assistant 内容和状态
5. 失败或取消时更新状态

location /api/ai/chat/stream {
    proxy_pass http://backend;
    proxy_http_version 1.1;
    proxy_buffering off;
    proxy_cache off;
    proxy_read_timeout 300s;
}

流式过程中内存聚合
完成后保存完整消息
异常或取消时保存当前已生成内容

{
  "type": "error",
  "content": "模型服务暂时不可用，请稍后重试"
}

用户提问
  ↓
改写查询
  ↓
向量召回
  ↓
重排序
  ↓
构造 prompt
  ↓
流式生成答案
  ↓
返回引用来源

{
  "type": "references",
  "content": [
    {
      "title": "售后政策",
      "url": "/docs/after-sales"
    }
  ]
}