feat(memory): P1 长期记忆升级 —— 异步攒批 Consolidate + 软删 + importance/last_seen #1

Merged
Blizzard merged 181 commits from feat/wails3 into main 2026-07-17 01:12:32 +00:00
10 changed files with 188 additions and 51 deletions
Showing only changes of commit c2811e79a3 - Show all commits
@@ -260,9 +260,11 @@ interface Cue {
} }
// 角色层:走位全在这里。桌子不跟着走,所以单独画在上层。 // 角色层:走位全在这里。桌子不跟着走,所以单独画在上层。
function Actor({ cue, reduced }: { cue: Cue; reduced: boolean }) { function Actor({ cue, reduced, hasLead }: { cue: Cue; reduced: boolean; hasLead: boolean }) {
const { seat, x, y, scale: s, index } = cue; const { seat, x, y, scale: s, index } = cue;
const done = seat.status === "done"; const finished = seat.status === "done";
// 没有协调者就没人可递:干完仍然眯眼笑、打✓,只是不演交付那一程。
const handoff = finished && hasLead;
// 入场:从门口走到工位(局部坐标 = 场景增量 ÷ 本工位缩放)。 // 入场:从门口走到工位(局部坐标 = 场景增量 ÷ 本工位缩放)。
const enterVars: Vars = { const enterVars: Vars = {
@@ -294,11 +296,11 @@ function Actor({ cue, reduced }: { cue: Cue; reduced: boolean }) {
hair={HAIRS[index % HAIRS.length]} hair={HAIRS[index % HAIRS.length]}
style={index % 3} style={index % 3}
typing={typing} typing={typing}
happy={done} happy={finished}
sad={seat.status === "error"} sad={seat.status === "error"}
delay={index * 0.23} delay={index * 0.23}
/> />
{done && !reduced && <Paper dur={HANDOFF_MS} delay={cue.handoffAt} />} {handoff && !reduced && <Paper dur={HANDOFF_MS} delay={cue.handoffAt} />}
{cue.bubble && <Bubble status={seat.status} delay={cue.bubbleAt} />} {cue.bubble && <Bubble status={seat.status} delay={cue.bubbleAt} />}
</> </>
); );
@@ -317,7 +319,7 @@ function Actor({ cue, reduced }: { cue: Cue; reduced: boolean }) {
<g transform={`translate(${x}, ${y}) scale(${s})`}> <g transform={`translate(${x}, ${y}) scale(${s})`}>
<g className="sdx-enter" style={enterVars}> <g className="sdx-enter" style={enterVars}>
<g className="sdx-walk" style={{ "--wcount": Math.round(ENTER_MS / 300), animationDelay: `${cue.enterAt}s` } as Vars}> <g className="sdx-walk" style={{ "--wcount": Math.round(ENTER_MS / 300), animationDelay: `${cue.enterAt}s` } as Vars}>
{done ? ( {handoff ? (
<g className="sdx-handoff" style={handoffVars}> <g className="sdx-handoff" style={handoffVars}>
<g className="sdx-walk" style={{ "--wcount": Math.round(HANDOFF_MS / 300), animationDelay: `${cue.handoffAt}s` } as Vars}> <g className="sdx-walk" style={{ "--wcount": Math.round(HANDOFF_MS / 300), animationDelay: `${cue.handoffAt}s` } as Vars}>
<ellipse cx={0} cy={2} rx={46} ry={9} fill="#000" opacity={0.09} /> <ellipse cx={0} cy={2} rx={46} ry={9} fill="#000" opacity={0.09} />
@@ -411,11 +413,12 @@ export function OfficeView({ events, output, phase }: { events: ExecEvent[]; out
typeAt: typeAtMs / 1000, typeAt: typeAtMs / 1000,
typeCount: streaming ? 0 : Math.max(1, Math.round((workEndMs - typeAtMs) / TYPE_MS)), typeCount: streaming ? 0 : Math.max(1, Math.round((workEndMs - typeAtMs) / TYPE_MS)),
handoffAt: workEndMs / 1000, handoffAt: workEndMs / 1000,
// ✓ 气泡等交付走完再冒;没有协调者就不演交付,收工即冒。
bubbleAt: bubbleAt:
(streaming (streaming
? 0 ? 0
: seat.status === "done" : seat.status === "done"
? workEndMs + HANDOFF_MS ? workEndMs + (lead ? HANDOFF_MS : 0)
: seat.status === "error" : seat.status === "error"
? workEndMs ? workEndMs
: typeAtMs) / 1000, : typeAtMs) / 1000,
@@ -478,7 +481,17 @@ export function OfficeView({ events, output, phase }: { events: ExecEvent[]; out
<foreignObject x={188} y={26} width={296} height={74}> <foreignObject x={188} y={26} width={296} height={74}>
<div style={{ fontFamily: "system-ui, sans-serif", color: "#4a4238", lineHeight: 1.35 }}> <div style={{ fontFamily: "system-ui, sans-serif", color: "#4a4238", lineHeight: 1.35 }}>
<div style={{ fontSize: 12, color: leadTone, fontWeight: 600 }}> <div style={{ fontSize: 12, color: leadTone, fontWeight: 600 }}>
{streaming ? "综合中…" : lead?.status === "error" ? "协调失败" : lead?.status === "done" ? "已综合" : "协调中"} {lead
? streaming
? "综合中…"
: lead.status === "error"
? "协调失败"
: lead.status === "done"
? "已综合"
: "协调中"
: streaming
? "生成中…"
: "任务产出"}
</div> </div>
<div <div
style={{ style={{
@@ -498,23 +511,26 @@ export function OfficeView({ events, output, phase }: { events: ExecEvent[]; out
</foreignObject> </foreignObject>
</g> </g>
{/* 协调者 */} {/* 协调者:只有真走了协调者节点才画。用户自己并排多个 agent 的图没有协调者,
<g transform={`translate(${LEAD_X}, ${LEAD_Y}) scale(${LEAD_SCALE})`}> 那就不该凭空画一个人出来——白板照样挂产出,专家各自干各自的。 */}
<ellipse cx={0} cy={2} rx={46} ry={9} fill="#000" opacity={0.09} /> {lead && (
<Chibi <g transform={`translate(${LEAD_X}, ${LEAD_Y}) scale(${LEAD_SCALE})`}>
shirt={RUN} <ellipse cx={0} cy={2} rx={46} ry={9} fill="#000" opacity={0.09} />
hair="#3f3229" <Chibi
style={2} shirt={RUN}
typing={streaming ? { count: Infinity, delay: 0 } : undefined} hair="#3f3229"
happy={lead?.status === "done"} style={2}
sad={lead?.status === "error"} typing={streaming ? { count: Infinity, delay: 0 } : undefined}
delay={0} happy={lead.status === "done"}
/> sad={lead.status === "error"}
<Deskette name={lead?.label ?? "协调者"} sub="协调者" tone={leadTone} glow={streaming} w={230} /> delay={0}
</g> />
<Deskette name={lead.label} sub="协调者" tone={leadTone} glow={streaming} w={230} />
</g>
)}
{/* 派发连线:只画正在工作的工位 */} {/* 派发连线:只画正在工作的工位;没有协调者就没有派发这回事 */}
{cues {(lead ? cues : [])
.filter((c) => c.seat.status === "running") .filter((c) => c.seat.status === "running")
.map((c) => ( .map((c) => (
<path <path
@@ -530,7 +546,7 @@ export function OfficeView({ events, output, phase }: { events: ExecEvent[]; out
{/* 角色层在前、桌子层在后:角色才能走到别人桌前而不被自己那张桌子焊死 */} {/* 角色层在前、桌子层在后:角色才能走到别人桌前而不被自己那张桌子焊死 */}
{cues.map((c) => ( {cues.map((c) => (
<Actor key={c.seat.node} cue={c} reduced={reduced} /> <Actor key={c.seat.node} cue={c} reduced={reduced} hasLead={!!lead} />
))} ))}
{cues.map((c) => ( {cues.map((c) => (
<g key={`d-${c.seat.node}`} transform={`translate(${c.x}, ${c.y}) scale(${c.scale})`}> <g key={`d-${c.seat.node}`} transform={`translate(${c.x}, ${c.y}) scale(${c.scale})`}>
@@ -110,12 +110,23 @@ export function TeamView({ events, output, phase }: { events: ExecEvent[]; outpu
<foreignObject x={CX - HUB_W / 2} y={CY - HUB_H / 2} width={HUB_W} height={HUB_H}> <foreignObject x={CX - HUB_W / 2} y={CY - HUB_H / 2} width={HUB_W} height={HUB_H}>
<div className="h-full rounded-xl border border-brand/40 bg-ink-850 px-3 py-2.5"> <div className="h-full rounded-xl border border-brand/40 bg-ink-850 px-3 py-2.5">
<div className="flex items-center gap-1.5 text-[11px] text-slate-500"> <div className="flex items-center gap-1.5 text-[11px] text-slate-500">
<Network className="h-3.5 w-3.5" strokeWidth={1.8} /> <Network className="h-3.5 w-3.5" strokeWidth={1.8} />
{lead ? "协调者" : "任务产出"}
</div> </div>
<div className="mt-1 truncate text-[13px] font-medium text-slate-100">{lead?.label ?? "多智能体协调"}</div> <div className="mt-1 truncate text-[13px] font-medium text-slate-100">{lead?.label ?? "多智能体任务"}</div>
<div className="mt-1.5 border-t border-line pt-1.5"> <div className="mt-1.5 border-t border-line pt-1.5">
<div className={cn("text-[11px]", streaming ? "text-brand-400" : "text-slate-500")}> <div className={cn("text-[11px]", streaming ? "text-brand-400" : "text-slate-500")}>
{streaming ? "综合中…" : lead?.status === "error" ? "协调失败" : lead?.status === "done" ? "已综合" : "协调中"} {lead
? streaming
? "综合中…"
: lead.status === "error"
? "协调失败"
: lead.status === "done"
? "已综合"
: "协调中"
: streaming
? "生成中…"
: "已产出"}
</div> </div>
<div className="mt-0.5 line-clamp-2 text-[10px] leading-snug text-slate-500"> <div className="mt-0.5 line-clamp-2 text-[10px] leading-snug text-slate-500">
{thinking || lead?.detail || ""} {thinking || lead?.detail || ""}
+34 -1
View File
@@ -86,13 +86,26 @@ describe("pendingApprovalHITL 待审批中断)", () => {
// 后端约定:协调者 coordinator:<id>/kind=model;专家 agent:<名>/kind=agent // 后端约定:协调者 coordinator:<id>/kind=model;专家 agent:<名>/kind=agent
// 收尾 detail="简报 X → Y";工具 tool:<名>/kind=tool。 // 收尾 detail="简报 X → Y";工具 tool:<名>/kind=tool。
describe("isMultiAgent是否走了多智能体协调", () => { describe("isMultiAgent值不值得看团队", () => {
it("有 coordinator: 节点 → true", () => { it("有 coordinator: 节点 → true", () => {
expect(isMultiAgent([ev("coordinator:c", "start", { kind: "model" })])).toBe(true); expect(isMultiAgent([ev("coordinator:c", "start", { kind: "model" })])).toBe(true);
}); });
it("普通任务(model/tool) → false,不显示团队 tab", () => { it("普通任务(model/tool) → false,不显示团队 tab", () => {
expect(isMultiAgent([ev("model", "start", { kind: "model" }), ev("tool:wiki", "end", { kind: "tool" })])).toBe(false); expect(isMultiAgent([ev("model", "start", { kind: "model" }), ev("tool:wiki", "end", { kind: "tool" })])).toBe(false);
}); });
// 编排里并排多个 agent 的图没有协调者,且 agent 节点发的是 kind=model(不是 kind=agent)。
// 按 kind 判会把这种图整个漏掉——用户从编排点执行后就看不到团队 tab。
it("图里 ≥2 个 agent 节点(kind=model,无协调者)→ true", () => {
expect(
isMultiAgent([
ev("agent:a", "start", { kind: "model" }),
ev("agent:b", "start", { kind: "model" }),
]),
).toBe(true);
});
it("只有 1 个 agent → false(一个人不叫团队)", () => {
expect(isMultiAgent([ev("agent:a", "start", { kind: "model" }), ev("agent:a", "end", { kind: "model" })])).toBe(false);
});
}); });
describe("deriveTeam(协调者 + 工位)", () => { describe("deriveTeam(协调者 + 工位)", () => {
@@ -134,6 +147,26 @@ describe("deriveTeam(协调者 + 工位)", () => {
expect(t.seats[0].output).toBe("命中 4 段"); expect(t.seats[0].output).toBe("命中 4 段");
}); });
it("图里的 agent 节点(kind=model)也算工位", () => {
const t = deriveTeam([
ev("agent:writer", "start", { kind: "model", ts: 0 }),
ev("agent:writer", "end", { kind: "model", ts: 300, ms: 300, detail: "成稿 800 字" }),
]);
expect(t.seats.map((s) => s.name)).toEqual(["writer"]);
expect(t.seats[0].status).toBe("done");
expect(t.lead).toBeNull(); // 没有协调者 → 不能凭空造一个
});
it("retriever/map/render 不当工位(同为 kind=tool,但不是 tool: 前缀)", () => {
const t = deriveTeam([
ev("retriever:r", "start", { kind: "tool", ts: 0 }),
ev("retriever:r", "end", { kind: "tool", ts: 50, ms: 50 }),
ev("map:m", "start", { kind: "plan", ts: 10 }),
ev("render:o", "end", { kind: "render", ts: 60 }),
]);
expect(t.seats).toEqual([]);
});
it("协调者被识别为 lead,且不混进工位", () => { it("协调者被识别为 lead,且不混进工位", () => {
const t = deriveTeam([ const t = deriveTeam([
ev("coordinator:c", "start", { kind: "model", label: "多智能体协调", detail: "2 个专家可派发" }), ev("coordinator:c", "start", { kind: "model", label: "多智能体协调", detail: "2 个专家可派发" }),
+13 -4
View File
@@ -128,9 +128,14 @@ export interface TeamModel {
t1: number; // 最晚结束(未完则为 now t1: number; // 最晚结束(未完则为 now
} }
// isMultiAgent 判定这次运行是否走了多智能体协调(据此才显示「团队」视图)。 // isMultiAgent 判定这次运行值不值得看「团队」:
// 1) 走了协调者(coordinator:)——orchestrator 自主派发;或
// 2) 图里挂了 ≥2 个专家(agent:)——用户自己在编排里并排/接力多个 agent,那也是个团队。
// 只有 1 个 agent 的普通任务不显示(一个人不叫团队,看轨迹就够)。
export function isMultiAgent(events: ExecEvent[]): boolean { export function isMultiAgent(events: ExecEvent[]): boolean {
return events.some((e) => e.node.startsWith("coordinator:")); if (events.some((e) => e.node.startsWith("coordinator:"))) return true;
const agents = new Set(events.filter((e) => e.node.startsWith("agent:")).map((e) => e.node));
return agents.size >= 2;
} }
// splitBrief 解析专家收尾 detail"简报 X → Y" → {brief:X, output:Y};不匹配则整段当产出。 // splitBrief 解析专家收尾 detail"简报 X → Y" → {brief:X, output:Y};不匹配则整段当产出。
@@ -164,8 +169,12 @@ export function deriveTeam(events: ExecEvent[], now = Date.now()): TeamModel {
else if (e.phase === "error") lead.status = "error"; else if (e.phase === "error") lead.status = "error";
continue; continue;
} }
const isAgent = e.kind === "agent"; // 按节点名前缀挑工位,不按 kind:两条产生 agent 的路径 kind 并不一致
const isTool = e.kind === "tool"; // —— 协调者派发的专家是 kind=agent,图里的 agent 节点是 kind=model。
// 前缀才是后端一直遵守的约定(agent:<名字> / tool:<名字>),
// 而且它天然把 retriever:/map:/render: 这些同为 kind=tool 的节点挡在工位之外。
const isAgent = e.node.startsWith("agent:");
const isTool = e.node.startsWith("tool:");
if (!isAgent && !isTool) continue; if (!isAgent && !isTool) continue;
let s = map.get(e.node); let s = map.get(e.node);
if (!s) { if (!s) {
@@ -66,7 +66,7 @@ func (o *Orchestrator) execComposeGraph(ctx context.Context, t *contract.Task, t
tr.info("task", "system", "无结构化图", "按单轮对话执行(compose") tr.info("task", "system", "无结构化图", "按单轮对话执行(compose")
b.profile = o.fetchMemory(ctx, b.uid, b.query) b.profile = o.fetchMemory(ctx, b.uid, b.query)
b.history = o.fetchHistory(ctx, b.sid) b.history = o.fetchHistory(ctx, b.sid)
o.runComposeConversation(ctx, t.ID, b, plan.System, tr, "agent") o.runComposeConversation(ctx, t.ID, b, plan.System, tr, "agent", "模型流式推理")
return b.answer, refsOf(b), b.fatalErr // 模型失败 → 上抛判 failed(对齐 graph.go return b.answer, refsOf(b), b.fatalErr // 模型失败 → 上抛判 failed(对齐 graph.go
} }
@@ -211,7 +211,7 @@ func (o *Orchestrator) execComposeGraph(ctx context.Context, t *contract.Task, t
if cerr != nil { if cerr != nil {
// 编译失败(罕见)→ 降级为单轮对话兜底(自研 graph.go 已退役,不再回退)。 // 编译失败(罕见)→ 降级为单轮对话兜底(自研 graph.go 已退役,不再回退)。
tr.info("task", "system", "compose 编译失败", "降级单轮对话:"+cerr.Error()) tr.info("task", "system", "compose 编译失败", "降级单轮对话:"+cerr.Error())
o.runComposeConversation(ctx, t.ID, b, plan.System, tr, "agent") o.runComposeConversation(ctx, t.ID, b, plan.System, tr, "agent", "模型流式推理")
return b.answer, refsOf(b), b.fatalErr return b.answer, refsOf(b), b.fatalErr
} }
// checkpoint id = task id:审批中断时 compose 据此把整图状态(含 board)落进 store。 // checkpoint id = task id:审批中断时 compose 据此把整图状态(含 board)落进 store。
@@ -253,7 +253,7 @@ func (o *Orchestrator) execComposeGraph(ctx context.Context, t *contract.Task, t
// 图里无 agent 节点(纯工具/检索图)也要出一段答复。 // 图里无 agent 节点(纯工具/检索图)也要出一段答复。
if fb.answer == "" { if fb.answer == "" {
o.runComposeConversation(ctx, t.ID, fb, plan.System, tr, "agent") o.runComposeConversation(ctx, t.ID, fb, plan.System, tr, "agent", "模型流式推理")
if fb.fatalErr != nil { // 兜底对话也可能触预算顶 / 模型失败 if fb.fatalErr != nil { // 兜底对话也可能触预算顶 / 模型失败
return fb.answer, nil, fb.fatalErr return fb.answer, nil, fb.fatalErr
} }
@@ -289,7 +289,7 @@ func (o *Orchestrator) execDSLNode(ctx context.Context, t *contract.Task, n dsl.
if cbool(n.Config, "autonomous") { if cbool(n.Config, "autonomous") {
o.runReactAgent(ctx, t.ID, b, sys, n, tr, "agent:"+n.ID) o.runReactAgent(ctx, t.ID, b, sys, n, tr, "agent:"+n.ID)
} else { } else {
o.runAgent(ctx, t.ID, b, sys, tr, "agent:"+n.ID) o.runAgent(ctx, t.ID, b, sys, tr, "agent:"+n.ID, labelOf(n, "智能体"))
} }
case "coordinator": // 多智能体协调:orchestrator 自主把子任务派给专家(agent-as-tool)再综合 case "coordinator": // 多智能体协调:orchestrator 自主把子任务派给专家(agent-as-tool)再综合
o.runCoordinator(ctx, t.ID, b, firstNonEmpty(cstr(n.Config, "system"), plan.System), n, tr, "coordinator:"+n.ID) o.runCoordinator(ctx, t.ID, b, firstNonEmpty(cstr(n.Config, "system"), plan.System), n, tr, "coordinator:"+n.ID)
@@ -16,17 +16,17 @@ import (
// runComposeConversation 用 Eino compose.Graph 跑对话主流程: // runComposeConversation 用 Eino compose.Graph 跑对话主流程:
// START → ChatModel 节点 → END,编译为 Runnable 后流式执行;可观测经 callbacks 桥到 ExecEvent。 // START → ChatModel 节点 → END,编译为 Runnable 后流式执行;可观测经 callbacks 桥到 ExecEvent。
// 模型未就绪 / 编译失败时降级回 runAgent(同样的流式回流,保证不回归)。 // 模型未就绪 / 编译失败时降级回 runAgent(同样的流式回流,保证不回归)。
func (o *Orchestrator) runComposeConversation(ctx context.Context, taskID string, b *board, system string, tr *execTracer, node string) { func (o *Orchestrator) runComposeConversation(ctx context.Context, taskID string, b *board, system string, tr *execTracer, node, label string) {
cm := o.pool.ChatModel() cm := o.pool.ChatModel()
if cm == nil { if cm == nil {
o.runAgent(ctx, taskID, b, system, tr, node) // 无模型 → 降级桩 o.runAgent(ctx, taskID, b, system, tr, node, label) // 无模型 → 降级桩
return return
} }
g := compose.NewGraph[[]*schema.Message, *schema.Message]() g := compose.NewGraph[[]*schema.Message, *schema.Message]()
if err := g.AddChatModelNode("model", cm); err != nil { if err := g.AddChatModelNode("model", cm); err != nil {
tr.info(node, "system", "compose 降级", "建图失败,退回自研路径:"+err.Error()) tr.info(node, "system", "compose 降级", "建图失败,退回自研路径:"+err.Error())
o.runAgent(ctx, taskID, b, system, tr, node) o.runAgent(ctx, taskID, b, system, tr, node, label)
return return
} }
_ = g.AddEdge(compose.START, "model") _ = g.AddEdge(compose.START, "model")
@@ -34,7 +34,7 @@ func (o *Orchestrator) runComposeConversation(ctx context.Context, taskID string
r, err := g.Compile(ctx) r, err := g.Compile(ctx)
if err != nil { if err != nil {
tr.info(node, "system", "compose 降级", "编译失败,退回自研路径:"+err.Error()) tr.info(node, "system", "compose 降级", "编译失败,退回自研路径:"+err.Error())
o.runAgent(ctx, taskID, b, system, tr, node) o.runAgent(ctx, taskID, b, system, tr, node, label)
return return
} }
@@ -2,7 +2,9 @@ package eino
import ( import (
"context" "context"
"encoding/json"
"strings" "strings"
"sync"
"testing" "testing"
"github.com/cloudwego/eino/components/model" "github.com/cloudwego/eino/components/model"
@@ -11,6 +13,7 @@ import (
"github.com/sundynix/sundynix-dispatcher/internal/harness" "github.com/sundynix/sundynix-dispatcher/internal/harness"
"github.com/sundynix/sundynix-dispatcher/internal/llm" "github.com/sundynix/sundynix-dispatcher/internal/llm"
"github.com/sundynix/sundynix-shared/contract"
) )
// stubModel 是实现 Eino model.BaseChatModel 的测试桩,固定回一段文本(确定性)。 // stubModel 是实现 Eino model.BaseChatModel 的测试桩,固定回一段文本(确定性)。
@@ -59,7 +62,7 @@ func TestComposeConversation(t *testing.T) {
sink: fs, sink: fs,
} }
b := &board{query: "你好"} b := &board{query: "你好"}
o.runComposeConversation(context.Background(), "task_compose", b, "", &execTracer{}, "agent") o.runComposeConversation(context.Background(), "task_compose", b, "", &execTracer{}, "agent", "模型流式推理")
if !strings.Contains(b.answer, "compose 路径") { if !strings.Contains(b.answer, "compose 路径") {
t.Fatalf("成稿未含模型输出: %q", b.answer) t.Fatalf("成稿未含模型输出: %q", b.answer)
@@ -79,8 +82,70 @@ func TestComposeConversationDegradesToRunAgent(t *testing.T) {
sink: fs, sink: fs,
} }
b := &board{query: "你好"} b := &board{query: "你好"}
o.runComposeConversation(context.Background(), "task_degrade", b, "", &execTracer{}, "agent") o.runComposeConversation(context.Background(), "task_degrade", b, "", &execTracer{}, "agent", "模型流式推理")
if !strings.Contains(b.answer, "降级路径") { if !strings.Contains(b.answer, "降级路径") {
t.Fatalf("无 ChatModel 应降级 runAgent 出稿: %q", b.answer) t.Fatalf("无 ChatModel 应降级 runAgent 出稿: %q", b.answer)
} }
} }
// 图里可以并排好几个 agent(研究/撰写/审查)。之前 runAgent 把轨迹标签写死成
// "模型流式推理",用户在编排里起的节点名整个丢了 —— 轨迹里三行同名,团队视图里
// 工位只能退回节点 ID(r/w/rev),谁是谁全靠猜。这里钉住:标签必须来自节点 label。
type execCapture struct {
mu sync.Mutex
buf [][]byte
}
func (c *execCapture) PublishExec(_ string, data []byte) error {
c.mu.Lock()
defer c.mu.Unlock()
c.buf = append(c.buf, append([]byte(nil), data...))
return nil
}
func (c *execCapture) CompleteExec(string) error { return nil }
func (c *execCapture) labels() []string {
c.mu.Lock()
defer c.mu.Unlock()
var out []string
for _, b := range c.buf {
var e contract.ExecEvent
if json.Unmarshal(b, &e) == nil {
out = append(out, e.Label)
}
}
return out
}
func TestRunAgentUsesNodeLabel(t *testing.T) {
cap := &execCapture{}
o := &Orchestrator{
pool: &fakeLLM{ready: true, cm: &stubModel{reply: "要点若干"}},
breaker: harness.NewCircuitBreaker(),
sink: &fakeSink{},
exec: cap,
}
o.runAgent(context.Background(), "task_label", &board{query: "选型"}, "", o.tracer("task_label"), "agent:r", "研究 Agent")
got := strings.Join(cap.labels(), "|")
if !strings.Contains(got, "研究 Agent") {
t.Fatalf("轨迹标签应为节点名「研究 Agent」,实际: %q", got)
}
if strings.Contains(got, "模型流式推理") {
t.Fatalf("不该再出现写死的兜底标签: %q", got)
}
}
func TestRunAgentLabelFallback(t *testing.T) {
cap := &execCapture{}
o := &Orchestrator{
pool: &fakeLLM{ready: true, cm: &stubModel{reply: "x"}},
breaker: harness.NewCircuitBreaker(),
sink: &fakeSink{},
exec: cap,
}
o.runAgent(context.Background(), "task_fb", &board{query: "q"}, "", o.tracer("task_fb"), "agent:x", "")
if got := strings.Join(cap.labels(), "|"); !strings.Contains(got, "模型流式推理") {
t.Fatalf("节点没起名时应回兜底标签: %q", got)
}
}
@@ -222,13 +222,13 @@ func (o *Orchestrator) runCoordinator(ctx context.Context, taskID string, b *boa
tcm := o.pool.ToolCallingModel() tcm := o.pool.ToolCallingModel()
if tcm == nil || len(specs) == 0 { if tcm == nil || len(specs) == 0 {
tr.info(node, "system", "协调者降级", "模型不支持函数调用或未配置专家,退回普通对话") tr.info(node, "system", "协调者降级", "模型不支持函数调用或未配置专家,退回普通对话")
o.runAgent(ctx, taskID, b, system, tr, node) o.runAgent(ctx, taskID, b, system, tr, node, labelOf(n, "多智能体协调"))
return return
} }
specialists := o.buildSpecialists(ctx, specs, b, taskID, tr) specialists := o.buildSpecialists(ctx, specs, b, taskID, tr)
if len(specialists) == 0 { if len(specialists) == 0 {
tr.info(node, "system", "协调者降级", "无可用专家,退回普通对话") tr.info(node, "system", "协调者降级", "无可用专家,退回普通对话")
o.runAgent(ctx, taskID, b, system, tr, node) o.runAgent(ctx, taskID, b, system, tr, node, labelOf(n, "多智能体协调"))
return return
} }
@@ -240,7 +240,7 @@ func (o *Orchestrator) runCoordinator(ctx context.Context, taskID string, b *boa
}) })
if err != nil { if err != nil {
tr.emit(node, "model", "error", "构建协调者", err.Error(), 0) tr.emit(node, "model", "error", "构建协调者", err.Error(), 0)
o.runAgent(ctx, taskID, b, system, tr, node) o.runAgent(ctx, taskID, b, system, tr, node, labelOf(n, "多智能体协调"))
return return
} }
+7 -4
View File
@@ -132,7 +132,10 @@ func (o *Orchestrator) execToolNode(ctx context.Context, taskID string, n dsl.No
} }
// runAgent 执行 agent/模型节点:据黑板拼消息 → 流式回流 token → 累计成稿。 // runAgent 执行 agent/模型节点:据黑板拼消息 → 流式回流 token → 累计成稿。
func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, system string, tr *execTracer, node string) { // label 是这个节点在轨迹里的显示名,一律由调用方按 labelOf(n, 兜底) 传入:
// 图里可以并排好几个 agent,写死成同一个名字的话轨迹/团队视图里根本分不出谁是谁。
func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, system string, tr *execTracer, node, label string) {
label = firstNonEmpty(label, "模型流式推理")
rc := &RunCtx{ rc := &RunCtx{
System: firstNonEmpty(system, defaultAgentSystem), System: firstNonEmpty(system, defaultAgentSystem),
Query: b.query, Query: b.query,
@@ -155,7 +158,7 @@ func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, sy
return return
} }
} }
tr.emit(node, "model", "start", "模型流式推理", "", 0) tr.emit(node, "model", "start", label, "", 0)
t0 := time.Now() t0 := time.Now()
n := 0 n := 0
var produced strings.Builder // 本节点自身产出(用于沿图向下游传递) var produced strings.Builder // 本节点自身产出(用于沿图向下游传递)
@@ -184,7 +187,7 @@ func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, sy
tr.info(node, "model", "推理过程", fmt.Sprintf("思考 %d 字:%s", len([]rune(rc)), truncate(rc, 200))) tr.info(node, "model", "推理过程", fmt.Sprintf("思考 %d 字:%s", len([]rune(rc)), truncate(rc, 200)))
} }
if err != nil { if err != nil {
tr.emit(node, "model", "error", "模型流式推理", err.Error(), time.Since(t0).Milliseconds()) tr.emit(node, "model", "error", label, err.Error(), time.Since(t0).Milliseconds())
// 未产出任何 token 即失败 → 标记致命错,让任务判 failed(暴露原因,便于监控告警), // 未产出任何 token 即失败 → 标记致命错,让任务判 failed(暴露原因,便于监控告警),
// 而非静默 done-空。已流出部分 token 的中断也算失败(结果不完整)。 // 而非静默 done-空。已流出部分 token 的中断也算失败(结果不完整)。
if b.fatalErr == nil { if b.fatalErr == nil {
@@ -200,7 +203,7 @@ func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, sy
tr.info(node, "system", "输出护栏", fmt.Sprintf("已脱敏 %d 处疑似密钥/PII", red.Hits())) tr.info(node, "system", "输出护栏", fmt.Sprintf("已脱敏 %d 处疑似密钥/PII", red.Hits()))
} }
o.recordAgentOutput(b, produced.String()) // 产出入黑板:成当前成稿 + 供下游接力 o.recordAgentOutput(b, produced.String()) // 产出入黑板:成当前成稿 + 供下游接力
tr.emit(node, "model", "end", "模型流式推理", tr.emit(node, "model", "end", label,
fmt.Sprintf("%d tokens / %d 字", n, len([]rune(produced.String()))), time.Since(t0).Milliseconds()) fmt.Sprintf("%d tokens / %d 字", n, len([]rune(produced.String()))), time.Since(t0).Milliseconds())
} }
@@ -174,7 +174,7 @@ func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *boar
tools := o.agentTools(b, taskID, tr) tools := o.agentTools(b, taskID, tr)
if tcm == nil || len(tools) == 0 { if tcm == nil || len(tools) == 0 {
tr.info(node, "system", "ReAct 降级", "模型不支持函数调用或无可用工具,退回普通对话") tr.info(node, "system", "ReAct 降级", "模型不支持函数调用或无可用工具,退回普通对话")
o.runAgent(ctx, taskID, b, system, tr, node) o.runAgent(ctx, taskID, b, system, tr, node, labelOf(n, "ReAct 智能体"))
return return
} }
@@ -188,7 +188,7 @@ func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *boar
}) })
if err != nil { if err != nil {
tr.emit(node, "model", "error", "构建 ReAct 智能体", err.Error(), 0) tr.emit(node, "model", "error", "构建 ReAct 智能体", err.Error(), 0)
o.runAgent(ctx, taskID, b, system, tr, node) o.runAgent(ctx, taskID, b, system, tr, node, labelOf(n, "ReAct 智能体"))
return return
} }