mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 11:13:43 +08:00
feat(generative): unified async pipeline + live SSE delivery for music/video/image
This commit is contained in:
matevip
parent
1f368336de
commit
e3ab06d57c
@ -1374,8 +1374,13 @@ public class ChatController {
|
||||
* 注册 SseEmitter 的完整生命周期回调
|
||||
*/
|
||||
private void registerEmitterCallbacks(SseEmitter emitter, String conversationId) {
|
||||
emitter.onCompletion(() ->
|
||||
log.debug("SSE emitter completed: conversationId={}", conversationId));
|
||||
emitter.onCompletion(() -> {
|
||||
log.debug("SSE emitter completed: conversationId={}", conversationId);
|
||||
// Detach immediately so a subsequent broadcast (heartbeat / async_task_*)
|
||||
// doesn't waste a send call on the zombie emitter and emit
|
||||
// "Removing dead subscriber ... ResponseBodyEmitter has already completed".
|
||||
streamTracker.detach(conversationId, emitter);
|
||||
});
|
||||
emitter.onTimeout(() -> {
|
||||
log.debug("SSE emitter timeout: conversationId={}", conversationId);
|
||||
streamTracker.detach(conversationId, emitter);
|
||||
|
||||
@ -246,33 +246,73 @@ public class ChatStreamTracker {
|
||||
}
|
||||
|
||||
/**
|
||||
* 广播事件到所有订阅者并缓存到 buffer
|
||||
* 注意："done" 事件即使在流已完成状态下也会被发送，确保客户端能收到完成信号
|
||||
* 广播事件到所有订阅者并缓存到 buffer.
|
||||
* <p>
|
||||
* Two event categories survive {@code state.done=true}:
|
||||
* <ul>
|
||||
* <li>{@code "done"} — the lifecycle marker itself. If a client missed
|
||||
* this on a broken pipe and reconnects within the 5-minute retention
|
||||
* window, replay surfaces it so the UI exits "生成中" state.</li>
|
||||
* <li>{@code "async_task_*"} — task lifecycle events from
|
||||
* {@code AsyncTaskService} (image/video/music generation). These
|
||||
* routinely fire <em>after</em> the agent's reasoning turn finishes
|
||||
* (long-running upstream calls). Without this carve-out the events
|
||||
* are silently dropped and the UI never sees the audio/error.</li>
|
||||
* </ul>
|
||||
* For all other events, the prior {@code state==null || state.done}
|
||||
* early-return remains.
|
||||
*/
|
||||
public void broadcast(String conversationId, String eventName, String jsonData) {
|
||||
RunState state = runs.get(conversationId);
|
||||
|
||||
// 特殊处理 "done" 事件：即使流已完成，仍然尝试发送给所有订阅者，
|
||||
// 并且**也必须入 buffer**——这样如果客户端在生成期间 SSE 断了
|
||||
// (broken pipe / 浏览器 tab throttle / 网络抖动)，刷新页面重连
|
||||
// 时仍能从 buffer 回放 done 事件，UI 不再永远卡在"生成中"。
|
||||
// 之前的设计 done 不入 buffer，配合 complete() 立即 runs.remove()
|
||||
// 一起，使得 SSE 中途断开 = done 永远丢，是这次故障的根源。
|
||||
if ("done".equals(eventName)) {
|
||||
if (state != null) {
|
||||
SseEvent doneEvent = new SseEvent(eventName, jsonData);
|
||||
synchronized (state.lock) {
|
||||
state.buffer.add(doneEvent);
|
||||
Iterator<SseEmitter> it = state.subscribers.iterator();
|
||||
while (it.hasNext()) {
|
||||
SseEmitter emitter = it.next();
|
||||
try {
|
||||
emitter.send(SseEmitter.event().name(eventName).data(jsonData));
|
||||
boolean isDone = "done".equals(eventName);
|
||||
boolean isAsyncTask = eventName != null && eventName.startsWith("async_task_");
|
||||
boolean isHeartbeat = "heartbeat".equals(eventName);
|
||||
|
||||
if (isDone || isAsyncTask) {
|
||||
if (state == null) return;
|
||||
SseEvent ev = new SseEvent(eventName, jsonData);
|
||||
synchronized (state.lock) {
|
||||
state.buffer.add(ev);
|
||||
if (state.buffer.size() > MAX_BUFFER_SIZE) {
|
||||
trimBuffer(state.buffer);
|
||||
}
|
||||
Iterator<SseEmitter> it = state.subscribers.iterator();
|
||||
while (it.hasNext()) {
|
||||
SseEmitter emitter = it.next();
|
||||
try {
|
||||
emitter.send(SseEmitter.event().name(eventName).data(jsonData));
|
||||
if (isDone) {
|
||||
log.debug("Sent final 'done' event to subscriber for {}", conversationId);
|
||||
} catch (IOException | IllegalStateException e) {
|
||||
log.debug("Removing dead subscriber for {} while sending done event: {}", conversationId, e.getMessage());
|
||||
it.remove();
|
||||
}
|
||||
} catch (IOException | IllegalStateException e) {
|
||||
log.debug("Removing dead subscriber for {} while sending {} event: {}",
|
||||
conversationId, eventName, e.getMessage());
|
||||
it.remove();
|
||||
}
|
||||
}
|
||||
}
|
||||
// done events do not flow through eventRelays; async_task_* should
|
||||
// also short-circuit since relays exist for delta-style streaming
|
||||
// events, not lifecycle markers.
|
||||
return;
|
||||
}
|
||||
|
||||
// Heartbeat is ephemeral keep-alive — must reach subscribers even when
|
||||
// state.done=true (e.g. a reconnected emitter waiting for late
|
||||
// async_task_* events). Skip the buffer (heartbeats are not replayable).
|
||||
if (isHeartbeat) {
|
||||
if (state == null) return;
|
||||
synchronized (state.lock) {
|
||||
Iterator<SseEmitter> it = state.subscribers.iterator();
|
||||
while (it.hasNext()) {
|
||||
SseEmitter emitter = it.next();
|
||||
try {
|
||||
emitter.send(SseEmitter.event().name(eventName).data(jsonData));
|
||||
} catch (IOException | IllegalStateException e) {
|
||||
log.debug("Removing dead subscriber for {} while sending heartbeat: {}",
|
||||
conversationId, e.getMessage());
|
||||
it.remove();
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -382,18 +422,26 @@ public class ChatStreamTracker {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
// 流已完成：buffer 已回放完毕（含 done event），不需要订阅后续事件
|
||||
// Stream complete: buffer replayed (including the `done` event itself).
|
||||
// We DO NOT auto-complete the emitter here — keep it subscribed so any
|
||||
// late-arriving async_task_* events (image/video/music generation that
|
||||
// outlasts the agent's reasoning turn) reach the client live. Idle
|
||||
// emitters are pruned naturally when:
|
||||
// - the next broadcast hits a broken pipe and removes the dead subscriber
|
||||
// - cleanupStaleRuns() removes the RunState after DONE_RETENTION_MS (5 min)
|
||||
// - the frontend explicitly disconnects (component unmount / navigation)
|
||||
// Without this, async_task_completed fired after `done` would be silently
|
||||
// dropped, leaving the chat UI stuck on the "正在生成中" placeholder.
|
||||
state.subscribers.add(emitter);
|
||||
if (state.done) {
|
||||
log.info("[SSE] Replayed {} buffered events to reconnecting client for completed stream: {}",
|
||||
log.info("[SSE] Replayed {} buffered events; emitter stays subscribed for late async events: {}",
|
||||
state.buffer.size(), conversationId);
|
||||
try {
|
||||
emitter.complete();
|
||||
} catch (Exception ignored) {
|
||||
// emitter 已被 servlet 容器关掉了，无需处理
|
||||
}
|
||||
// Restart heartbeat so the proxy/Tomcat 60s idle timeout doesn't
|
||||
// close the reconnected emitter before the async_task_* event fires.
|
||||
// The scheduler self-stops once subscribers go empty (see startHeartbeat).
|
||||
startHeartbeat(conversationId);
|
||||
return true;
|
||||
}
|
||||
state.subscribers.add(emitter);
|
||||
}
|
||||
log.info("[SSE] Client reconnected for conversation={}, replaying {} buffered events",
|
||||
conversationId, state.buffer.size());
|
||||
@ -527,7 +575,16 @@ public class ChatStreamTracker {
|
||||
state.heartbeatFuture = heartbeatScheduler.scheduleAtFixedRate(() -> {
|
||||
try {
|
||||
RunState s = runs.get(conversationId);
|
||||
if (s == null || s.done) {
|
||||
if (s == null) {
|
||||
stopHeartbeat(conversationId);
|
||||
return;
|
||||
}
|
||||
// Continue heartbeating post-done as long as someone is still listening
|
||||
// (reconnected emitter waiting for late async_task_* events). Stop only
|
||||
// when the run is done AND the subscribers list is empty — otherwise the
|
||||
// 60s idle proxy timeout drops the reconnected emitter and async events
|
||||
// never reach the client live.
|
||||
if (s.done && s.subscribers.isEmpty()) {
|
||||
stopHeartbeat(conversationId);
|
||||
return;
|
||||
}
|
||||
|
||||
@ -61,6 +61,7 @@ public class SystemSettingService {
|
||||
private static final String KLING_SECRET_KEY_KEY = "klingSecretKey";
|
||||
private static final String RUNWAY_API_KEY_KEY = "runwayApiKey";
|
||||
private static final String MINIMAX_API_KEY_KEY = "minimaxApiKey";
|
||||
private static final String MINIMAX_REGION_KEY = "minimaxRegion";
|
||||
|
||||
private final SystemSettingMapper systemSettingMapper;
|
||||
|
||||
@ -107,6 +108,7 @@ public class SystemSettingService {
|
||||
dto.setKlingSecretKeyMasked(maskApiKey(getValue(KLING_SECRET_KEY_KEY, "")));
|
||||
dto.setRunwayApiKeyMasked(maskApiKey(getValue(RUNWAY_API_KEY_KEY, "")));
|
||||
dto.setMinimaxApiKeyMasked(maskApiKey(getValue(MINIMAX_API_KEY_KEY, "")));
|
||||
dto.setMinimaxRegion(getValue(MINIMAX_REGION_KEY, "global"));
|
||||
|
||||
// 图片生成配置
|
||||
dto.setImageEnabled(Boolean.parseBoolean(getValue(IMAGE_ENABLED_KEY, "false")));
|
||||
@ -237,6 +239,9 @@ public class SystemSettingService {
|
||||
if (dto.getMinimaxApiKey() != null && !dto.getMinimaxApiKey().isBlank()) {
|
||||
saveValue(MINIMAX_API_KEY_KEY, dto.getMinimaxApiKey(), "MiniMax API Key");
|
||||
}
|
||||
if (dto.getMinimaxRegion() != null && !dto.getMinimaxRegion().isBlank()) {
|
||||
saveValue(MINIMAX_REGION_KEY, dto.getMinimaxRegion(), "MiniMax API 区域 (global / cn)");
|
||||
}
|
||||
|
||||
// 图片生成配置
|
||||
if (dto.getImageEnabled() != null) {
|
||||
|
||||
@ -34,9 +34,12 @@ public class AsyncTaskService implements ApplicationRunner {
|
||||
private final AsyncTaskMapper asyncTaskMapper;
|
||||
private final ChatStreamTracker streamTracker;
|
||||
|
||||
/** 轮询线程池（小池，2 线程足够） */
|
||||
/** Polling thread pool. Bumped from 2 to 8 in P0 — image+video+future generative
|
||||
* tasks all share this pool, and per-task work (poll HTTP + DB write + file
|
||||
* download in completion callbacks) is non-trivial; 2 threads saturate
|
||||
* immediately under any concurrent load. */
|
||||
private final ScheduledExecutorService pollExecutor =
|
||||
Executors.newScheduledThreadPool(2, r -> {
|
||||
Executors.newScheduledThreadPool(8, r -> {
|
||||
Thread t = new Thread(r, "async-task-poll");
|
||||
t.setDaemon(true);
|
||||
return t;
|
||||
@ -267,12 +270,26 @@ public class AsyncTaskService implements ApplicationRunner {
|
||||
|
||||
public void broadcastTaskEvent(AsyncTaskEntity task, String eventName,
|
||||
boolean success, String videoUrl, String imageUrl, String errorMessage) {
|
||||
Map<String, Object> extra = new HashMap<>();
|
||||
if (videoUrl != null) extra.put("videoUrl", videoUrl);
|
||||
if (imageUrl != null) extra.put("imageUrl", imageUrl);
|
||||
broadcastTaskEventWithData(task, eventName, success, extra, errorMessage);
|
||||
}
|
||||
|
||||
/**
|
||||
* Generic task-event broadcaster. Use this for any media kind where the URL
|
||||
* field name varies (audioUrl, modelUrl, ...) — pass it via {@code extraData}.
|
||||
* Named distinctly from {@link #broadcastTaskEvent} to avoid overload
|
||||
* ambiguity when callers pass {@code null} for the 4th argument.
|
||||
*/
|
||||
public void broadcastTaskEventWithData(AsyncTaskEntity task, String eventName,
|
||||
boolean success, Map<String, Object> extraData,
|
||||
String errorMessage) {
|
||||
Map<String, Object> data = new HashMap<>();
|
||||
data.put("taskId", task.getTaskId());
|
||||
data.put("taskType", task.getTaskType());
|
||||
data.put("success", success);
|
||||
if (videoUrl != null) data.put("videoUrl", videoUrl);
|
||||
if (imageUrl != null) data.put("imageUrl", imageUrl);
|
||||
if (extraData != null) data.putAll(extraData);
|
||||
if (errorMessage != null) data.put("errorMessage", errorMessage);
|
||||
streamTracker.broadcastObject(task.getConversationId(), eventName, data);
|
||||
}
|
||||
|
||||
@ -38,7 +38,9 @@ public class ImageGenerationResult {
|
||||
.providerName(providerName)
|
||||
.submitted(true)
|
||||
.completed(false)
|
||||
.message("图片生成任务已提交（任务 ID: " + taskId + "）。预计 30 秒 - 2 分钟完成，完成后会自动显示在对话中。")
|
||||
// Format MUST keep `taskId=...` so the frontend reconnect detector
|
||||
// (useChat.ts TASK_ID_PATTERN) can extract it from the tool result.
|
||||
.message("图片生成任务已提交（taskId=" + taskId + ", provider=" + providerName + "）。预计 30 秒 - 2 分钟完成，完成后会自动显示在对话中。")
|
||||
.build();
|
||||
}
|
||||
|
||||
|
||||
@ -253,8 +253,10 @@ public class ImageGenerationService {
|
||||
ImageProviderCapabilities caps = provider.detailedCapabilities();
|
||||
if (caps == null) return;
|
||||
|
||||
request.setSize(caps.normalizeSize(request.getSize()));
|
||||
request.setAspectRatio(caps.normalizeAspectRatio(request.getAspectRatio()));
|
||||
// Resolve aspect ratio first so size normalization can preserve orientation.
|
||||
String aspectRatio = caps.normalizeAspectRatio(request.getAspectRatio());
|
||||
request.setAspectRatio(aspectRatio);
|
||||
request.setSize(caps.normalizeSize(request.getSize(), aspectRatio));
|
||||
if (request.getCount() != null) {
|
||||
request.setCount(caps.normalizeCount(request.getCount()));
|
||||
}
|
||||
|
||||
@ -39,20 +39,50 @@ public class ImageProviderCapabilities {
|
||||
private List<String> models = List.of();
|
||||
|
||||
/**
|
||||
* 将请求的 size 就近匹配到 provider 支持的值
|
||||
* Match the requested size against supported sizes by area only.
|
||||
* Orientation-blind — prefer {@link #normalizeSize(String, String)} when an
|
||||
* aspect ratio is available so portrait/landscape intent is preserved.
|
||||
*/
|
||||
public String normalizeSize(String requested) {
|
||||
if (requested == null || requested.isBlank()) {
|
||||
return supportedSizes.isEmpty() ? "1024x1024" : supportedSizes.get(0);
|
||||
return normalizeSize(requested, null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Match the requested size against supported sizes, preserving orientation.
|
||||
* <p>Resolution order:
|
||||
* <ol>
|
||||
* <li>If {@code requestedSize} is already in {@code supportedSizes}, return it.</li>
|
||||
* <li>If {@code requestedAspectRatio} is given, narrow {@code supportedSizes}
|
||||
* to those whose orientation matches (portrait / landscape / square),
|
||||
* then pick by closest area.</li>
|
||||
* <li>Otherwise pick by closest area across all supported sizes.</li>
|
||||
* </ol>
|
||||
*/
|
||||
public String normalizeSize(String requestedSize, String requestedAspectRatio) {
|
||||
if (supportedSizes.isEmpty()) {
|
||||
return "1024x1024";
|
||||
}
|
||||
if (supportedSizes.contains(requested)) {
|
||||
return requested;
|
||||
if (requestedSize != null && supportedSizes.contains(requestedSize)) {
|
||||
return requestedSize;
|
||||
}
|
||||
// 就近匹配：解析面积，找最接近的
|
||||
long reqArea = parseArea(requested);
|
||||
String closest = supportedSizes.get(0);
|
||||
|
||||
Orientation targetOrientation = orientationFor(requestedAspectRatio);
|
||||
List<String> candidates = supportedSizes;
|
||||
if (targetOrientation != null) {
|
||||
List<String> matching = supportedSizes.stream()
|
||||
.filter(s -> orientationOf(s) == targetOrientation)
|
||||
.toList();
|
||||
if (!matching.isEmpty()) {
|
||||
candidates = matching;
|
||||
}
|
||||
}
|
||||
|
||||
long reqArea = (requestedSize == null || requestedSize.isBlank())
|
||||
? 1024L * 1024L
|
||||
: parseArea(requestedSize);
|
||||
String closest = candidates.get(0);
|
||||
long minDiff = Math.abs(reqArea - parseArea(closest));
|
||||
for (String s : supportedSizes) {
|
||||
for (String s : candidates) {
|
||||
long diff = Math.abs(reqArea - parseArea(s));
|
||||
if (diff < minDiff) {
|
||||
minDiff = diff;
|
||||
@ -62,6 +92,34 @@ public class ImageProviderCapabilities {
|
||||
return closest;
|
||||
}
|
||||
|
||||
private enum Orientation { PORTRAIT, LANDSCAPE, SQUARE }
|
||||
|
||||
private static Orientation orientationFor(String aspectRatio) {
|
||||
if (aspectRatio == null || aspectRatio.isBlank()) return null;
|
||||
String[] parts = aspectRatio.split(":");
|
||||
if (parts.length != 2) return null;
|
||||
try {
|
||||
double w = Double.parseDouble(parts[0].trim());
|
||||
double h = Double.parseDouble(parts[1].trim());
|
||||
if (w == h) return Orientation.SQUARE;
|
||||
return w > h ? Orientation.LANDSCAPE : Orientation.PORTRAIT;
|
||||
} catch (NumberFormatException e) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private static Orientation orientationOf(String size) {
|
||||
try {
|
||||
String[] parts = size.toLowerCase().split("x");
|
||||
long w = Long.parseLong(parts[0].trim());
|
||||
long h = Long.parseLong(parts[1].trim());
|
||||
if (w == h) return Orientation.SQUARE;
|
||||
return w > h ? Orientation.LANDSCAPE : Orientation.PORTRAIT;
|
||||
} catch (Exception e) {
|
||||
return Orientation.SQUARE;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 将请求的 aspectRatio 就近匹配或回退到默认
|
||||
*/
|
||||
|
||||
@ -100,9 +100,11 @@ public class DashScopeImageProvider implements ImageGenerationProvider {
|
||||
input.put("prompt", request.getPrompt());
|
||||
|
||||
ObjectNode parameters = body.putObject("parameters");
|
||||
String size = aspectRatioToSize(request.getAspectRatio());
|
||||
if (size != null) {
|
||||
parameters.put("size", size);
|
||||
// request.size already normalized by ImageGenerationService to one of supportedSizes.
|
||||
// DashScope API uses '*' separator instead of 'x'.
|
||||
String size = request.getSize();
|
||||
if (size != null && !size.isBlank()) {
|
||||
parameters.put("size", size.replace("x", "*"));
|
||||
}
|
||||
int count = request.getCount() != null ? Math.min(request.getCount(), 4) : 1;
|
||||
parameters.put("n", count);
|
||||
@ -177,15 +179,6 @@ public class DashScopeImageProvider implements ImageGenerationProvider {
|
||||
}
|
||||
}
|
||||
|
||||
private String aspectRatioToSize(String aspectRatio) {
|
||||
if (aspectRatio == null) return "1024*1024";
|
||||
return switch (aspectRatio) {
|
||||
case "16:9" -> "1280*720";
|
||||
case "9:16" -> "720*1280";
|
||||
default -> "1024*1024";
|
||||
};
|
||||
}
|
||||
|
||||
private String extractImageUrl(JsonNode output) {
|
||||
JsonNode results = output.path("results");
|
||||
if (results.isArray() && !results.isEmpty()) {
|
||||
|
||||
@ -89,8 +89,9 @@ public class FalImageProvider implements ImageGenerationProvider {
|
||||
ObjectNode body = objectMapper.createObjectNode();
|
||||
body.put("prompt", request.getPrompt());
|
||||
|
||||
// fal.ai 使用 image_size 对象或字符串
|
||||
String size = normalizeSize(request.getSize(), request.getAspectRatio());
|
||||
// request.size already normalized by ImageGenerationService to one of supportedSizes.
|
||||
// fal.ai expects image_size as a {width, height} object.
|
||||
String size = request.getSize();
|
||||
ObjectNode imageSize = body.putObject("image_size");
|
||||
String[] parts = size.split("x");
|
||||
imageSize.put("width", Integer.parseInt(parts[0]));
|
||||
@ -179,22 +180,6 @@ public class FalImageProvider implements ImageGenerationProvider {
|
||||
}
|
||||
}
|
||||
|
||||
private String normalizeSize(String size, String aspectRatio) {
|
||||
if (size != null && !size.isBlank()) {
|
||||
return size;
|
||||
}
|
||||
if (aspectRatio != null) {
|
||||
return switch (aspectRatio) {
|
||||
case "16:9" -> "1536x1024";
|
||||
case "9:16" -> "1024x1536";
|
||||
case "4:3" -> "1024x768";
|
||||
case "3:4" -> "768x1024";
|
||||
default -> "1024x1024";
|
||||
};
|
||||
}
|
||||
return "1024x1024";
|
||||
}
|
||||
|
||||
private String extractImageUrl(JsonNode result) {
|
||||
JsonNode images = result.path("images");
|
||||
if (images.isArray() && !images.isEmpty()) {
|
||||
|
||||
@ -94,8 +94,9 @@ public class ZhipuImageProvider implements ImageGenerationProvider {
|
||||
body.put("model", model);
|
||||
body.put("prompt", request.getPrompt());
|
||||
|
||||
String size = aspectRatioToSize(request.getAspectRatio());
|
||||
if (size != null) {
|
||||
// request.size already normalized by ImageGenerationService to one of supportedSizes.
|
||||
String size = request.getSize();
|
||||
if (size != null && !size.isBlank()) {
|
||||
body.put("size", size);
|
||||
}
|
||||
|
||||
@ -134,14 +135,4 @@ public class ZhipuImageProvider implements ImageGenerationProvider {
|
||||
}
|
||||
}
|
||||
|
||||
private String aspectRatioToSize(String aspectRatio) {
|
||||
if (aspectRatio == null) return "1024x1024";
|
||||
return switch (aspectRatio) {
|
||||
case "16:9" -> "1344x768";
|
||||
case "9:16" -> "768x1344";
|
||||
case "4:3" -> "1152x864";
|
||||
case "3:4" -> "864x1152";
|
||||
default -> "1024x1024";
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
@ -21,7 +21,7 @@ public class MusicGenerateTool {
|
||||
|
||||
private final MusicGenerationService musicGenerationService;
|
||||
|
||||
@Tool(description = "生成音乐或歌曲。支持文字描述生成音乐、歌词谱曲、纯音乐等模式。支持 Google Lyria 和 MiniMax Music 等 Provider。")
|
||||
@Tool(description = "生成音乐或歌曲。支持文字描述生成音乐、歌词谱曲、纯音乐等模式。支持 Google Lyria 和 MiniMax Music 等 Provider。任务异步执行（约 1-3 分钟），工具立即返回任务 ID，前端会在生成完成时自动接收 SSE 事件并把音频推到对话中。无需用户手动刷新。")
|
||||
public String music_generate(
|
||||
@ToolParam(description = "音乐风格/场景描述，如：'轻快的钢琴爵士乐'、'史诗电影配乐'、'欢快的流行歌曲'") String prompt,
|
||||
@ToolParam(description = "歌词文本（可选，不填则由 AI 生成或生成纯音乐）") String lyrics,
|
||||
@ -33,6 +33,7 @@ public class MusicGenerateTool {
|
||||
if (conversationId == null) {
|
||||
return "无法获取会话 ID";
|
||||
}
|
||||
String username = ToolExecutionContext.username(ctx);
|
||||
|
||||
MusicGenerationRequest request = MusicGenerationRequest.builder()
|
||||
.prompt(prompt)
|
||||
@ -40,15 +41,13 @@ public class MusicGenerateTool {
|
||||
.instrumental(instrumental != null ? instrumental : false)
|
||||
.build();
|
||||
|
||||
Map<String, Object> result = musicGenerationService.generate(conversationId, request);
|
||||
Map<String, Object> result = musicGenerationService.submitGeneration(
|
||||
conversationId, request, username);
|
||||
|
||||
if (Boolean.TRUE.equals(result.get("success"))) {
|
||||
StringBuilder sb = new StringBuilder("音乐生成完成！\n");
|
||||
sb.append("播放链接: ").append(result.get("audioUrl"));
|
||||
if (result.containsKey("lyrics") && result.get("lyrics") != null) {
|
||||
sb.append("\n\n歌词:\n").append(result.get("lyrics"));
|
||||
}
|
||||
return sb.toString();
|
||||
return "音乐生成任务已提交（taskId=" + result.get("taskId")
|
||||
+ ", provider=" + result.get("providerName")
|
||||
+ "）。生成需要约 1-2 分钟，完成后会自动推送到对话。";
|
||||
} else {
|
||||
return "音乐生成失败: " + result.get("error");
|
||||
}
|
||||
|
||||
@ -1,19 +1,42 @@
|
||||
package vip.mate.tool.music;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.fasterxml.jackson.databind.node.ObjectNode;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
import vip.mate.system.model.SystemSettingsDTO;
|
||||
import vip.mate.system.service.SystemSettingService;
|
||||
import vip.mate.task.AsyncTaskService;
|
||||
import vip.mate.task.model.AsyncTaskEntity;
|
||||
import vip.mate.workspace.conversation.ConversationService;
|
||||
import vip.mate.workspace.conversation.model.MessageContentPart;
|
||||
|
||||
import jakarta.annotation.PreDestroy;
|
||||
import java.io.IOException;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.Paths;
|
||||
import java.util.*;
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
|
||||
/**
|
||||
* 音乐生成服务
|
||||
* Music generation service.
|
||||
* <p>
|
||||
* Music providers (MiniMax, Lyria) are synchronous from the upstream API
|
||||
* perspective — a single HTTP call blocks for ~120s and returns the audio
|
||||
* bytes. Running that on the chat SSE thread freezes the conversation's
|
||||
* event stream for two minutes; instead we register an
|
||||
* {@link AsyncTaskEntity} for state + SSE bookkeeping and dispatch the actual
|
||||
* provider call to a dedicated virtual-thread worker. The chat thread returns
|
||||
* with a taskId immediately.
|
||||
*
|
||||
* @author MateClaw Team
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@ -22,47 +45,138 @@ public class MusicGenerationService {
|
||||
|
||||
private final SystemSettingService systemSettingService;
|
||||
private final MusicProviderRegistry providerRegistry;
|
||||
private final AsyncTaskService asyncTaskService;
|
||||
private final ConversationService conversationService;
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
private static final Path UPLOAD_ROOT = Paths.get("data", "chat-uploads");
|
||||
private static final String TASK_TYPE = "music_generation";
|
||||
|
||||
public Map<String, Object> generate(String conversationId, MusicGenerationRequest request) {
|
||||
/** Dedicated virtual-thread worker. Music generation blocks on a single
|
||||
* upstream HTTP call (~120s) — keeping it off the polling pool and the
|
||||
* chat SSE thread is the whole point of P0b. */
|
||||
private final ExecutorService musicWorker = Executors.newThreadPerTaskExecutor(
|
||||
Thread.ofVirtual().name("music-gen-", 0).factory());
|
||||
|
||||
/**
|
||||
* Submit a music generation request. Returns immediately with a taskId;
|
||||
* actual generation runs on {@link #musicWorker} and completion is
|
||||
* broadcast via {@link AsyncTaskService#broadcastTaskEvent}.
|
||||
*/
|
||||
public Map<String, Object> submitGeneration(String conversationId,
|
||||
MusicGenerationRequest request,
|
||||
String createdBy) {
|
||||
SystemSettingsDTO config = systemSettingService.getAllSettings();
|
||||
|
||||
if (!Boolean.TRUE.equals(config.getMusicEnabled())) {
|
||||
return Map.of("success", false, "error", "音乐生成功能未启用");
|
||||
}
|
||||
|
||||
MusicGenerationResult result = generateWithFallback(request, config);
|
||||
if (!result.isSuccess()) {
|
||||
return Map.of("success", false, "error", result.getErrorMessage());
|
||||
MusicGenerationProvider primary = providerRegistry.resolve(config);
|
||||
if (primary == null) {
|
||||
return Map.of("success", false, "error", "没有可用的音乐 Provider");
|
||||
}
|
||||
|
||||
AsyncTaskEntity task;
|
||||
try {
|
||||
String fileId = UUID.randomUUID().toString().replace("-", "").substring(0, 12);
|
||||
Path dir = UPLOAD_ROOT.resolve(conversationId);
|
||||
Files.createDirectories(dir);
|
||||
String fileName = "music_" + fileId + "." + result.getFormat();
|
||||
Path filePath = dir.resolve(fileName);
|
||||
Files.write(filePath, result.getAudioData());
|
||||
String requestJson = objectMapper.writeValueAsString(request);
|
||||
String localProviderTaskId = "music_" + UUID.randomUUID().toString()
|
||||
.replace("-", "").substring(0, 12);
|
||||
task = asyncTaskService.createTask(
|
||||
TASK_TYPE, conversationId, null,
|
||||
primary.id(),
|
||||
localProviderTaskId,
|
||||
requestJson, createdBy);
|
||||
} catch (Exception e) {
|
||||
log.error("[Music] Failed to create async task: {}", e.getMessage(), e);
|
||||
return Map.of("success", false, "error", "创建任务失败: " + e.getMessage());
|
||||
}
|
||||
|
||||
String audioUrl = "/api/v1/chat/files/" + conversationId + "/" + fileName;
|
||||
AsyncTaskEntity finalTask = task;
|
||||
musicWorker.execute(() -> doGenerate(finalTask, request, config, conversationId));
|
||||
|
||||
Map<String, Object> response = new LinkedHashMap<>();
|
||||
response.put("success", true);
|
||||
response.put("audioUrl", audioUrl);
|
||||
response.put("contentType", result.getContentType());
|
||||
response.put("format", result.getFormat());
|
||||
if (result.getLyrics() != null) {
|
||||
response.put("lyrics", result.getLyrics());
|
||||
Map<String, Object> response = new LinkedHashMap<>();
|
||||
response.put("success", true);
|
||||
response.put("taskId", task.getTaskId());
|
||||
response.put("providerName", primary.id());
|
||||
return response;
|
||||
}
|
||||
|
||||
// ==================== Worker thread ====================
|
||||
|
||||
private void doGenerate(AsyncTaskEntity task, MusicGenerationRequest request,
|
||||
SystemSettingsDTO config, String conversationId) {
|
||||
try {
|
||||
asyncTaskService.updateStatus(task.getTaskId(), "running", null, null, null);
|
||||
|
||||
MusicGenerationResult result = generateWithFallback(request, config);
|
||||
if (!result.isSuccess()) {
|
||||
asyncTaskService.updateStatus(task.getTaskId(), "failed", null, null,
|
||||
result.getErrorMessage());
|
||||
asyncTaskService.broadcastTaskEventWithData(task, "async_task_completed",
|
||||
false, Map.of(), result.getErrorMessage());
|
||||
return;
|
||||
}
|
||||
return response;
|
||||
} catch (IOException e) {
|
||||
log.error("[Music] Failed to save audio: {}", e.getMessage());
|
||||
return Map.of("success", false, "error", "音频文件保存失败");
|
||||
|
||||
String audioUrl = persistAudio(conversationId, task.getTaskId(), result);
|
||||
|
||||
saveAssistantMessage(conversationId, audioUrl, result);
|
||||
|
||||
ObjectNode resultJson = objectMapper.createObjectNode();
|
||||
resultJson.put("audioUrl", audioUrl);
|
||||
resultJson.put("format", result.getFormat());
|
||||
if (result.getLyrics() != null) {
|
||||
resultJson.put("lyrics", result.getLyrics());
|
||||
}
|
||||
asyncTaskService.updateStatus(task.getTaskId(), "succeeded", 100,
|
||||
resultJson.toString(), null);
|
||||
|
||||
Map<String, Object> extra = new LinkedHashMap<>();
|
||||
extra.put("audioUrl", audioUrl);
|
||||
extra.put("format", result.getFormat());
|
||||
if (result.getLyrics() != null) {
|
||||
extra.put("lyrics", result.getLyrics());
|
||||
}
|
||||
asyncTaskService.broadcastTaskEventWithData(task, "async_task_completed",
|
||||
true, extra, null);
|
||||
|
||||
log.info("[Music] Task {} succeeded, audio at {}", task.getTaskId(), audioUrl);
|
||||
} catch (Exception e) {
|
||||
log.error("[Music] Task {} worker failed: {}", task.getTaskId(), e.getMessage(), e);
|
||||
asyncTaskService.updateStatus(task.getTaskId(), "failed", null, null,
|
||||
"音乐生成异常: " + e.getMessage());
|
||||
asyncTaskService.broadcastTaskEventWithData(task, "async_task_completed",
|
||||
false, Map.of(), "音乐生成异常: " + e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private MusicGenerationResult generateWithFallback(MusicGenerationRequest request, SystemSettingsDTO config) {
|
||||
private String persistAudio(String conversationId, String taskId,
|
||||
MusicGenerationResult result) throws IOException {
|
||||
Path dir = UPLOAD_ROOT.resolve(conversationId);
|
||||
Files.createDirectories(dir);
|
||||
String fileName = "music_" + taskId + "." + result.getFormat();
|
||||
Path filePath = dir.resolve(fileName);
|
||||
Files.write(filePath, result.getAudioData());
|
||||
return "/api/v1/chat/files/" + conversationId + "/" + fileName;
|
||||
}
|
||||
|
||||
private void saveAssistantMessage(String conversationId, String audioUrl,
|
||||
MusicGenerationResult result) {
|
||||
MessageContentPart audioPart = MessageContentPart.audio(null,
|
||||
audioUrl.substring(audioUrl.lastIndexOf('/') + 1));
|
||||
audioPart.setFileUrl(audioUrl);
|
||||
audioPart.setContentType(result.getContentType());
|
||||
|
||||
StringBuilder content = new StringBuilder("音乐生成完成");
|
||||
if (result.getLyrics() != null && !result.getLyrics().isBlank()) {
|
||||
content.append("\n\n歌词:\n").append(result.getLyrics());
|
||||
}
|
||||
conversationService.saveMessage(conversationId, "assistant",
|
||||
content.toString(), List.of(audioPart), "completed");
|
||||
}
|
||||
|
||||
private MusicGenerationResult generateWithFallback(MusicGenerationRequest request,
|
||||
SystemSettingsDTO config) {
|
||||
MusicGenerationProvider primary = providerRegistry.resolve(config);
|
||||
if (primary == null) return MusicGenerationResult.failure("没有可用的音乐 Provider");
|
||||
|
||||
@ -74,6 +188,7 @@ public class MusicGenerationService {
|
||||
|
||||
if (Boolean.TRUE.equals(config.getMusicFallbackEnabled())) {
|
||||
for (MusicGenerationProvider fb : providerRegistry.fallbackCandidates(config, primary.id())) {
|
||||
log.info("[Music] Trying fallback provider: {}", fb.id());
|
||||
result = fb.generate(request, config);
|
||||
if (result.isSuccess()) return result;
|
||||
errors.add(fb.id() + ": " + result.getErrorMessage());
|
||||
@ -82,4 +197,9 @@ public class MusicGenerationService {
|
||||
|
||||
return MusicGenerationResult.failure("所有音乐 Provider 均失败\n" + String.join("\n", errors));
|
||||
}
|
||||
|
||||
@PreDestroy
|
||||
public void shutdown() {
|
||||
musicWorker.shutdown();
|
||||
}
|
||||
}
|
||||
|
||||
@ -7,19 +7,38 @@ import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.fasterxml.jackson.databind.node.ObjectNode;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.util.StringUtils;
|
||||
import vip.mate.llm.event.ModelConfigChangedEvent;
|
||||
import vip.mate.llm.model.ModelProviderEntity;
|
||||
import vip.mate.llm.service.ModelProviderService;
|
||||
import vip.mate.system.model.SystemSettingsDTO;
|
||||
import vip.mate.tool.music.MusicGenerationProvider;
|
||||
import vip.mate.tool.music.MusicGenerationRequest;
|
||||
import vip.mate.tool.music.MusicGenerationResult;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
/**
|
||||
* MiniMax 音乐生成 Provider — music-2.5+
|
||||
* MiniMax 音乐生成 Provider — music-2.5+.
|
||||
* <p>
|
||||
* 复用视频生成中的 MiniMax API Key。
|
||||
* Credential resolution order (first match wins):
|
||||
* <ol>
|
||||
* <li>{@code mate_model_provider} entry with id {@code minimax-cn} — China region
|
||||
* (api.minimaxi.com). The base URL declared for the LLM endpoint typically
|
||||
* ends in {@code /anthropic}; we strip that and append the music path.</li>
|
||||
* <li>{@code mate_model_provider} entry with id {@code minimax} — international
|
||||
* region (api.minimax.io).</li>
|
||||
* <li>Legacy {@code mate_system_setting.minimaxApiKey} — only kept for
|
||||
* backwards compatibility with the pre-RFC-202605 single-key setup; defaults
|
||||
* to the international endpoint.</li>
|
||||
* </ol>
|
||||
* The China and international MiniMax APIs use different keys and different hosts;
|
||||
* sourcing credentials from the unified LLM provider table avoids the prior bug
|
||||
* where users had to configure the same key in two places (and the music provider
|
||||
* silently used the wrong region).
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
@ -27,9 +46,21 @@ import java.util.List;
|
||||
public class MiniMaxMusicProvider implements MusicGenerationProvider {
|
||||
|
||||
private final ObjectMapper objectMapper;
|
||||
private final ModelProviderService modelProviderService;
|
||||
|
||||
private static final String BASE_URL = "https://api.minimax.io";
|
||||
private static final String DEFAULT_BASE_URL = "https://api.minimax.io";
|
||||
private static final String DEFAULT_MODEL = "music-2.5+";
|
||||
private static final String[] LLM_PROVIDER_IDS = {"minimax-cn", "minimax"};
|
||||
|
||||
/** Cache TTL — credentials rarely change at runtime; 60s is generous and
|
||||
* invalidated immediately on {@link ModelConfigChangedEvent}. Without this
|
||||
* every music task hit {@code SELECT mate_model_provider WHERE provider_id=?}
|
||||
* 4-6 times (registry probe + worker re-resolve + per-call resolve). */
|
||||
private static final long CACHE_TTL_NANOS = 60L * 1_000_000_000L;
|
||||
|
||||
private final AtomicReference<CachedCredentials> cache = new AtomicReference<>();
|
||||
|
||||
private record CachedCredentials(Credentials creds, long expiresAtNanos) {}
|
||||
|
||||
@Override public String id() { return "minimax"; }
|
||||
@Override public String label() { return "MiniMax Music"; }
|
||||
@ -40,13 +71,78 @@ public class MiniMaxMusicProvider implements MusicGenerationProvider {
|
||||
|
||||
@Override
|
||||
public boolean isAvailable(SystemSettingsDTO config) {
|
||||
return StringUtils.hasText(config.getMinimaxApiKey());
|
||||
return resolveCredentials(config) != null;
|
||||
}
|
||||
|
||||
private record Credentials(String apiKey, String baseUrl) {}
|
||||
|
||||
private Credentials resolveCredentials(SystemSettingsDTO config) {
|
||||
CachedCredentials hit = cache.get();
|
||||
long now = System.nanoTime();
|
||||
if (hit != null && now < hit.expiresAtNanos()) {
|
||||
return hit.creds();
|
||||
}
|
||||
Credentials fresh = resolveCredentialsUncached(config);
|
||||
cache.set(new CachedCredentials(fresh, now + CACHE_TTL_NANOS));
|
||||
return fresh;
|
||||
}
|
||||
|
||||
private Credentials resolveCredentialsUncached(SystemSettingsDTO config) {
|
||||
for (String providerId : LLM_PROVIDER_IDS) {
|
||||
try {
|
||||
if (!modelProviderService.isProviderConfigured(providerId)) {
|
||||
continue;
|
||||
}
|
||||
ModelProviderEntity p = modelProviderService.getProviderConfig(providerId);
|
||||
if (StringUtils.hasText(p.getApiKey())) {
|
||||
return new Credentials(p.getApiKey(), normalizeBaseUrl(p.getBaseUrl()));
|
||||
}
|
||||
} catch (Exception ignore) {
|
||||
// try next id
|
||||
}
|
||||
}
|
||||
String legacyKey = config.getMinimaxApiKey();
|
||||
if (StringUtils.hasText(legacyKey)) {
|
||||
return new Credentials(legacyKey, DEFAULT_BASE_URL);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/** Drop the credential cache when any provider config changes — keeps the
|
||||
* 60s TTL safe under the user-edits-key-mid-task scenario. */
|
||||
@EventListener
|
||||
public void onModelConfigChanged(ModelConfigChangedEvent event) {
|
||||
cache.set(null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Reduce any configured base URL to its scheme+host origin (e.g.
|
||||
* {@code https://api.minimaxi.com/anthropic} → {@code https://api.minimaxi.com}).
|
||||
* The music API lives at {@code /v1/music_generation} regardless of which
|
||||
* upstream path the LLM client uses.
|
||||
*/
|
||||
private static String normalizeBaseUrl(String baseUrl) {
|
||||
if (!StringUtils.hasText(baseUrl)) return DEFAULT_BASE_URL;
|
||||
try {
|
||||
java.net.URI uri = java.net.URI.create(baseUrl.trim());
|
||||
if (uri.getScheme() == null || uri.getAuthority() == null) {
|
||||
return DEFAULT_BASE_URL;
|
||||
}
|
||||
return uri.getScheme() + "://" + uri.getAuthority();
|
||||
} catch (IllegalArgumentException e) {
|
||||
return DEFAULT_BASE_URL;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public MusicGenerationResult generate(MusicGenerationRequest request, SystemSettingsDTO config) {
|
||||
try {
|
||||
String apiKey = config.getMinimaxApiKey();
|
||||
Credentials creds = resolveCredentials(config);
|
||||
if (creds == null) {
|
||||
return MusicGenerationResult.failure("MiniMax 凭据未配置（在「模型与凭据」中配置 minimax-cn 或 minimax）");
|
||||
}
|
||||
String apiKey = creds.apiKey();
|
||||
String baseUrl = creds.baseUrl();
|
||||
String model = request.getModel() != null ? request.getModel() : DEFAULT_MODEL;
|
||||
|
||||
ObjectNode body = objectMapper.createObjectNode();
|
||||
@ -75,11 +171,11 @@ public class MiniMaxMusicProvider implements MusicGenerationProvider {
|
||||
audioSetting.put("bitrate", 256000);
|
||||
audioSetting.put("format", "mp3");
|
||||
|
||||
HttpResponse response = HttpRequest.post(BASE_URL + "/v1/music_generation")
|
||||
HttpResponse response = HttpRequest.post(baseUrl + "/v1/music_generation")
|
||||
.header("Authorization", "Bearer " + apiKey)
|
||||
.header("Content-Type", "application/json")
|
||||
.body(body.toString())
|
||||
.timeout(120_000)
|
||||
.timeout(300_000)
|
||||
.execute();
|
||||
|
||||
JsonNode result = objectMapper.readTree(response.body());
|
||||
@ -91,27 +187,42 @@ public class MiniMaxMusicProvider implements MusicGenerationProvider {
|
||||
return MusicGenerationResult.failure(errMsg);
|
||||
}
|
||||
|
||||
// 提取音频 URL 或内联数据
|
||||
String audioUrl = result.has("audio_url") ? result.get("audio_url").asText(null)
|
||||
: result.path("data").path("audio_url").asText(null);
|
||||
String lyrics = result.has("lyrics") ? result.get("lyrics").asText(null)
|
||||
: result.path("data").path("lyrics").asText(null);
|
||||
// Response shape varies between regions:
|
||||
// - Some responses put the CDN URL in {audio_url} or {data.audio_url}.
|
||||
// - Others put it in {audio} or {data.audio} (the field nominally for
|
||||
// inline hex/base64 binary). Earlier versions of this code treated
|
||||
// {data.audio} as inline data unconditionally and base64-decoded a
|
||||
// URL, throwing "Illegal base64 character 3a" on the ':' in https://.
|
||||
// Fix: resolve a single audioCandidate, then route to URL download or
|
||||
// binary decode based on whether it parses as http(s) URL.
|
||||
String audioCandidate = firstNonBlank(
|
||||
nullableText(result, "audio"),
|
||||
nullableText(result.path("data"), "audio"));
|
||||
String audioUrl = firstNonBlank(
|
||||
nullableText(result, "audio_url"),
|
||||
nullableText(result.path("data"), "audio_url"));
|
||||
if (audioUrl == null && isLikelyRemoteUrl(audioCandidate)) {
|
||||
audioUrl = audioCandidate;
|
||||
}
|
||||
String inlineAudio = isLikelyRemoteUrl(audioCandidate) ? null : audioCandidate;
|
||||
String lyrics = firstNonBlank(
|
||||
nullableText(result, "lyrics"),
|
||||
nullableText(result.path("data"), "lyrics"));
|
||||
|
||||
if (StringUtils.hasText(audioUrl)) {
|
||||
// 下载音频
|
||||
if (audioUrl != null) {
|
||||
byte[] audioData = HttpRequest.get(audioUrl).timeout(30_000).execute().bodyBytes();
|
||||
log.info("[MiniMax Music] Generated {} bytes audio (model={})", audioData.length, model);
|
||||
log.info("[MiniMax Music] Generated {} bytes audio via URL (model={})", audioData.length, model);
|
||||
return MusicGenerationResult.successWithLyrics(audioData, "audio/mpeg", "mp3", lyrics);
|
||||
}
|
||||
|
||||
// 尝试 inline audio
|
||||
String inlineAudio = result.has("audio") ? result.get("audio").asText(null)
|
||||
: result.path("data").path("audio").asText(null);
|
||||
if (StringUtils.hasText(inlineAudio)) {
|
||||
byte[] audioData = decodeAudio(inlineAudio);
|
||||
log.info("[MiniMax Music] Generated {} bytes audio inline (model={})", audioData.length, model);
|
||||
return MusicGenerationResult.successWithLyrics(audioData, "audio/mpeg", "mp3", lyrics);
|
||||
}
|
||||
|
||||
log.warn("[MiniMax Music] Response missing audio output. Body keys: {}",
|
||||
iteratorToString(result.fieldNames()));
|
||||
return MusicGenerationResult.failure("MiniMax 未返回音频数据");
|
||||
|
||||
} catch (Exception e) {
|
||||
@ -120,12 +231,43 @@ public class MiniMaxMusicProvider implements MusicGenerationProvider {
|
||||
}
|
||||
}
|
||||
|
||||
private byte[] decodeAudio(String data) {
|
||||
// MiniMax 可能返回 hex 或 base64
|
||||
if (data.matches("^[0-9a-fA-F]+$") && data.length() % 2 == 0) {
|
||||
return hexToBytes(data);
|
||||
private static String nullableText(JsonNode node, String field) {
|
||||
if (node == null || !node.has(field)) return null;
|
||||
JsonNode v = node.get(field);
|
||||
if (v == null || v.isNull()) return null;
|
||||
String s = v.asText(null);
|
||||
return (s == null || s.isBlank()) ? null : s.trim();
|
||||
}
|
||||
|
||||
private static String firstNonBlank(String... values) {
|
||||
for (String v : values) {
|
||||
if (v != null && !v.isBlank()) return v;
|
||||
}
|
||||
return java.util.Base64.getDecoder().decode(data);
|
||||
return null;
|
||||
}
|
||||
|
||||
private static boolean isLikelyRemoteUrl(String value) {
|
||||
if (value == null || value.isBlank()) return false;
|
||||
String trimmed = value.trim();
|
||||
return trimmed.regionMatches(true, 0, "http://", 0, 7)
|
||||
|| trimmed.regionMatches(true, 0, "https://", 0, 8);
|
||||
}
|
||||
|
||||
private static String iteratorToString(java.util.Iterator<String> it) {
|
||||
StringBuilder sb = new StringBuilder("[");
|
||||
while (it.hasNext()) {
|
||||
sb.append(it.next());
|
||||
if (it.hasNext()) sb.append(',');
|
||||
}
|
||||
return sb.append(']').toString();
|
||||
}
|
||||
|
||||
private byte[] decodeAudio(String data) {
|
||||
String trimmed = data.trim();
|
||||
if (trimmed.matches("^[0-9a-fA-F]+$") && trimmed.length() % 2 == 0) {
|
||||
return hexToBytes(trimmed);
|
||||
}
|
||||
return java.util.Base64.getDecoder().decode(trimmed);
|
||||
}
|
||||
|
||||
private byte[] hexToBytes(String hex) {
|
||||
|
||||
@ -33,7 +33,9 @@ public class VideoGenerationResult {
|
||||
.providerName(providerName)
|
||||
.status("submitted")
|
||||
.submitted(true)
|
||||
.message("视频生成任务已提交（任务 ID: " + taskId + "）。预计 1-5 分钟完成，完成后会自动显示在对话中。")
|
||||
// Format MUST keep `taskId=...` so the frontend reconnect detector
|
||||
// (useChat.ts TASK_ID_PATTERN) can extract it from the tool result.
|
||||
.message("视频生成任务已提交（taskId=" + taskId + ", provider=" + providerName + "）。预计 1-5 分钟完成，完成后会自动显示在对话中。")
|
||||
.build();
|
||||
}
|
||||
|
||||
|
||||
@ -73,6 +73,11 @@ tool.edit_file.error.edit_exception=\u7f16\u8f91\u6587\u4ef6\u5f02\u5e38: {0}
|
||||
tool.shell.error.timeout=\u547d\u4ee4\u6267\u884c\u8d85\u65f6\uff08{0}\u79d2\uff09\uff0c\u5df2\u5f3a\u5236\u7ec8\u6b62
|
||||
tool.shell.error.exception=\u6267\u884c\u5f02\u5e38: {0}
|
||||
|
||||
# --- Generative tools ---
|
||||
tool.image_generate.desc=\u751f\u6210\u56fe\u7247\u3002\u6839\u636e\u6587\u5b57\u63cf\u8ff0\u521b\u4f5c\u56fe\u50cf\uff0c\u652f\u6301 DashScope\u3001OpenAI\u3001fal.ai\u3001\u667a\u8c31 CogView \u7b49 Provider\u3002\u4efb\u52a1\u5f02\u6b65\u6267\u884c\uff0c\u5b8c\u6210\u540e\u524d\u7aef\u4f1a\u81ea\u52a8\u63a5\u6536 SSE \u4e8b\u4ef6 async_task_completed \u5e76\u628a\u56fe\u7247\u63a8\u5230\u5bf9\u8bdd\u4e2d\uff0c\u65e0\u9700\u624b\u52a8\u5237\u65b0\u3002
|
||||
tool.video_generate.desc=\u751f\u6210\u89c6\u9891\u3002\u652f\u6301\u6587\u751f\u89c6\u9891\u3001\u56fe\u751f\u89c6\u9891\u7b49\u6a21\u5f0f\uff0c\u652f\u6301 DashScope\u3001\u667a\u8c31 CogVideo\u3001Kling\u3001Runway\u3001MiniMax \u3001fal.ai \u7b49 Provider\u3002\u4efb\u52a1\u5f02\u6b65\u6267\u884c\uff08\u7ea6 1-5 \u5206\u949f\uff09\uff0c\u5b8c\u6210\u540e\u524d\u7aef\u4f1a\u81ea\u52a8\u63a5\u6536 SSE \u4e8b\u4ef6 async_task_completed \u5e76\u628a\u89c6\u9891\u63a8\u5230\u5bf9\u8bdd\u4e2d\uff0c\u65e0\u9700\u624b\u52a8\u5237\u65b0\u3002
|
||||
tool.music_generate.desc=\u751f\u6210\u97f3\u4e50\u6216\u6b4c\u66f2\u3002\u652f\u6301\u6587\u5b57\u63cf\u8ff0\u751f\u6210\u97f3\u4e50\u3001\u6b4c\u8bcd\u8c31\u66f2\u3001\u7eaf\u97f3\u4e50\u7b49\u6a21\u5f0f\uff0c\u652f\u6301 Google Lyria \u548c MiniMax Music \u7b49 Provider\u3002\u4efb\u52a1\u5f02\u6b65\u6267\u884c\uff08\u7ea6 1-3 \u5206\u949f\uff09\uff0c\u5b8c\u6210\u540e\u524d\u7aef\u4f1a\u81ea\u52a8\u63a5\u6536 SSE \u4e8b\u4ef6 async_task_completed \u5e76\u628a\u97f3\u9891\u63a8\u5230\u5bf9\u8bdd\u4e2d\uff0c\u65e0\u9700\u624b\u52a8\u5237\u65b0\u3002
|
||||
|
||||
# --- Guard Rules ---
|
||||
guard.SHELL_RM_RF_ROOT.name=\u9012\u5f52\u5f3a\u5236\u5220\u9664\u6839\u76ee\u5f55
|
||||
guard.SHELL_RM_RF_ROOT.fix=\u8bf7\u6307\u5b9a\u5177\u4f53\u76ee\u5f55\u8def\u5f84\u800c\u975e\u6839\u76ee\u5f55
|
||||
|
||||
@ -73,6 +73,11 @@ tool.edit_file.error.edit_exception=Edit file exception: {0}
|
||||
tool.shell.error.timeout=Command timed out ({0} seconds), forcefully terminated
|
||||
tool.shell.error.exception=Execution exception: {0}
|
||||
|
||||
# --- Generative tools ---
|
||||
tool.image_generate.desc=Generate an image from a text description. Supports DashScope, OpenAI, fal.ai, Zhipu CogView, etc. Runs asynchronously; the client receives the image automatically via the SSE async_task_completed event without a manual refresh.
|
||||
tool.video_generate.desc=Generate a video. Supports text-to-video and image-to-video modes via DashScope, Zhipu CogVideo, Kling, Runway, MiniMax, fal.ai, etc. Runs asynchronously (1-5 minutes); the client receives the video automatically via the SSE async_task_completed event without a manual refresh.
|
||||
tool.music_generate.desc=Generate music or a song. Supports text-to-music, lyrics composition, and instrumental modes via Google Lyria and MiniMax Music. Runs asynchronously (1-3 minutes); the client receives the audio automatically via the SSE async_task_completed event without a manual refresh.
|
||||
|
||||
# --- Guard Rules ---
|
||||
guard.SHELL_RM_RF_ROOT.name=Recursive force delete root directory
|
||||
guard.SHELL_RM_RF_ROOT.fix=Specify a concrete directory path instead of root
|
||||
|
||||
@ -196,6 +196,18 @@
|
||||
/>
|
||||
<span class="message-attachment-video__name">{{ attachment.name }}</span>
|
||||
</div>
|
||||
<div
|
||||
v-for="attachment in audioAttachments"
|
||||
:key="'audio-' + attachment.storedName"
|
||||
class="message-attachment-audio"
|
||||
>
|
||||
<audio
|
||||
:src="getDisplayUrl(attachment)"
|
||||
controls
|
||||
preload="metadata"
|
||||
/>
|
||||
<span class="message-attachment-audio__name">{{ attachment.name }}</span>
|
||||
</div>
|
||||
<button
|
||||
v-for="attachment in fileAttachments"
|
||||
:key="attachment.storedName"
|
||||
@ -295,7 +307,7 @@ import type { ChatErrorInfo } from '@/types/chatError'
|
||||
const { renderMarkdown } = useMarkdownRenderer()
|
||||
const { t } = useI18n()
|
||||
const { getToolLabel } = useToolLabel()
|
||||
const { blobUrls, loadAllImages, loadAllVideos, downloadFile, openImage, getDisplayUrl, revokeAll } = useAuthenticatedAttachment()
|
||||
const { blobUrls, loadAllImages, loadAllVideos, loadAllAudios, downloadFile, openImage, getDisplayUrl, revokeAll } = useAuthenticatedAttachment()
|
||||
|
||||
interface Props {
|
||||
message: Message
|
||||
@ -545,20 +557,65 @@ onBeforeUnmount(() => {
|
||||
})
|
||||
|
||||
// --- 附件 ---
|
||||
const attachments = computed(() => props.message.attachments || [])
|
||||
// MessageContentPart media (image/audio/video produced by generation tools) live
|
||||
// in `contentParts` rather than `attachments`. Synthesize virtual attachment
|
||||
// entries so the existing render + auth-blob loader works for them too.
|
||||
//
|
||||
// Dedup against `props.message.attachments` by URL — user-uploaded images often
|
||||
// land in BOTH lists (the upload endpoint registers them as ChatAttachment AND
|
||||
// the message persistence echoes them back as a `type: 'image'` MessageContentPart).
|
||||
// Without this guard each user image shows twice in the bubble.
|
||||
const mediaPartAttachments = computed<ChatAttachment[]>(() => {
|
||||
const parts = (props.message as any).contentParts as Array<any> | undefined
|
||||
if (!parts || !parts.length) return []
|
||||
const existingUrls = new Set(
|
||||
(props.message.attachments || []).map(a => a.url).filter(Boolean)
|
||||
)
|
||||
const out: ChatAttachment[] = []
|
||||
const seen = new Set<string>()
|
||||
for (const p of parts) {
|
||||
if (!p || !p.fileUrl) continue
|
||||
if (p.type !== 'image' && p.type !== 'audio' && p.type !== 'video') continue
|
||||
if (existingUrls.has(p.fileUrl) || seen.has(p.fileUrl)) continue
|
||||
seen.add(p.fileUrl)
|
||||
const fileName = p.fileName || p.fileUrl.split('/').pop() || `${p.type}-${out.length}`
|
||||
const ct = p.contentType
|
||||
|| (p.type === 'image' ? 'image/png' : p.type === 'audio' ? 'audio/mpeg' : 'video/mp4')
|
||||
out.push({
|
||||
name: fileName,
|
||||
size: 0,
|
||||
url: p.fileUrl,
|
||||
storedName: fileName,
|
||||
path: p.fileUrl,
|
||||
contentType: ct,
|
||||
})
|
||||
}
|
||||
return out
|
||||
})
|
||||
|
||||
const attachments = computed(() => [
|
||||
...(props.message.attachments || []),
|
||||
...mediaPartAttachments.value,
|
||||
])
|
||||
const imageAttachments = computed(() => attachments.value.filter(a => a.contentType?.startsWith('image/')))
|
||||
const videoAttachments = computed(() => attachments.value.filter(a => a.contentType?.startsWith('video/')))
|
||||
const audioAttachments = computed(() => attachments.value.filter(a => a.contentType?.startsWith('audio/')))
|
||||
const fileAttachments = computed(() => attachments.value.filter(a =>
|
||||
!a.contentType?.startsWith('image/') && !a.contentType?.startsWith('video/')
|
||||
!a.contentType?.startsWith('image/')
|
||||
&& !a.contentType?.startsWith('video/')
|
||||
&& !a.contentType?.startsWith('audio/')
|
||||
))
|
||||
|
||||
// 增量加载图片/视频附件的鉴权 blob URL（watch 覆盖首次 + 后续变化）
|
||||
// 增量加载图片/视频/音频附件的鉴权 blob URL（watch 覆盖首次 + 后续变化）
|
||||
watch(imageAttachments, (atts) => {
|
||||
if (atts.length > 0) loadAllImages(atts)
|
||||
}, { immediate: true })
|
||||
watch(videoAttachments, (atts) => {
|
||||
if (atts.length > 0) loadAllVideos(atts)
|
||||
}, { immediate: true })
|
||||
watch(audioAttachments, (atts) => {
|
||||
if (atts.length > 0) loadAllAudios(atts)
|
||||
}, { immediate: true })
|
||||
|
||||
// --- 时间 ---
|
||||
const formattedTime = computed(() => {
|
||||
@ -1374,6 +1431,27 @@ watch(isGenerating, (generating) => {
|
||||
white-space: nowrap;
|
||||
}
|
||||
|
||||
.message-attachment-audio {
|
||||
border-radius: 12px;
|
||||
overflow: hidden;
|
||||
}
|
||||
|
||||
.message-attachment-audio audio {
|
||||
width: 100%;
|
||||
max-width: 400px;
|
||||
display: block;
|
||||
}
|
||||
|
||||
.message-attachment-audio__name {
|
||||
display: block;
|
||||
margin-top: 4px;
|
||||
font-size: 12px;
|
||||
opacity: 0.76;
|
||||
overflow: hidden;
|
||||
text-overflow: ellipsis;
|
||||
white-space: nowrap;
|
||||
}
|
||||
|
||||
.message-attachment {
|
||||
display: flex;
|
||||
align-items: center;
|
||||
|
||||
@ -255,6 +255,38 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
||||
headers: streamHeaders,
|
||||
})
|
||||
|
||||
// ===== Async-task lifecycle bridge =====
|
||||
// Generative tools (music / video / image) return a taskId synchronously and
|
||||
// finish asynchronously via `async_task_completed`. If the upstream provider
|
||||
// is slow (MiniMax music ~2-3 min) the agent's reasoning turn finishes long
|
||||
// before the audio is ready, the SSE stream emits `done`, and any later
|
||||
// `async_task_completed` event lands on a closed emitter. We track which
|
||||
// taskIds are still pending here, and when `done` fires with non-empty set
|
||||
// we re-attach to the same conversation's stream so buffered + future async
|
||||
// events can flow through. ChatStreamTracker.attach was extended (RFC P0)
|
||||
// to keep the new emitter subscribed even when state.done=true.
|
||||
const pendingAsyncTaskIds = new Set<string>()
|
||||
// 16-hex taskId emitted by AsyncTaskService.createTask. Tool result text
|
||||
// varies — `taskId=xxx` is the canonical form (music/video/image), but
|
||||
// earlier video/image versions used 中文「任务 ID: xxx」 and old strings may
|
||||
// still flow through if the LLM cached them. Match both defensively.
|
||||
const TASK_ID_PATTERNS: RegExp[] = [
|
||||
/taskId[=:"\s]+([a-f0-9]{16})/i,
|
||||
/任务\s*ID[=:"\s]+([a-f0-9]{16})/i,
|
||||
/task[_\s]*id[=:"\s]+([a-f0-9]{16})/i,
|
||||
]
|
||||
const ASYNC_TOOL_NAMES = new Set(['music_generate', 'video_generate', 'image_generate'])
|
||||
let reconnectingForAsyncTasks = false
|
||||
|
||||
function extractTaskId(result: unknown): string | null {
|
||||
if (typeof result !== 'string') return null
|
||||
for (const re of TASK_ID_PATTERNS) {
|
||||
const m = result.match(re)
|
||||
if (m) return m[1]
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
// ===== SSE event handlers =====
|
||||
|
||||
stream.on('content_delta', (data) => {
|
||||
@ -449,6 +481,28 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
||||
persisted: data.persisted,
|
||||
messageCount: data.messageCount,
|
||||
})
|
||||
|
||||
// Re-attach SSE if any generative task is still in flight, so the eventual
|
||||
// async_task_completed event reaches us live (otherwise the user has to
|
||||
// refresh). Skip if we're already in a reconnect cycle, or for non-completed
|
||||
// terminal statuses where reconnect is misleading.
|
||||
const reconnectableStatus = !data.status
|
||||
|| data.status === 'completed'
|
||||
|| data.status === 'idle'
|
||||
if (reconnectableStatus
|
||||
&& !reconnectingForAsyncTasks
|
||||
&& pendingAsyncTaskIds.size > 0
|
||||
&& streamConversationId) {
|
||||
const targetConv = streamConversationId
|
||||
reconnectingForAsyncTasks = true
|
||||
// Defer one tick so the current 'done' handler chain finishes before
|
||||
// disconnect() fires inside connect().
|
||||
setTimeout(() => {
|
||||
stream.connect({ conversationId: targetConv, reconnect: true })
|
||||
.catch(() => { /* swallow — handled by stream.error event */ })
|
||||
.finally(() => { reconnectingForAsyncTasks = false })
|
||||
}, 50)
|
||||
}
|
||||
})
|
||||
|
||||
let errorFired = false
|
||||
@ -579,6 +633,14 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
||||
}
|
||||
flushSegmentsToMessage()
|
||||
}
|
||||
|
||||
// Track async generative tools whose taskId is in the result string.
|
||||
if (data.success !== false && ASYNC_TOOL_NAMES.has(data.toolName)) {
|
||||
const taskId = extractTaskId(data.result)
|
||||
if (taskId) {
|
||||
pendingAsyncTaskIds.add(taskId)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
// ===== Browser action events =====
|
||||
@ -1080,50 +1142,78 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
||||
phaseInfo.value = null
|
||||
})
|
||||
|
||||
// ===== Async task completion events (video generation, image generation, etc.) =====
|
||||
// ===== Async task completion events (video / image / music generation) =====
|
||||
stream.on('async_task_completed', (data) => {
|
||||
if (isStaleEvent(data)) return
|
||||
if (data.success && streamConversationId) {
|
||||
let mediaPart: MessageContentPart | null = null
|
||||
if (data.videoUrl) {
|
||||
mediaPart = {
|
||||
type: 'video',
|
||||
fileUrl: data.videoUrl,
|
||||
fileName: `video_${data.taskId}.mp4`,
|
||||
contentType: 'video/mp4',
|
||||
} as MessageContentPart
|
||||
} else if (data.imageUrl) {
|
||||
mediaPart = {
|
||||
type: 'image',
|
||||
fileUrl: data.imageUrl,
|
||||
fileName: `image_${data.taskId}.png`,
|
||||
contentType: 'image/png',
|
||||
} as MessageContentPart
|
||||
}
|
||||
if (!streamConversationId) return
|
||||
|
||||
if (!mediaPart) return
|
||||
// Mark this taskId resolved so done-after-pending reconnect logic
|
||||
// doesn't keep the SSE alive longer than needed.
|
||||
if (data.taskId) pendingAsyncTaskIds.delete(data.taskId)
|
||||
|
||||
// Prefer appending to the current assistant message (avoids image appearing above the text reply)
|
||||
if (currentAssistantId.value) {
|
||||
const msg = getMessage(currentAssistantId.value)
|
||||
if (msg) {
|
||||
const existingParts = (msg as any).contentParts || []
|
||||
updateMessage(currentAssistantId.value, {
|
||||
contentParts: [...existingParts, mediaPart],
|
||||
} as any)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Fallback: agent already finished — create a standalone message
|
||||
// Failure path — surface error so user knows the task is over.
|
||||
if (!data.success) {
|
||||
const taskLabel = data.taskType === 'music_generation' ? '音乐'
|
||||
: data.taskType === 'video_generation' ? '视频'
|
||||
: data.taskType === 'image_generation' ? '图片'
|
||||
: '任务'
|
||||
addMessage({
|
||||
role: 'assistant',
|
||||
content: '',
|
||||
contentParts: [mediaPart],
|
||||
content: `${taskLabel}生成失败: ${data.errorMessage || '未知错误'}`,
|
||||
contentParts: [],
|
||||
status: 'completed',
|
||||
conversationId: streamConversationId,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
let mediaPart: MessageContentPart | null = null
|
||||
if (data.videoUrl) {
|
||||
mediaPart = {
|
||||
type: 'video',
|
||||
fileUrl: data.videoUrl,
|
||||
fileName: `video_${data.taskId}.mp4`,
|
||||
contentType: 'video/mp4',
|
||||
} as MessageContentPart
|
||||
} else if (data.imageUrl) {
|
||||
mediaPart = {
|
||||
type: 'image',
|
||||
fileUrl: data.imageUrl,
|
||||
fileName: `image_${data.taskId}.png`,
|
||||
contentType: 'image/png',
|
||||
} as MessageContentPart
|
||||
} else if (data.audioUrl) {
|
||||
const fmt = (data.format || 'mp3') as string
|
||||
mediaPart = {
|
||||
type: 'audio',
|
||||
fileUrl: data.audioUrl,
|
||||
fileName: `music_${data.taskId}.${fmt}`,
|
||||
contentType: fmt === 'wav' ? 'audio/wav' : 'audio/mpeg',
|
||||
} as MessageContentPart
|
||||
}
|
||||
|
||||
if (!mediaPart) return
|
||||
|
||||
// Prefer appending to the current assistant message (avoids media appearing above the text reply)
|
||||
if (currentAssistantId.value) {
|
||||
const msg = getMessage(currentAssistantId.value)
|
||||
if (msg) {
|
||||
const existingParts = (msg as any).contentParts || []
|
||||
updateMessage(currentAssistantId.value, {
|
||||
contentParts: [...existingParts, mediaPart],
|
||||
} as any)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Fallback: agent already finished — create a standalone message
|
||||
addMessage({
|
||||
role: 'assistant',
|
||||
content: '',
|
||||
contentParts: [mediaPart],
|
||||
status: 'completed',
|
||||
conversationId: streamConversationId,
|
||||
})
|
||||
})
|
||||
|
||||
// ===== Auto TTS =====
|
||||
|
||||
@ -71,6 +71,19 @@ export function useAuthenticatedAttachment() {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 批量加载所有音频附件的 blob URL（<audio :src> 同样不带 Authorization 头）
|
||||
*/
|
||||
async function loadAllAudios(attachments: ChatAttachment[]) {
|
||||
const audioAtts = attachments.filter(a => a.contentType?.startsWith('audio/'))
|
||||
for (const att of audioAtts) {
|
||||
const key = att.storedName || att.url
|
||||
if (!att.previewUrl && att.url && !blobUrls.value[key]) {
|
||||
await loadBlobUrl(att.url, key)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 鉴权下载文件：fetch blob → 创建临时 <a download> → 触发点击
|
||||
*/
|
||||
@ -140,6 +153,7 @@ export function useAuthenticatedAttachment() {
|
||||
loadBlobUrl,
|
||||
loadAllImages,
|
||||
loadAllVideos,
|
||||
loadAllAudios,
|
||||
downloadFile,
|
||||
openImage,
|
||||
getDisplayUrl,
|
||||
|
||||
Loading…
Reference in New Issue
Block a user