代码提交

This commit is contained in:
董宏凯 committed 2026-08-24 20:47:31 +08:00
1 parent 007ed5a7de
commit babc5e95c6
21 files changed
+1170 -145

No files matched your search

@@ -362,6 +362,11 @@ export interface StreamMultiAgentResult {
* an empty `dispatched` as consensus.
*/
clarification: ClarificationPayload | null;
/**
* 席位本轮无可见正文(仅思考 / 断流 / 空串)。此时 `content` 为空,调用方不得把
* 流式过程文本当成已交付,也不得写入 priorDeliveries。
*/
invalidDelivery?: { reason: string } | null;
}
/**
@@ -564,6 +569,7 @@ export async function streamMultiAgent(
// for diagnosing why a sub-agent stream "looks empty".
let lastEvent = "";
let frameSeq = 0;
let gotFinalStatus = false;
const debugLabel = `[stream:${req.agentType}:${req.agentName}]`;
// Stream 帧日志(frame# / delta)仅 opt-in 开启。默认关闭,避免 Step 2 流式期间
@@ -580,6 +586,7 @@ export async function streamMultiAgent(
agentName: null,
clarification: null,
modelUsed: null,
invalidDelivery: null,
};
// Inferred SSE phase + a tiny helper so we only fire onPhaseChange when the
@@ -833,6 +840,7 @@ export async function streamMultiAgent(
}
if (isFinalStatusFrame(parsed)) {
gotFinalStatus = true;
if (Array.isArray(parsed.status)) {
result.dispatched = parsed.status;
} else {
@@ -857,8 +865,15 @@ export async function streamMultiAgent(
),
};
}
if (parsed.status === "invalid_delivery") {
const reasonRaw = (parsed as Record<string, unknown>).reason;
result.invalidDelivery = {
reason: typeof reasonRaw === "string" && reasonRaw ? reasonRaw : "empty",
};
result.content = "";
}
}
if (typeof parsed.content === "string") {
if (typeof parsed.content === "string" && !result.invalidDelivery) {
result.content = parsed.content;
}
if (typeof parsed.agent_name === "string") {
@@ -896,6 +911,10 @@ export async function streamMultiAgent(
if (debugEnabled) {
console.debug(debugLabel, `stream ended after ${frameSeq} frames; final content length=${result.content.length}`);
}
if (req.agentType === "special" && !gotFinalStatus) {
result.invalidDelivery = { reason: "truncated" };
result.content = "";
}
return result;
} catch (err) {
if (!reported && req.signal?.aborted !== true && (err as Error)?.name !== "AbortError") {
@@ -27,6 +27,7 @@ import {
type StreamMultiAgentRequest,
type ThreadIdMap,
} from "@/roundtable-planning/api/multi-agent";
import { isValidSeatDelivery, stripReasoning } from "@/roundtable-planning/lib/reasoning";
import { uploadFiles } from "@/core/uploads/api";
import { useUpdateSubtask } from "@/core/tasks/context";
@@ -47,6 +48,35 @@ function takeOnce<T>(ref: { current: T | null }): T | undefined {
return v;
}
const PARALLEL_JUDGE_NOTE =
"\n\n【派活方式】派活前请先判断各席位子任务之间的依赖关系:相互独立的子任务请在" +
"**同一轮**通过 agent_orchestration 一次性派给多个席位(系统会并行执行、显著提速);" +
"有先后依赖的子任务则每轮只派一个席位,等其交付后再派下一个。";
const INVALID_DELIVERY_CONFIDENCE = "交付无效(仅思考或中断)";
function isInvalidSeatStream(res: { invalidDelivery?: { reason: string } | null; content?: string | null }): boolean {
return !!res.invalidDelivery || !isValidSeatDelivery(res.content);
}
function buildLeaderContinueTask(invalidNames: string[]): string {
if (invalidNames.length > 0) {
return (
`以下席位本轮未产出有效正文(仅思考或中断,思考过程不算交付):${invalidNames.join("、")}。` +
"这些席位**不算已交付**。请通过 agent_orchestration **重新派活**给它们," +
"拿到可见正文后再判断是否收口;严禁声称它们已交付,也不得现在输出最终方案。" +
PARALLEL_JUDGE_NOTE
);
}
return (
"各子智能体已就上一轮派活完成交付,请你结合最新结论判断:" +
"① 若仍有关键信息缺口或席位职责未覆盖 → 继续派活(按上述派活方式判断串/并行);" +
"② 若信息已足够 → **停止派活**,直接综合各席位交付输出最终方案,并在结论最开头用一行标注" +
"「【已达成共识】(置信度:高/中/低)」。不要在信息不足时草率收口,也不要在已足够时反复派活。" +
PARALLEL_JUDGE_NOTE
);
}
/**
* 从一帧里识别子智能体 custom 事件(task_started/running/completed/failed/cancelled/
* timed_out)。custom 流帧通常是裸 payload 对象 `{type:"task_*", task_id, ...}`;
@@ -1804,10 +1834,6 @@ export function useStep2Orchestration(
// 串/并行交给**总控判断**(2026-06-13):各席位子任务相互独立 → 同一轮一次性
// 派给多个席位(前端并行执行,单列可写显示);有先后依赖 → 每轮只派一个。
const PARALLEL_JUDGE_NOTE =
"\n\n【派活方式】派活前请先判断各席位子任务之间的依赖关系:相互独立的子任务请在" +
"**同一轮**通过 agent_orchestration 一次性派给多个席位(系统会并行执行、显著提速);" +
"有先后依赖的子任务则每轮只派一个席位,等其交付后再派下一个。";
// 优化1+2(2026-06-13):每个席位任务都拼上「直接产出别提问 + 承接前序 + 结构化交付模板」。
// 提升交付质量,并让 Step3 三产物(尤其大屏 seats.sections)的提炼有稳定结构可循。
const SEAT_DELIVERY_GUIDE =
@@ -1829,7 +1855,7 @@ export function useStep2Orchestration(
bubbleId: number;
seatThreadId: string | undefined;
isParallel: boolean;
}): Promise<{ content: string; aborted: boolean }> => {
}): Promise<{ content: string; aborted: boolean; invalid: boolean }> => {
const meta = agentNameToRoleRef.current[a.agentId.toLowerCase()];
const perSeatModel = meta ? agentConfigsRef.current[meta.speakerId]?.model : undefined;
const fullTask = a.task + SEAT_DELIVERY_GUIDE;
@@ -1869,15 +1895,27 @@ export function useStep2Orchestration(
activeAbortersRef.current.delete(subAborter);
if (isStale()) {
finalizeDialogue(a.bubbleId, specRes.content || "(已停止)", "已停止");
return { content: "", aborted: true };
return { content: "", aborted: true, invalid: false };
}
if (isInvalidSeatStream(specRes)) {
finalizeDialogue(
a.bubbleId,
specRes.content || "(无有效正文)",
INVALID_DELIVERY_CONFIDENCE,
);
return { content: "", aborted: false, invalid: true };
}
finalizeDialogue(a.bubbleId, specRes.content, "子智能体交付");
return { content: specRes.content ?? "", aborted: false };
return {
content: stripReasoning(specRes.content) || specRes.content,
aborted: false,
invalid: false,
};
} catch (err) {
activeAbortersRef.current.delete(subAborter);
if (subAborter.signal.aborted || (err as Error).name === "AbortError") {
finalizeDialogue(a.bubbleId, "(已停止)", "已停止");
return { content: "", aborted: true };
return { content: "", aborted: true, invalid: false };
}
if (attempt === 0) {
// 第一次失败 → 清掉半截内容,自动重试一次。
@@ -1889,10 +1927,10 @@ export function useStep2Orchestration(
}
console.error(`[special:${a.agentId}] error:`, err);
finalizeDialogue(a.bubbleId, `(错误) ${(err as Error).message}`, "调用失败");
return { content: "", aborted: false };
return { content: "", aborted: false, invalid: true };
}
}
return { content: "", aborted: false };
return { content: "", aborted: false, invalid: true };
};
let nextLeaderTask = seedMessage + PARALLEL_JUDGE_NOTE;
@@ -2007,6 +2045,7 @@ export function useStep2Orchestration(
}
const dispatched = leaderRes.dispatched as LeaderDispatch[];
const invalidNames: string[] = [];
if (dispatched.length > 1) {
// ── 总控一轮派给多个席位 = 它判定子任务相互独立 → **批内并行**执行
// (2026-06-13)。显示走「业务链 sandbox 单列可写」同款:头像条 + 单列
@@ -2060,7 +2099,7 @@ export function useStep2Orchestration(
markBatchStatus(agentId, "done");
if (meta) markSpeakerIdle(meta.speakerId);
parallelStreamRef.current.delete(bubbleId);
return r.content.trim() ? { name: seatName, content: r.content } : null;
return { seatName, r };
}),
);
@@ -2069,17 +2108,21 @@ export function useStep2Orchestration(
setSelectedParallelSeats([]);
resetParallelStreams();
recommendParallelRef.current = false;
// 累积本批交付(供后续轮席位 read_peer_delivery 取全文)。
for (const r of batchResults) if (r) priorDeliveries.push(r);
// 累积本批**有效**交付(思考-only / 中断不算,供后续轮 read_peer_delivery)。
for (const { seatName, r } of batchResults) {
if (r.invalid) invalidNames.push(seatName);
else if (r.content.trim()) priorDeliveries.push({ name: seatName, content: r.content });
}
} else {
// **顺序**派活(单席位 / 总控判定有依赖)—— 不要改回 Promise.all,详见 §5.4。
for (const [subName, task] of dispatched) {
if (isStale()) break;
const meta = agentNameToRoleRef.current[subName.toLowerCase()];
const seatName = meta?.displayName ?? subName;
if (meta) markSpeakerStreaming(meta.speakerId);
const seatThreadId = resolveSeatThreadId(ids, subName);
const subBubbleId = appendStreamingDialogue({
sender: meta?.displayName ?? subName,
sender: seatName,
avatarType: meta?.avatarType ?? "blue",
initialContent: "",
confidence: "调度中",
@@ -2087,14 +2130,15 @@ export function useStep2Orchestration(
});
const r = await runSeatDelivery({
agentId: subName,
displayName: meta?.displayName ?? subName,
displayName: seatName,
task,
bubbleId: subBubbleId,
seatThreadId,
isParallel: false,
});
if (meta) markSpeakerIdle(meta.speakerId);
if (r.content.trim()) priorDeliveries.push({ name: meta?.displayName ?? subName, content: r.content });
if (r.invalid) invalidNames.push(seatName);
else if (r.content.trim()) priorDeliveries.push({ name: seatName, content: r.content });
if (r.aborted) break;
}
}
@@ -2105,14 +2149,8 @@ export function useStep2Orchestration(
Math.min(95, parseFloat((prev + 8).toFixed(1))),
);
// 优化3:续轮强化收敛指引——信息足够就**明确收口**,结论开头标注【已达成共识】,
// 减少弱模型该收口时还乱派 / 该继续时草草收尾的两类偏差。
nextLeaderTask =
"各子智能体已就上一轮派活完成交付,请你结合最新结论判断:" +
"① 若仍有关键信息缺口或席位职责未覆盖 → 继续派活(按上述派活方式判断串/并行);" +
"② 若信息已足够 → **停止派活**,直接综合各席位交付输出最终方案,并在结论最开头用一行标注" +
"「【已达成共识】(置信度:高/中/低)」。不要在信息不足时草率收口,也不要在已足够时反复派活。" +
PARALLEL_JUDGE_NOTE;
// 续轮:有无效席位则强制重派;全部有效才允许判断收口。
nextLeaderTask = buildLeaderContinueTask(invalidNames);
}
if (safetyCounter >= MAX_CYCLES) {
@@ -2178,6 +2216,8 @@ export function useStep2Orchestration(
}
const leaderThreadId = resolveSeatThreadId(ids, coord);
let finishedAllSeats = true;
const chainInvalidNames: string[] = [];
const retryByIndex: Record<number, number> = {};
try {
for (let i = startSeatIndex; i < total; i += 1) {
@@ -2288,6 +2328,7 @@ export function useStep2Orchestration(
const subAborter = new AbortController();
activeAbortersRef.current.add(subAborter);
let aborted = false;
let seatInvalid = false;
try {
const specRes = await streamMultiAgent({
agentType: "special",
@@ -2322,6 +2363,13 @@ export function useStep2Orchestration(
if (isStale()) {
finalizeDialogue(subBubbleId, specRes.content || "(已停止)", "已停止");
aborted = true;
} else if (isInvalidSeatStream(specRes)) {
finalizeDialogue(
subBubbleId,
specRes.content || "(无有效正文)",
INVALID_DELIVERY_CONFIDENCE,
);
seatInvalid = true;
} else {
finalizeDialogue(subBubbleId, specRes.content, "子智能体交付");
}
@@ -2332,6 +2380,7 @@ export function useStep2Orchestration(
} else {
console.error(`[chain-special:${seat.agent_id}] error:`, err);
finalizeDialogue(subBubbleId, `(错误) ${(err as Error).message}`, "调用失败");
seatInvalid = true;
}
} finally {
activeAbortersRef.current.delete(subAborter);
@@ -2341,6 +2390,21 @@ export function useStep2Orchestration(
finishedAllSeats = false;
break;
}
if (seatInvalid) {
const tries = retryByIndex[i] ?? 0;
if (tries < 2) {
retryByIndex[i] = tries + 1;
i -= 1;
continue;
}
if (!chainInvalidNames.includes(seatName)) chainInvalidNames.push(seatName);
finishedAllSeats = false;
if (humanValidationsRef.current[seat.agent_id]) {
pauseForHumanValidation({ mode: "chain", nextSeatIndex: i }, [seatName]);
break;
}
continue;
}
setConsensusPercentage(Math.min(95, Math.round(((i + 1) / total) * 90)));
if (humanValidationsRef.current[seat.agent_id]) {
@@ -2353,6 +2417,14 @@ export function useStep2Orchestration(
}
}
// 有席位仅思考/中断:禁止链条收口,交给总控重新派活。
if (!isStale() && chainInvalidNames.length > 0) {
isOrchestratingRef.current = false;
setIsOrchestrating(false);
await runOrchestration(buildLeaderContinueTask(chainInvalidNames), ids, coord);
return;
}
// ── (c) 总控收口:综合全链交付,产出最终结论 ──
if (!isStale() && finishedAllSeats && total > 0) {
highlightSpeaker(null);
@@ -2430,6 +2502,7 @@ export function useStep2Orchestration(
ingestSubagentTaskFrame,
resolveSeatThreadId,
pauseForHumanValidation,
runOrchestration,
],
);
@@ -2480,6 +2553,7 @@ export function useStep2Orchestration(
humanValidationContinuationRef.current = null;
}
let finishedAllStages = true;
const dagInvalidNames: string[] = [];
// 跨 stage 累积的**完整交付** {name, content}[] —— 注入下一 stage 席位的 run context
// (peerDeliveries),供 read_peer_delivery 工具按需取全文(§12.9 传摘要按需传全文)。
// 研讨前并行取数(gatherFirst)的产物作为初始注入:首个 stage 席位开局即带共享资料。
@@ -2639,7 +2713,7 @@ export function useStep2Orchestration(
// 每个席位独立 try/catch 兜底,保证**单席位失败不拖垮整 stage**(§9):
// 任务函数永不 reject → Promise.all 不会因一个席位异常整体崩。
const stageResults = await Promise.all(
seatPrep.map(async ({ agentId, meta, seatName, seatThreadId, bubbleId }): Promise<{ name: string; content: string } | null> => {
seatPrep.map(async ({ agentId, meta, seatName, seatThreadId, bubbleId }): Promise<{ seatName: string; invalid: boolean; content: string }> => {
const perSeatModel = meta
? agentConfigsRef.current[meta.speakerId]?.model
: undefined;
@@ -2698,11 +2772,23 @@ export function useStep2Orchestration(
if (isStale()) {
finalizeDialogue(bubbleId, specRes.content || "(已停止)", "已停止");
markBatchStatus(agentId, "done");
return null;
return { seatName, invalid: false, content: "" };
}
if (isInvalidSeatStream(specRes)) {
finalizeDialogue(
bubbleId,
specRes.content || "(无有效正文)",
INVALID_DELIVERY_CONFIDENCE,
);
markBatchStatus(agentId, "done");
return { seatName, invalid: true, content: "" };
}
finalizeDialogue(bubbleId, specRes.content, "子智能体交付");
markBatchStatus(agentId, "done");
return specRes.content?.trim() ? { name: seatName, content: specRes.content } : null;
const visible = stripReasoning(specRes.content);
return visible.trim()
? { seatName, invalid: false, content: visible }
: { seatName, invalid: true, content: "" };
} catch (err) {
if (subAborter.signal.aborted || (err as Error).name === "AbortError") {
finalizeDialogue(bubbleId, "(已停止)", "已停止");
@@ -2712,7 +2798,7 @@ export function useStep2Orchestration(
finalizeDialogue(bubbleId, `(错误) ${(err as Error).message}`, "调用失败");
markBatchStatus(agentId, "error");
}
return null;
return { seatName, invalid: true, content: "" };
} finally {
activeAbortersRef.current.delete(subAborter);
if (meta) markSpeakerIdle(meta.speakerId);
@@ -2738,9 +2824,11 @@ export function useStep2Orchestration(
finishedAllStages = false;
break;
}
// 累积本 stage 的**完整交付**,供后续 stage 席位 read_peer_delivery 按需取全文(§12.9)。
// 累积本 stage 的**有效**交付(思考-only / 中断不算)。
for (const r of stageResults) {
if (r && r.content?.trim()) priorDeliveries.push(r);
if (!r) continue;
if (r.invalid) dagInvalidNames.push(r.seatName);
else if (r.content.trim()) priorDeliveries.push({ name: r.seatName, content: r.content });
}
setConsensusPercentage(
Math.min(95, Math.round(((s + 1) / plan.stages.length) * 90)),
@@ -2772,7 +2860,9 @@ export function useStep2Orchestration(
isOrchestratingRef.current = false; // 释放锁,让 runOrchestration 即刻接管
setIsOrchestrating(false);
await runOrchestration(
"各阶段席位已按依赖编排完成交付。请你作为总控,综合各席位的交付:若仍需补充则继续派活,否则直接输出最终方案并收口。",
dagInvalidNames.length > 0
? buildLeaderContinueTask(dagInvalidNames)
: "各阶段席位已按依赖编排完成交付。请你作为总控,综合各席位的交付:若仍需补充则继续派活,否则直接输出最终方案并收口。",
ids,
coord,
);
@@ -2896,7 +2986,9 @@ export function useStep2Orchestration(
markBatchStatus(agentId, "done");
if (meta) markSpeakerIdle(meta.speakerId);
parallelStreamRef.current.delete(bubbleId);
return specRes.content?.trim() ? { name: seatName, content: specRes.content } : null;
if (isInvalidSeatStream(specRes)) return null;
const visible = stripReasoning(specRes.content);
return visible.trim() ? { name: seatName, content: visible } : null;
} catch (err) {
activeAbortersRef.current.delete(subAborter);
// 取数失败不阻断研讨:该席位无资料即可,研讨阶段它仍可自行检索。
@@ -3068,7 +3160,7 @@ export function useStep2Orchestration(
}
void runActiveOrchestration(
"用户已恢复研讨,请继续上一轮的协调工作。",
"用户已恢复研讨,请继续上一轮的协调工作。请以系统核对的席位状态(✅ 已交付 / ⬜ 未交付 / ⚠️ 仅思考或中断)为准:思考过程与中断输出都不算交付,⬜ 与 ⚠️ 席位必须重新派活拿到可见正文后才能收口。",
threadIdsRef.current,
coordinatorNameRef.current,
);
@@ -1,7 +1,7 @@
import assert from "node:assert/strict";
import test from "node:test";
import { splitInlineReasoning, stripReasoning } from "./reasoning";
import { isValidSeatDelivery, splitInlineReasoning, stripReasoning } from "./reasoning";
test("空输入返回空正文与 null 推理", () => {
assert.deepEqual(splitInlineReasoning(""), { content: "", reasoning: null });
@@ -59,3 +59,13 @@ test("stripReasoning 只返回正文", () => {
assert.equal(stripReasoning("<think>推理</think>交付正文"), "交付正文");
assert.equal(stripReasoning("前导推理</think>交付正文"), "交付正文");
});
test("isValidSeatDelivery:剥思考后有正文才算交付", () => {
assert.equal(isValidSeatDelivery("可见结论"), true);
assert.equal(isValidSeatDelivery("<think>推理</think>可见结论"), true);
assert.equal(isValidSeatDelivery("<think>只思考没有正文</think>"), false);
assert.equal(isValidSeatDelivery("前导推理</think>"), false);
assert.equal(isValidSeatDelivery(""), false);
assert.equal(isValidSeatDelivery(null), false);
assert.equal(isValidSeatDelivery(" "), false);
});
@@ -80,3 +80,8 @@ export function splitInlineReasoning(text: string): SplitReasoningResult {
export function stripReasoning(text: string): string {
return splitInlineReasoning(text).content;
}
/** 席位是否产出有效交付:剥掉思考后仍有可见正文。思考-only / 空串都不算。 */
export function isValidSeatDelivery(text: string | null | undefined): boolean {
return stripReasoning(text ?? "").trim().length > 0;
}