From 16a6c4b1aad35eca60aa3d73f0c4950e639638a9 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Wed, 24 Jun 2026 13:18:47 +0800 Subject: [PATCH 1/9] =?UTF-8?q?feat(hitl):=20=E4=BA=BA=E5=B7=A5=E5=AE=A1?= =?UTF-8?q?=E6=89=B9=E4=B8=AD=E6=96=AD=EF=BC=88Eino=20Phase=20D=EF=BC=89?= =?UTF-8?q?=E2=80=94=E2=80=94=20=E5=AE=A1=E6=89=B9=E8=8A=82=E7=82=B9?= =?UTF-8?q?=E6=9A=82=E5=81=9C=E2=86=92=E6=89=B9=E5=87=86=E7=BB=AD=E8=B7=91?= =?UTF-8?q?/=E6=8B=92=E7=BB=9D=E4=B8=AD=E6=AD=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 专用「审批」节点方案,全栈打通。 后端: - contract:TaskWaiting/TaskRejected 状态 + ApprovalSubject/ApprovalDecision。 - bus:PublishApproval + WaitApproval(订阅决定主题,带超时);消费者 AckWait 30s→15min(阻塞等人审期间消息未 ack,否则重投成重复任务)。 - dispatcher:ApprovalWaiter 接口 + approvalNode——执行到审批节点发 await 事件 + 置 waiting,阻塞等决定。批准→回 running 放行下游;拒绝/超时→errRejected 哨兵→ 剪下游→rejected,优雅收尾不计熔断。超时安全默认拒绝。 taskExecTimeout 3→10min(审批5 < 执行10 < AckWait15)。 - 网关:POST /tasks/:id/approve(仅 waiting 态受理,幂等)。 桌面端: - Studio 新增「人工审批」节点(nodeCatalog,可填标题/说明)。 - run.ts pendingApproval() 从 exec 流派生待审中断 + waiting 节点状态。 - BottomDrawer ApprovalBar:琥珀审批条(摘要 + 批准/拒绝 + 备注,调 api.approveTask)。 - ExecTrace waiting 状态灯。 验证:后端 curl 实测 waiting→批准→running→done;waiting→拒绝→rejected(下游未跑)。 前端 tsc + vitest 36 过(pendingApproval 4 例 + waiting 状态)。全模块 build+vet+test 全绿。 EINO_ADOPTION Phase D 标记 HITL 完成。 未覆盖:compose 路径(EINO_COMPOSE,默认关)的 approval 节点;桌面端审批条未在 GUI 实点。 Co-Authored-By: Claude Opus 4.8 (1M context) --- EINO_ADOPTION.md | 2 +- .../frontend/src/components/ExecTrace.tsx | 1 + sundynix-desktop/frontend/src/lib/api.ts | 12 ++++ sundynix-desktop/frontend/src/lib/run.test.ts | 34 +++++++++- sundynix-desktop/frontend/src/lib/run.ts | 27 +++++++- .../frontend/src/shell/BottomDrawer.tsx | 62 ++++++++++++++++++- .../frontend/src/studio/nodeCatalog.ts | 12 ++++ sundynix-dispatcher/cmd/dispatcher/main.go | 5 +- sundynix-dispatcher/internal/eino/graph.go | 51 +++++++++++++++ .../internal/eino/orchestrator.go | 47 ++++++++++---- .../internal/nats/subscriber.go | 5 ++ .../internal/handler/task_handler.go | 28 +++++++++ sundynix-gateway/internal/nats/publisher.go | 5 ++ sundynix-gateway/internal/router/router.go | 39 ++++++------ sundynix-shared/bus/bus.go | 45 ++++++++++++++ sundynix-shared/contract/task.go | 24 ++++++- 16 files changed, 360 insertions(+), 39 deletions(-) diff --git a/EINO_ADOPTION.md b/EINO_ADOPTION.md index 1ac2753..b3774c8 100644 --- a/EINO_ADOPTION.md +++ b/EINO_ADOPTION.md @@ -10,7 +10,7 @@ - [x] **Phase A · 地基**:`llm.Pool` 换 Eino ChatModel 组件(commit d84b1ec,验收通过) - [x] **Phase B · 质变**:MCP 工具→`InvokableTool` + ReAct agent(模型自主调工具,验收 7/7 命中) - [x] **Phase C · 编排归一**:✅ 全图 `DSL→compose.Graph` 编译器(全节点 + branch + DAG 并行调度)+ callbacks→ExecEvent 归一,等价回归通过(EINO_COMPOSE 灰度开关,默认关;过渡期后 graph.go 退役) -- [~] **Phase D · 状态化执行**:✅ 任务生命周期 FSM(已完成)/ ⬜ HITL 中断恢复 / ⬜ 多智能体(按场景) +- [~] **Phase D · 状态化执行**:✅ 任务生命周期 FSM / ✅ HITL 人工审批中断(审批节点暂停→waiting→批准回 running / 拒绝→rejected,NATS 决定回传 + AckWait 续租,全栈含 Studio 审批节点 + 运行抽屉批准条)/ ⬜ 多智能体(按场景) **组件化补完(A)**:检索 → `ragRetriever`(`components/retriever.Retriever`,`eino_components.go`);提示词 → `buildMessages` 改用 `prompt.FromMessages`+`MessagesPlaceholder`;工具 → `mcpTool`(`InvokableTool`,模型自主调用)。至此终态架构 8 层中 模型/工具/检索/提示词/编排/智能体(单)/可观测 均已 Eino 组件化;剩 人机交互(中断恢复,按场景)。 **自主 agent 工具集动态化**:agent 工具集不再硬编码——mcp-go 注册表(单一事实源)每个工具声明 `agent/params/inject`,`list_tools` 上报,dispatcher `agentTools()` 动态发现并建 `InvokableTool`、运行时注入 `user_id/session_id/kb`(不暴露给模型)。**加工具只改 mcp-go 注册表,dispatcher 零改动。** 当前暴露 4 个(wiki_search/recall_user_memory/remember_user_fact/history_get);实测模型自主调用新暴露的 remember_user_fact 成功(参数自生成、user_id 服务端注入)。 diff --git a/sundynix-desktop/frontend/src/components/ExecTrace.tsx b/sundynix-desktop/frontend/src/components/ExecTrace.tsx index a07eebb..dfc6968 100644 --- a/sundynix-desktop/frontend/src/components/ExecTrace.tsx +++ b/sundynix-desktop/frontend/src/components/ExecTrace.tsx @@ -37,6 +37,7 @@ function StatusDot({ status }: { status: NodeTrace["status"] }) { if (status === "running") return ; if (status === "done") return ; if (status === "error") return ; + if (status === "waiting") return ; // 待审批 return ; } diff --git a/sundynix-desktop/frontend/src/lib/api.ts b/sundynix-desktop/frontend/src/lib/api.ts index ff83de4..ad0e9e6 100644 --- a/sundynix-desktop/frontend/src/lib/api.ts +++ b/sundynix-desktop/frontend/src/lib/api.ts @@ -109,6 +109,18 @@ export async function submitTask(dsl: TaskDsl, id: Identity): Promise { return data.task_id; } +// approveTask: POST /api/v1/tasks/:id/approve —— HITL 人工审批决定(批准放行 / 拒绝中止)。 +export async function approveTask(taskId: string, approved: boolean, opts?: { node?: string; note?: string }): Promise { + const res = guard401( + await fetch(`${GATEWAY}/api/v1/tasks/${taskId}/approve`, { + method: "POST", + headers: { "Content-Type": "application/json", ...bearer() }, + body: JSON.stringify({ approved, node: opts?.node ?? "", note: opts?.note ?? "" }), + }), + ); + if (!res.ok) throw new Error(`approve failed: ${res.status} ${await res.text()}`); +} + // streamTokens: 订阅 SSE /api/v1/tasks/:id/stream,逐 token 回调,done 收尾。 // 返回关闭函数。注意 EventSource 无法带请求头,但流按 task_id 寻址,无需身份头。 export function streamTokens( diff --git a/sundynix-desktop/frontend/src/lib/run.test.ts b/sundynix-desktop/frontend/src/lib/run.test.ts index 286a78d..5efe026 100644 --- a/sundynix-desktop/frontend/src/lib/run.test.ts +++ b/sundynix-desktop/frontend/src/lib/run.test.ts @@ -1,6 +1,6 @@ import { describe, it, expect } from "vitest"; import type { ExecEvent } from "./api"; -import { deriveNodes } from "./run"; +import { deriveNodes, pendingApproval } from "./run"; let seq = 0; function ev(node: string, phase: string, extra: Partial = {}): ExecEvent { @@ -48,4 +48,36 @@ describe("deriveNodes(ExecEvent 流 → 节点轨迹)", () => { const [n] = deriveNodes([ev("a", "start", { label: "旧" }), ev("a", "end", { label: "新" })]); expect(n.label).toBe("新"); }); + + it("await 事件 → 节点状态 waiting", () => { + const [n] = deriveNodes([ev("ap", "await", { kind: "approval", label: "审批" })]); + expect(n.status).toBe("waiting"); + }); +}); + +describe("pendingApproval(HITL 待审批中断)", () => { + it("await 未收口 → 返回待审批", () => { + const p = pendingApproval([ev("ap", "await", { kind: "approval", label: "高危审批", detail: "摘要" })]); + expect(p).toEqual({ node: "ap", title: "高危审批", summary: "摘要" }); + }); + + it("await 后 end(同节点)→ 已决,返回 null", () => { + const p = pendingApproval([ + ev("ap", "await", { kind: "approval" }), + ev("ap", "end", { kind: "approval" }), + ]); + expect(p).toBeNull(); + }); + + it("await 后 error(超时/拒绝)→ 已决,返回 null", () => { + const p = pendingApproval([ + ev("ap", "await", { kind: "approval" }), + ev("ap", "error", { kind: "approval" }), + ]); + expect(p).toBeNull(); + }); + + it("无审批事件 → null", () => { + expect(pendingApproval([ev("a", "start"), ev("a", "end")])).toBeNull(); + }); }); diff --git a/sundynix-desktop/frontend/src/lib/run.ts b/sundynix-desktop/frontend/src/lib/run.ts index f22d5ca..fbb0211 100644 --- a/sundynix-desktop/frontend/src/lib/run.ts +++ b/sundynix-desktop/frontend/src/lib/run.ts @@ -21,7 +21,7 @@ export const emptyRun: RunState = { phase: "idle", output: "", events: [], exec: // ---- 执行轨迹派生:把扁平 ExecEvent 流归并为按节点聚合的轨迹 ---- -export type NodeStatus = "running" | "done" | "error" | "info"; +export type NodeStatus = "running" | "done" | "error" | "info" | "waiting"; export interface NodeTrace { node: string; @@ -34,6 +34,28 @@ export interface NodeTrace { order: number; } +// PendingApproval 是一个待人工审批的中断(HITL):审批节点已发 await 事件、尚未被 end/error 收口。 +export interface PendingApproval { + node: string; + title: string; + summary: string; +} + +// pendingApproval 从执行事件流里找出当前待审批的中断:取最后一个 kind=approval & phase=await, +// 且其后该节点没有 end/error(未被批准/拒绝收口)的事件。无则 null。 +export function pendingApproval(events: ExecEvent[]): PendingApproval | null { + let pending: PendingApproval | null = null; + for (const e of events) { + if (e.kind !== "approval") continue; + if (e.phase === "await") { + pending = { node: e.node, title: e.label || "人工审批", summary: e.detail || "" }; + } else if (e.phase === "end" || e.phase === "error") { + if (pending && pending.node === e.node) pending = null; // 已决,收口 + } + } + return pending; +} + // deriveNodes 把事件流按 node 归并:start→running,end→done(带耗时),error→error,info→点事件/附注。 export function deriveNodes(events: ExecEvent[]): NodeTrace[] { const map = new Map(); @@ -50,6 +72,9 @@ export function deriveNodes(events: ExecEvent[]): NodeTrace[] { case "start": if (n.status !== "done" && n.status !== "error") n.status = "running"; break; + case "await": // HITL 审批节点暂停,等人工决定 + n.status = "waiting"; + break; case "end": n.status = "done"; n.ms = e.ms; diff --git a/sundynix-desktop/frontend/src/shell/BottomDrawer.tsx b/sundynix-desktop/frontend/src/shell/BottomDrawer.tsx index af4b093..e79ddb7 100644 --- a/sundynix-desktop/frontend/src/shell/BottomDrawer.tsx +++ b/sundynix-desktop/frontend/src/shell/BottomDrawer.tsx @@ -1,7 +1,8 @@ import { useState } from "react"; -import { ChevronDown, ChevronUp, Wrench } from "lucide-react"; -import { deriveNodes, type RunState } from "../lib/run"; +import { ChevronDown, ChevronUp, Wrench, ShieldCheck, Check, X } from "lucide-react"; +import { deriveNodes, pendingApproval, type RunState } from "../lib/run"; import { ExecTrace } from "../components/ExecTrace"; +import { approveTask } from "../lib/api"; import { Tabs, Badge, cn, type TabDef } from "../ui"; type Tab = "output" | "trace" | "tools" | "cite" | "eval"; @@ -26,8 +27,11 @@ export function BottomDrawer({ run }: { run: RunState }) { const statusText = run.phase === "streaming" ? "流式中…" : run.phase === "done" ? "完成 ✓" : run.phase === "error" ? `✗ ${run.error ?? "出错"}` : run.phase === "submitting" ? "提交中…" : "就绪"; + const approval = run.taskId ? pendingApproval(run.exec) : null; + return (
+ {approval && run.taskId && }
(null); + const [err, setErr] = useState(""); + + const decide = async (approved: boolean) => { + setBusy(approved ? "approve" : "reject"); + setErr(""); + try { + await approveTask(taskId, approved, { node, note }); + } catch (e) { + setErr((e as Error).message); + setBusy(null); + } + }; + + return ( +
+
+ + 人工审批 + {title} + 等待决定 +
+ {summary &&
{summary}
} +
+ setNote(e.target.value)} + placeholder="备注(可选,拒绝原因等)" + className="min-w-0 flex-1 rounded border border-line bg-ink-950/60 px-2 py-1 text-[11px] text-slate-200 placeholder:text-slate-600 focus:border-amber-500/50 focus:outline-none" + /> + + +
+ {err &&

提交失败:{err}

} +
+ ); +} + // ToolCalls:从执行事件里筛出工具调用节点,逐条展示入参 → 产出 + 耗时/状态。 function ToolCalls({ run }: { run: RunState }) { const tools = deriveNodes(run.exec).filter((n) => n.kind === "tool"); diff --git a/sundynix-desktop/frontend/src/studio/nodeCatalog.ts b/sundynix-desktop/frontend/src/studio/nodeCatalog.ts index 800a988..e320146 100644 --- a/sundynix-desktop/frontend/src/studio/nodeCatalog.ts +++ b/sundynix-desktop/frontend/src/studio/nodeCatalog.ts @@ -102,6 +102,18 @@ export const NODE_KINDS: Record = { fields: [{ key: "condition", label: "条件", type: "text", placeholder: "score > 0.8" }], defaults: { condition: "" }, }, + approval: { + kind: "approval", + label: "人工审批", + accent: "border-l-amber-500", + badge: "bg-amber-100 text-amber-700", + desc: "HITL:暂停等人工批准", + fields: [ + { key: "title", label: "审批标题", type: "text", placeholder: "如:高危操作审批" }, + { key: "prompt", label: "审批说明", type: "textarea", placeholder: "向审批人说明待执行的操作…" }, + ], + defaults: { title: "人工审批", prompt: "请审批是否继续执行后续步骤" }, + }, map: { kind: "map", label: "并行 / Map", diff --git a/sundynix-dispatcher/cmd/dispatcher/main.go b/sundynix-dispatcher/cmd/dispatcher/main.go index b3d11d4..fd1655a 100644 --- a/sundynix-dispatcher/cmd/dispatcher/main.go +++ b/sundynix-dispatcher/cmd/dispatcher/main.go @@ -45,8 +45,9 @@ func main() { } go sub.FetchModelConfigWithRetry(context.Background(), pool.SetConfig) - // sub 同时作为 Token 回流(TokenSink)、MCP 工具调用(ToolCaller)、执行事件(ExecSink)与任务状态回写(StatusSink)出口。 - orch, err := eino.NewOrchestrator(pool, breaker, eval, sub, sub, sub, sub) + // sub 同时作为 Token 回流(TokenSink)、MCP 工具调用(ToolCaller)、执行事件(ExecSink)、 + // 任务状态回写(StatusSink)与 HITL 审批等待(ApprovalWaiter)出口。 + orch, err := eino.NewOrchestrator(pool, breaker, eval, sub, sub, sub, sub, sub) if err != nil { log.Fatalf("[dispatcher] build eino graph: %v", err) } diff --git a/sundynix-dispatcher/internal/eino/graph.go b/sundynix-dispatcher/internal/eino/graph.go index 39a09bc..3729e0f 100644 --- a/sundynix-dispatcher/internal/eino/graph.go +++ b/sundynix-dispatcher/internal/eino/graph.go @@ -32,6 +32,7 @@ type board struct { sections []reportSection // map 并行 fan-out 产出的分项成稿(供 render 多章渲染) answer string // 当前成稿(多 agent 协作时 = 最近一个 agent 的产出 = 成品) agentOut []string // 各上游 agent 的产出(按序),注入下游 agent 上下文以实现接力协作 + rejected bool // HITL 审批节点拒绝/超时 → 置位,runGraph 中止并返回 errRejected } // runGraph 按 DSL 图的真实拓扑与连线执行(替代旧的线性拍平 compileFlow)。 @@ -144,6 +145,8 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT o.renderNode(nctx, t.ID, n, b, tr) case "branch": propagate = o.branchNode(n, b, outE[n.ID], nodeByID, tr) + case "approval": + propagate = o.approvalNode(nctx, t.ID, n, b, tr, outE[n.ID]) case "map": o.mapNode(nctx, t.ID, n, b, tr) case "output": @@ -152,11 +155,18 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT tr.info(n.Kind+":"+n.ID, "system", labelOf(n, n.Kind), "未识别节点,跳过") } nspan.End() + if b.rejected { + break // 审批拒绝/超时:中止后续节点,整图判 rejected + } for _, tgt := range propagate { active[tgt] = true } } + if b.rejected { + return b.answer, errRejected // 合法终态,Handle 据此判 rejected 并优雅收尾 + } + // 图里无 agent 节点(纯工具/检索图)也要出一段模型答复,否则没有输出。 if b.answer == "" { o.runConversation(ctx, t.ID, b, plan.System, tr, "agent") @@ -369,6 +379,47 @@ func (o *Orchestrator) branchNode(n dsl.Node, b *board, outs []dsl.Edge, byID ma return chosen } +// approvalNode 是 HITL 人工审批中断:执行到此暂停,把待审摘要推给 UI(exec 事件 kind=approval/phase=await +// + 状态 waiting),阻塞等人工批准/拒绝(带超时,安全默认拒绝)。 +// 批准 → 状态回 running 并放行下游;拒绝/超时 → 置 b.rejected 中止全图。返回应激活的下游(拒绝=空)。 +func (o *Orchestrator) approvalNode(ctx context.Context, taskID string, n dsl.Node, b *board, tr *execTracer, outs []dsl.Edge) []string { + title := firstNonEmpty(cstr(n.Config, "title"), labelOf(n, "人工审批")) + prompt := firstNonEmpty(cstr(n.Config, "prompt"), "请审批是否继续执行后续步骤") + summary := prompt + if b.answer != "" { // 带上当前产出预览,便于审批人判断 + summary = prompt + "\n—— 当前产出预览 ——\n" + truncate(b.answer, 400) + } + + // 未接审批通道(单测/降级)→ 自动放行,避免无人应答卡死。 + if o.approval == nil { + tr.info("approval:"+n.ID, "approval", title, "未接审批通道,自动放行") + return targetsOf(outs) + } + + // 暂停:发待审事件(UI 据 kind=approval & phase=await 弹批准/拒绝)+ 置任务 waiting。 + tr.emit("approval:"+n.ID, "approval", "await", title, summary, 0) + o.setStatus(taskID, contract.TaskWaiting, title) + + dec, err := o.approval.WaitApproval(ctx, taskID, approvalTimeout) + switch { + case err != nil: // 超时 / ctx 取消 → 安全默认拒绝 + b.rejected = true + b.answer = "❌ 审批超时未决,已自动拒绝:" + title + tr.emit("approval:"+n.ID, "approval", "error", title, "审批超时,自动拒绝", 0) + return nil + case !dec.Approved: + b.rejected = true + note := firstNonEmpty(dec.Note, "审批人拒绝") + b.answer = "❌ 已被拒绝:" + note + tr.emit("approval:"+n.ID, "approval", "end", title, "拒绝:"+note, 0) + return nil + default: // 批准 → 恢复执行,放行下游 + o.setStatus(taskID, contract.TaskRunning, "审批通过,继续执行") + tr.emit("approval:"+n.ID, "approval", "end", title, "批准:"+firstNonEmpty(dec.Note, "放行"), 0) + return targetsOf(outs) + } +} + // targetsOf 取一组边的目标节点 ID(保持顺序)。 func targetsOf(edges []dsl.Edge) []string { out := make([]string, 0, len(edges)) diff --git a/sundynix-dispatcher/internal/eino/orchestrator.go b/sundynix-dispatcher/internal/eino/orchestrator.go index 11d8ed9..4019d4b 100644 --- a/sundynix-dispatcher/internal/eino/orchestrator.go +++ b/sundynix-dispatcher/internal/eino/orchestrator.go @@ -40,6 +40,15 @@ type StatusSink interface { PublishTaskStatus(taskID, status, detail string) error } +// ApprovalWaiter 阻塞等待审批节点的人工决定(由 NATS bus 实现;可为 nil → 审批节点自动放行)。 +type ApprovalWaiter interface { + WaitApproval(ctx context.Context, taskID string, timeout time.Duration) (*contract.ApprovalDecision, error) +} + +// errRejected 是审批节点拒绝(或超时)时图执行返回的哨兵错误:它是合法终态而非故障, +// Handle 据此判 rejected 并优雅收尾(不计熔断失败)。 +var errRejected = errors.New("approval rejected") + // LLM 是编排所需的语言模型能力(生产由 *llm.Pool 实现)。抽成接口便于测试注入假模型。 type LLM interface { Ready() bool @@ -56,17 +65,22 @@ type LLM interface { const toolCallTimeout = 3 * time.Second // taskExecTimeout 是单个任务整体执行上限;超时即判 timeout(状态机),避免无限期"运行中"。 -const taskExecTimeout = 3 * time.Minute +// 含 HITL 审批等待预算(approvalTimeout)+ 常规图执行;须 < bus 消费者 AckWait(15min) 以免重投。 +const taskExecTimeout = 10 * time.Minute + +// approvalTimeout 是单个审批节点等待人工决定的上限;超时安全默认拒绝(fail-safe)。 +const approvalTimeout = 5 * time.Minute // Orchestrator 把每个 DSL 任务动态编译为 Eino 图并执行(记忆召回 → 工具节点 → 注入 → 流式)。 type Orchestrator struct { - pool LLM - breaker *harness.CircuitBreaker - eval *harness.Evaluator - sink TokenSink - tools ToolCaller - exec ExecSink - status StatusSink // 任务生命周期状态回写(可为 nil) + pool LLM + breaker *harness.CircuitBreaker + eval *harness.Evaluator + sink TokenSink + tools ToolCaller + exec ExecSink + status StatusSink // 任务生命周期状态回写(可为 nil) + approval ApprovalWaiter // HITL 审批等待(可为 nil → 审批节点自动放行) turnMu sync.Mutex // 保护 turns(攒批计数,多任务 goroutine 共享) turns map[string]int // sessionID → 累计轮次,用于每 N 轮触发 consolidate @@ -74,9 +88,9 @@ type Orchestrator struct { // NewOrchestrator 持有依赖;图按任务的 DSL 在 Handle 内动态编译。 // exec 为执行可视化事件出口(可为 nil,则不发轨迹事件);eval 为自动化评测(可为 nil); -// status 为任务生命周期状态回写出口(可为 nil)。 -func NewOrchestrator(pool LLM, breaker *harness.CircuitBreaker, eval *harness.Evaluator, sink TokenSink, tools ToolCaller, exec ExecSink, status StatusSink) (*Orchestrator, error) { - return &Orchestrator{pool: pool, breaker: breaker, eval: eval, sink: sink, tools: tools, exec: exec, status: status}, nil +// status 为任务生命周期状态回写出口(可为 nil);approval 为 HITL 审批等待(可为 nil)。 +func NewOrchestrator(pool LLM, breaker *harness.CircuitBreaker, eval *harness.Evaluator, sink TokenSink, tools ToolCaller, exec ExecSink, status StatusSink, approval ApprovalWaiter) (*Orchestrator, error) { + return &Orchestrator{pool: pool, breaker: breaker, eval: eval, sink: sink, tools: tools, exec: exec, status: status, approval: approval}, nil } // setStatus 回写一次任务状态流转(status 为 nil 时静默跳过)。 @@ -144,6 +158,17 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error { // 按 DSL 图执行:compose.Graph(EINO_COMPOSE=1)或自研 graph.go(默认);agent 节点流式回流 token。 answer, err := o.executeGraph(tctx, t, tr) + if errors.Is(err, errRejected) { + // HITL 拒绝:合法终态,非故障。收尾流 + 置 rejected,不计熔断、不重投。 + slog.InfoContext(ctx, "task rejected by approval", "task_id", t.ID) + if answer != "" { + _ = o.sink.PublishToken(t.ID, []byte(answer)) + } + _ = o.sink.CompleteStream(t.ID) + o.breaker.Report(true) // 拒绝是人为决策,不算后端失败 + o.setStatus(t.ID, contract.TaskRejected, truncate(answer, 120)) + return nil + } if err != nil { span.RecordError(err) span.SetStatus(codes.Error, err.Error()) diff --git a/sundynix-dispatcher/internal/nats/subscriber.go b/sundynix-dispatcher/internal/nats/subscriber.go index bebaecf..c9b4f13 100644 --- a/sundynix-dispatcher/internal/nats/subscriber.go +++ b/sundynix-dispatcher/internal/nats/subscriber.go @@ -81,6 +81,11 @@ func (s *Subscriber) PublishTaskStatus(taskID, status, detail string) error { }) } +// WaitApproval 让 Subscriber 满足 eino.ApprovalWaiter,阻塞等待审批节点的人工决定。 +func (s *Subscriber) WaitApproval(ctx context.Context, taskID string, timeout time.Duration) (*contract.ApprovalDecision, error) { + return s.inner.WaitApproval(ctx, taskID, timeout) +} + // RequestModelConfig 向控制面(Gateway)取当前激活的对话模型配置。 func (s *Subscriber) RequestModelConfig(ctx context.Context) (*contract.ModelConfig, error) { return s.inner.RequestConfig(ctx, contract.ConfigKindChat) diff --git a/sundynix-gateway/internal/handler/task_handler.go b/sundynix-gateway/internal/handler/task_handler.go index b4c6be8..dabdc2c 100644 --- a/sundynix-gateway/internal/handler/task_handler.go +++ b/sundynix-gateway/internal/handler/task_handler.go @@ -89,6 +89,34 @@ func (h *Handler) TaskStatus(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"task_id": id, "status": status, "detail": detail}) } +// ApproveTask: POST /api/v1/tasks/:id/approve {approved, node?, note?} —— 人工审批决定(HITL)。 +// 把决定经 NATS 发给 dispatcher,解除审批节点的阻塞(批准放行 / 拒绝中止)。 +func (h *Handler) ApproveTask(c *gin.Context) { + id := c.Param("id") + var body struct { + Approved bool `json:"approved"` + Node string `json:"node"` + Note string `json:"note"` + } + if err := c.ShouldBindJSON(&body); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + // 仅在任务确为等待审批时受理(幂等:重复/迟到的决定不报错,dispatcher 侧已只取首条)。 + if status, _ := h.db.GetTaskStatus(c.Request.Context(), id); status != contract.TaskWaiting { + c.JSON(http.StatusConflict, gin.H{"error": "任务当前非待审批状态", "status": status}) + return + } + if err := h.bus.PublishApproval(&contract.ApprovalDecision{ + TaskID: id, Node: body.Node, Approved: body.Approved, Note: body.Note, + By: userID(c), TS: time.Now().UnixMilli(), + }); err != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"task_id": id, "approved": body.Approved}) +} + // StreamTask: 以 SSE 把 Token Stream 推给客户端。 // 优先从 Redis Stream 读(可回放 + 断点续传,根治连晚/重连丢 token);Redis 降级时回退 live NATS。 func (h *Handler) StreamTask(c *gin.Context) { diff --git a/sundynix-gateway/internal/nats/publisher.go b/sundynix-gateway/internal/nats/publisher.go index 181a8ea..4905d07 100644 --- a/sundynix-gateway/internal/nats/publisher.go +++ b/sundynix-gateway/internal/nats/publisher.go @@ -64,6 +64,11 @@ func (b *Bus) SubscribeTaskStatus(onEvent func(*contract.TaskStatusEvent)) (func return b.inner.SubscribeTaskStatus(onEvent) } +// PublishApproval 把一次人工审批决定发给 dispatcher(解除审批节点阻塞)。 +func (b *Bus) PublishApproval(dec *contract.ApprovalDecision) error { + return b.inner.PublishApproval(dec) +} + // ServeConfig 让网关作为配置控制面,响应某 kind 的配置请求。 func (b *Bus) ServeConfig(kind string, provide func() *contract.ModelConfig) (func() error, error) { return b.inner.ServeConfig(kind, provide) diff --git a/sundynix-gateway/internal/router/router.go b/sundynix-gateway/internal/router/router.go index 977d851..f455292 100644 --- a/sundynix-gateway/internal/router/router.go +++ b/sundynix-gateway/internal/router/router.go @@ -50,25 +50,26 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob. // —— 受保护:owner 作用域业务,必须携带有效 JWT —— p := api.Group("", middleware.RequireAuth()) { - p.POST("/tasks", h.SubmitTask) // 解析 DSL 并 Publish 到 NATS(带已验证 uid) - p.GET("/tasks/:id", h.TaskStatus) // 任务生命周期状态(UI 轮询 submitted/running/done/failed/timeout) - p.PUT("/memory", h.SetMemory) // 偏好记忆登记(→ mcp-go memory_upsert) - p.GET("/memory", h.ListMemory) // 列出当前用户偏好(记忆面板) - p.DELETE("/memory", h.DeleteMemory) // 软删一条偏好(?key=) - p.GET("/kb/list", h.KbList) // 当前用户的知识库列表(owner 隔离) - p.POST("/kb/create", h.KbCreate) // 新建知识库 - p.POST("/kb/ingest", h.KbIngest) // 文本入库 - p.POST("/kb/ingest_file", h.KbIngestFile) // 文件入库 - p.POST("/kb/search", h.KbSearch) // 检索台 - p.GET("/kb/vault", h.KbVault) // 文库列表 - p.GET("/kb/doc", h.KbDoc) // 取单篇文档 - p.GET("/kb/links", h.KbLinks) // 某库双链 - p.POST("/kb/note", h.KbSaveNote) // 新建/编辑笔记 - p.GET("/kb/graph", h.KbGraph) // 知识图谱三元组 - p.GET("/agents", h.AgentList) // 我的编排列表(owner 隔离) - p.POST("/agents", h.AgentSave) // 保存/更新编排 - p.DELETE("/agents", h.AgentDelete) // 删除编排 - p.POST("/reports", h.GenerateReport) // 报告生成 + p.POST("/tasks", h.SubmitTask) // 解析 DSL 并 Publish 到 NATS(带已验证 uid) + p.GET("/tasks/:id", h.TaskStatus) // 任务生命周期状态(UI 轮询 submitted/running/done/failed/timeout/waiting/rejected) + p.POST("/tasks/:id/approve", h.ApproveTask) // HITL 人工审批决定(批准/拒绝) + p.PUT("/memory", h.SetMemory) // 偏好记忆登记(→ mcp-go memory_upsert) + p.GET("/memory", h.ListMemory) // 列出当前用户偏好(记忆面板) + p.DELETE("/memory", h.DeleteMemory) // 软删一条偏好(?key=) + p.GET("/kb/list", h.KbList) // 当前用户的知识库列表(owner 隔离) + p.POST("/kb/create", h.KbCreate) // 新建知识库 + p.POST("/kb/ingest", h.KbIngest) // 文本入库 + p.POST("/kb/ingest_file", h.KbIngestFile) // 文件入库 + p.POST("/kb/search", h.KbSearch) // 检索台 + p.GET("/kb/vault", h.KbVault) // 文库列表 + p.GET("/kb/doc", h.KbDoc) // 取单篇文档 + p.GET("/kb/links", h.KbLinks) // 某库双链 + p.POST("/kb/note", h.KbSaveNote) // 新建/编辑笔记 + p.GET("/kb/graph", h.KbGraph) // 知识图谱三元组 + p.GET("/agents", h.AgentList) // 我的编排列表(owner 隔离) + p.POST("/agents", h.AgentSave) // 保存/更新编排 + p.DELETE("/agents", h.AgentDelete) // 删除编排 + p.POST("/reports", h.GenerateReport) // 报告生成 p.GET("/billing", h.Billing) } diff --git a/sundynix-shared/bus/bus.go b/sundynix-shared/bus/bus.go index 939ff56..7f08bd9 100644 --- a/sundynix-shared/bus/bus.go +++ b/sundynix-shared/bus/bus.go @@ -314,6 +314,48 @@ func (b *Bus) SubscribeTaskStatus(onEvent func(*contract.TaskStatusEvent)) (unsu return sub.Unsubscribe, nil } +// ---- 人工审批(HITL,core NATS pub-sub)---- + +// PublishApproval 广播一次人工审批决定(网关在收到 UI 的批准/拒绝后调用)。 +func (b *Bus) PublishApproval(dec *contract.ApprovalDecision) error { + data, err := json.Marshal(dec) + if err != nil { + return err + } + return b.nc.Publish(contract.ApprovalSubject(dec.TaskID), data) +} + +// WaitApproval 阻塞等待某任务的人工审批决定,直到收到、ctx 取消或超时。 +// dispatcher 在审批节点调用:先订阅再等待(订阅早于决定到达,避免错过)。 +// 超时返回 (nil, error) —— 调用方据安全默认(拒绝)处理。 +func (b *Bus) WaitApproval(ctx context.Context, taskID string, timeout time.Duration) (*contract.ApprovalDecision, error) { + ch := make(chan *contract.ApprovalDecision, 1) + sub, err := b.nc.Subscribe(contract.ApprovalSubject(taskID), func(m *nats.Msg) { + var dec contract.ApprovalDecision + if json.Unmarshal(m.Data, &dec) == nil { + select { + case ch <- &dec: + default: // 已收到一条,丢弃后续重复决定 + } + } + }) + if err != nil { + return nil, fmt.Errorf("subscribe approval: %w", err) + } + defer func() { _ = sub.Unsubscribe() }() + + timer := time.NewTimer(timeout) + defer timer.Stop() + select { + case dec := <-ch: + return dec, nil + case <-timer.C: + return nil, fmt.Errorf("approval timeout after %s", timeout) + case <-ctx.Done(): + return nil, ctx.Err() + } +} + // ---- 配置控制面(core NATS request-reply + broadcast)---- // RequestConfig 向控制面(Gateway)请求某 kind 当前激活配置(chat/embedding)。 @@ -406,6 +448,9 @@ func (b *Bus) ConsumeTasks(ctx context.Context, h TaskHandler) (stop func(), err Durable: contract.ConsumerDurable, AckPolicy: jetstream.AckExplicitPolicy, FilterSubject: contract.SubjectTasksAll, + // HITL:审批节点会让 Handle 阻塞等人工决定(最长约 5 分钟), + // AckWait 必须覆盖「审批等待 + 图执行」总时长,否则消息在途未 ack 会被重投成重复任务。 + AckWait: 15 * time.Minute, }) if err != nil { return nil, fmt.Errorf("create consumer: %w", err) diff --git a/sundynix-shared/contract/task.go b/sundynix-shared/contract/task.go index fa20099..4e4178a 100644 --- a/sundynix-shared/contract/task.go +++ b/sundynix-shared/contract/task.go @@ -33,26 +33,46 @@ const ( // core NATS pub-sub(状态是幂等覆盖,丢一条由下一条纠正,无需持久化)。 // 注意:必须在 sundynix.tasks.> 之外,否则会被任务流捕获成"幽灵任务"自我放大。 SubjectTaskStatus = "sundynix.status.task" + + // 人工审批(HITL)决定回传前缀:实际 sundynix.approval.。 + // core NATS pub-sub:UI 点批准/拒绝 → 网关发到此 → dispatcher 解除审批节点的阻塞。 + SubjectApproval = "sundynix.approval" ) // 任务生命周期状态机:submitted(网关建任务)→ running(dispatcher 开跑) // → done / failed / timeout(dispatcher 收尾)。 +// HITL:执行到审批节点 → waiting(等人工决定)→ 批准回 running / 拒绝→rejected。 const ( TaskSubmitted = "submitted" TaskRunning = "running" TaskDone = "done" TaskFailed = "failed" TaskTimeout = "timeout" + TaskWaiting = "waiting" // 等待人工审批 + TaskRejected = "rejected" // 人工拒绝(或审批超时,安全默认拒绝) ) // TaskStatusEvent 是一次任务状态流转事件(经 SubjectTaskStatus 回流给网关)。 type TaskStatusEvent struct { TaskID string `json:"task_id"` - Status string `json:"status"` // running / done / failed / timeout - Detail string `json:"detail,omitempty"` // 失败原因等 + Status string `json:"status"` // running / done / failed / timeout / waiting / rejected + Detail string `json:"detail,omitempty"` // 失败原因 / 审批摘要等 TS int64 `json:"ts"` // unix 毫秒 } +// ApprovalSubject 返回某任务的人工审批决定回传主题。 +func ApprovalSubject(id string) string { return SubjectApproval + "." + id } + +// ApprovalDecision 是一次人工审批的结果(UI → 网关 → dispatcher)。 +type ApprovalDecision struct { + TaskID string `json:"task_id"` + Node string `json:"node,omitempty"` // 审批节点 id(可空:按 task 维度兜底匹配) + Approved bool `json:"approved"` // true=批准放行,false=拒绝中止 + Note string `json:"note,omitempty"` // 审批人备注 + By string `json:"by,omitempty"` // 审批人(用户 id) + TS int64 `json:"ts"` // unix 毫秒 +} + const ( // MetaUserID 是 Task.Meta 中承载已登录用户标识的键(用于偏好记忆召回)。 MetaUserID = "user_id" From d7c6fab93301dac8a98ab354b65e2fb211dc2d2b Mon Sep 17 00:00:00 2001 From: Blizzard Date: Wed, 24 Jun 2026 14:51:37 +0800 Subject: [PATCH 2/9] =?UTF-8?q?fix(hitl):=20=E5=AE=A1=E6=89=B9=E5=BC=B9?= =?UTF-8?q?=E7=AA=97=E5=8F=AF=E9=9D=A0=E5=8C=96=EF=BC=88=E7=8A=B6=E6=80=81?= =?UTF-8?q?=E8=BD=AE=E8=AF=A2=E5=85=9C=E5=BA=95=EF=BC=89+=20=E5=AE=A1?= =?UTF-8?q?=E6=89=B9=E8=8A=82=E7=82=B9=E5=85=A5=E8=B0=83=E8=89=B2=E6=9D=BF?= =?UTF-8?q?=E5=B9=B6=E6=94=B9=E6=A9=99=E8=89=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 两个实测发现的问题: 1. 审批条偶发不弹、任务卡「运行中」:exec 的 await 事件走实时 core NATS(无回放), UI 的 EventSource 晚连一拍就漏掉 → 永远不弹审批。改为轮询持久化的任务状态 (GET /tasks/:id,每 1.5s),status==waiting 即弹审批条,不再依赖易抢跑的 exec 事件; exec 事件仅用于丰富摘要。新增 api.taskStatus + RunState.lifecycle/detail + App.tsx 轮询 + 终态停轮询;BottomDrawer 审批条触发改以 lifecycle==waiting 为准。 2. Studio 左侧调色板没有「人工审批」节点(只加了 NODE_KINDS,漏了 NODE_ORDER), 且其配色与「输入」同为 amber 难以区分:补进 NODE_ORDER(分支与并行之间), 配色改 orange(橙)与输入的 amber(琥珀金)区分。 验证:桌面端 live 重启确认审批条可靠弹出 + 批准/拒绝两条路径正常;左侧节点列表与 画布配色区分清晰。tsc 干净 + vitest 36 过。 Co-Authored-By: Claude Opus 4.8 (1M context) --- sundynix-desktop/frontend/src/App.tsx | 21 ++++++++++++++++++- sundynix-desktop/frontend/src/lib/api.ts | 9 ++++++++ sundynix-desktop/frontend/src/lib/run.ts | 2 ++ .../frontend/src/shell/BottomDrawer.tsx | 6 +++++- .../frontend/src/studio/nodeCatalog.ts | 5 +++-- 5 files changed, 39 insertions(+), 4 deletions(-) diff --git a/sundynix-desktop/frontend/src/App.tsx b/sundynix-desktop/frontend/src/App.tsx index 7171ff2..0c550de 100644 --- a/sundynix-desktop/frontend/src/App.tsx +++ b/sundynix-desktop/frontend/src/App.tsx @@ -14,7 +14,7 @@ import { Placeholder } from "./views/Placeholder"; import { CommandPalette, type Command } from "./components/CommandPalette"; import { UpdateBanner } from "./components/UpdateBanner"; import { Login } from "./views/Login"; -import { submitTask, streamTokens, streamExec, authMe, logout, type Identity, type AuthUser } from "./lib/api"; +import { submitTask, streamTokens, streamExec, taskStatus, authMe, logout, type Identity, type AuthUser } from "./lib/api"; import type { TaskDsl } from "./lib/dsl"; import { emptyRun, type RunState } from "./lib/run"; import { ToastProvider } from "./ui"; @@ -48,6 +48,13 @@ export default function App() { const closeRef = useRef<(() => void) | null>(null); const execCloseRef = useRef<(() => void) | null>(null); + const pollRef = useRef(null); + const stopPoll = () => { + if (pollRef.current != null) { + window.clearInterval(pollRef.current); + pollRef.current = null; + } + }; // 全局 ⌘K / Ctrl+K 唤起命令面板(键盘优先工作站入口)。 useEffect(() => { @@ -98,6 +105,7 @@ export default function App() { async (dsl: TaskDsl) => { closeRef.current?.(); execCloseRef.current?.(); + stopPoll(); const t0 = Date.now(); setRun({ phase: "submitting", output: "", events: [{ t: 0, label: "提交任务" }], exec: [] }); try { @@ -109,6 +117,17 @@ export default function App() { taskId, events: [...r.events, { t: Date.now() - t0, label: `已发布 ${taskId}` }], })); + // 轮询后端任务状态:可靠捕获 waiting(审批中断)—— exec 实时事件会抢跑,状态落 PG 不会丢。 + const terminal = new Set(["done", "failed", "timeout", "rejected"]); + pollRef.current = window.setInterval(async () => { + try { + const s = await taskStatus(taskId); + setRun((r) => (r.taskId === taskId ? { ...r, lifecycle: s.status, detail: s.detail } : r)); + if (terminal.has(s.status)) stopPoll(); + } catch { + /* 忽略瞬时失败,下个 tick 再试 */ + } + }, 1500); // 执行轨迹(运行·观测):与 token 流并行订阅,逐节点点亮。 execCloseRef.current = streamExec( taskId, diff --git a/sundynix-desktop/frontend/src/lib/api.ts b/sundynix-desktop/frontend/src/lib/api.ts index ad0e9e6..ba182ac 100644 --- a/sundynix-desktop/frontend/src/lib/api.ts +++ b/sundynix-desktop/frontend/src/lib/api.ts @@ -109,6 +109,15 @@ export async function submitTask(dsl: TaskDsl, id: Identity): Promise { return data.task_id; } +// taskStatus: GET /api/v1/tasks/:id —— 任务生命周期状态(submitted/running/waiting/done/failed/timeout/rejected)。 +// 轮询它来可靠捕获 waiting(审批中断)—— 不依赖易抢跑的实时 exec 事件。 +export async function taskStatus(taskId: string): Promise<{ status: string; detail: string }> { + const res = guard401(await fetch(`${GATEWAY}/api/v1/tasks/${taskId}`, { headers: bearer() })); + if (!res.ok) throw new Error(`status failed: ${res.status}`); + const d = (await res.json()) as { status?: string; detail?: string }; + return { status: d.status ?? "", detail: d.detail ?? "" }; +} + // approveTask: POST /api/v1/tasks/:id/approve —— HITL 人工审批决定(批准放行 / 拒绝中止)。 export async function approveTask(taskId: string, approved: boolean, opts?: { node?: string; note?: string }): Promise { const res = guard401( diff --git a/sundynix-desktop/frontend/src/lib/run.ts b/sundynix-desktop/frontend/src/lib/run.ts index fbb0211..0acf644 100644 --- a/sundynix-desktop/frontend/src/lib/run.ts +++ b/sundynix-desktop/frontend/src/lib/run.ts @@ -15,6 +15,8 @@ export interface RunState { events: RunEvent[]; exec: ExecEvent[]; // 后端回流的节点级执行轨迹(运行·观测) error?: string; + lifecycle?: string; // 轮询到的后端任务状态(waiting 时弹审批;done/rejected 等终态) + detail?: string; // 状态附带说明(如审批标题) } export const emptyRun: RunState = { phase: "idle", output: "", events: [], exec: [] }; diff --git a/sundynix-desktop/frontend/src/shell/BottomDrawer.tsx b/sundynix-desktop/frontend/src/shell/BottomDrawer.tsx index e79ddb7..4bdea0a 100644 --- a/sundynix-desktop/frontend/src/shell/BottomDrawer.tsx +++ b/sundynix-desktop/frontend/src/shell/BottomDrawer.tsx @@ -27,7 +27,11 @@ export function BottomDrawer({ run }: { run: RunState }) { const statusText = run.phase === "streaming" ? "流式中…" : run.phase === "done" ? "完成 ✓" : run.phase === "error" ? `✗ ${run.error ?? "出错"}` : run.phase === "submitting" ? "提交中…" : "就绪"; - const approval = run.taskId ? pendingApproval(run.exec) : null; + // 审批条触发以「后端状态 waiting」为准(可靠,落 PG);exec 的 await 事件仅用于丰富摘要(可能抢跑丢失)。 + const approval = + run.taskId && run.lifecycle === "waiting" + ? pendingApproval(run.exec) ?? { node: "", title: run.detail || "人工审批", summary: "" } + : null; return (
diff --git a/sundynix-desktop/frontend/src/studio/nodeCatalog.ts b/sundynix-desktop/frontend/src/studio/nodeCatalog.ts index e320146..c280829 100644 --- a/sundynix-desktop/frontend/src/studio/nodeCatalog.ts +++ b/sundynix-desktop/frontend/src/studio/nodeCatalog.ts @@ -105,8 +105,8 @@ export const NODE_KINDS: Record = { approval: { kind: "approval", label: "人工审批", - accent: "border-l-amber-500", - badge: "bg-amber-100 text-amber-700", + accent: "border-l-orange-500", + badge: "bg-orange-100 text-orange-700", desc: "HITL:暂停等人工批准", fields: [ { key: "title", label: "审批标题", type: "text", placeholder: "如:高危操作审批" }, @@ -162,6 +162,7 @@ export const NODE_ORDER = [ "tool", "memory", "branch", + "approval", "map", "aggregate", "render", From e3b0a4f81c03d3661bf3613de78657883aa62575 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Wed, 24 Jun 2026 15:03:00 +0800 Subject: [PATCH 3/9] =?UTF-8?q?perf(dispatcher):=20=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E5=B9=B6=E5=8F=91=E6=B6=88=E8=B4=B9=20=E2=80=94=E2=80=94=20?= =?UTF-8?q?=E5=BE=85=E5=AE=A1/=E6=85=A2=E4=BB=BB=E5=8A=A1=E4=B8=8D?= =?UTF-8?q?=E5=86=8D=E9=98=BB=E5=A1=9E=E5=85=B6=E4=BB=96=E4=BB=BB=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit bus.ConsumeTasks 由单条串行改为限并发分发:每个任务进独立 worker goroutine, 并发上限 = 信号量 + 消费者 MaxAckPending(DISPATCHER_CONCURRENCY,默认 8)。 - 背压:并发满则在 select{sem, ctx.Done} 处等空位;关停时留消息不 ack 待重投。 - 健壮:worker 内 recover panic → Term(避免崩溃循环);ack/nak/span 收口在 worker。 - 并发安全已核:CircuitBreaker 有锁、Orchestrator.turns 有 turnMu、pool RWMutex、 evaluate 本就异步。 效果:一个 HITL 待审任务(Handle 阻塞至多 5min)或长 LLM 生成不再冻结后续任务。 验证:单测 TestConcurrentConsume(A 阻塞时 B 完成);live 实测 HITL 停 waiting 期间 普通任务 3s 跑完且 HITL 不受影响。全模块 build+vet+test 全绿。 Co-Authored-By: Claude Opus 4.8 (1M context) --- sundynix-shared/bus/bus.go | 59 +++++++++++++++++++++------- sundynix-shared/bus/bus_e2e_test.go | 60 +++++++++++++++++++++++++++++ 2 files changed, 105 insertions(+), 14 deletions(-) diff --git a/sundynix-shared/bus/bus.go b/sundynix-shared/bus/bus.go index 7f08bd9..a33b201 100644 --- a/sundynix-shared/bus/bus.go +++ b/sundynix-shared/bus/bus.go @@ -6,6 +6,9 @@ import ( "context" "encoding/json" "fmt" + "log" + "os" + "strconv" "time" "github.com/nats-io/nats.go" @@ -441,9 +444,21 @@ func (b *Bus) SubscribeConfigUpdated(kind string, onUpdate func(*contract.ModelC // TaskHandler 处理一个消费到的任务。 type TaskHandler func(ctx context.Context, t *contract.Task) error +// taskConcurrency 返回任务并发处理上限(env DISPATCHER_CONCURRENCY,默认 8)。 +func taskConcurrency() int { + if v := os.Getenv("DISPATCHER_CONCURRENCY"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + return n + } + } + return 8 +} + // ConsumeTasks 在持久消费者上消费任务,队列组内负载均衡。 -// 返回的 stop 函数用于优雅停止消费。 +// 每个任务分发到独立 worker goroutine 并发执行——一个慢任务/HITL 待审不再阻塞后续任务。 +// 并发上限由信号量 + 消费者 MaxAckPending 双重约束(背压)。返回的 stop 用于优雅停止消费。 func (b *Bus) ConsumeTasks(ctx context.Context, h TaskHandler) (stop func(), err error) { + concurrency := taskConcurrency() cons, err := b.js.CreateOrUpdateConsumer(ctx, contract.StreamTasks, jetstream.ConsumerConfig{ Durable: contract.ConsumerDurable, AckPolicy: jetstream.AckExplicitPolicy, @@ -451,31 +466,47 @@ func (b *Bus) ConsumeTasks(ctx context.Context, h TaskHandler) (stop func(), err // HITL:审批节点会让 Handle 阻塞等人工决定(最长约 5 分钟), // AckWait 必须覆盖「审批等待 + 图执行」总时长,否则消息在途未 ack 会被重投成重复任务。 AckWait: 15 * time.Minute, + // 在途未 ack 上限 = 并发上限:服务端不会下发超过本节点同时能处理的量(背压)。 + MaxAckPending: concurrency, }) if err != nil { return nil, fmt.Errorf("create consumer: %w", err) } + sem := make(chan struct{}, concurrency) // 限并发:最多 N 个任务同时执行 cc, err := cons.Consume(func(msg jetstream.Msg) { t, err := contract.Unmarshal(msg.Data()) if err != nil { _ = msg.Term() // 脏数据,丢弃不重投 return } - // 从消息头还原上游链路,开消费 span(成为 gateway 发布 span 的子节点)。 - mctx := extractTrace(ctx, nats.Header(msg.Headers())) - mctx, span := tracer().Start(mctx, "nats.consume task", - trace.WithSpanKind(trace.SpanKindConsumer), - trace.WithAttributes(attribute.String("sundynix.task_id", t.ID))) - herr := h(mctx, t) - if herr != nil { - span.RecordError(herr) - } - span.End() - if herr != nil { - _ = msg.NakWithDelay(time.Second) // 处理失败,延迟重投 + // 背压:并发已满则在此等空位;正在关停则留消息不 ack(稍后重投)。 + select { + case sem <- struct{}{}: + case <-ctx.Done(): return } - _ = msg.Ack() + go func() { + defer func() { + <-sem // 释放并发额度 + if r := recover(); r != nil { + // 任务处理 panic:丢弃不重投(避免崩溃循环),记录后继续。 + log.Printf("[bus] task %s handler panic: %v", t.ID, r) + _ = msg.Term() + } + }() + // 从消息头还原上游链路,开消费 span(成为 gateway 发布 span 的子节点)。 + mctx := extractTrace(ctx, nats.Header(msg.Headers())) + mctx, span := tracer().Start(mctx, "nats.consume task", + trace.WithSpanKind(trace.SpanKindConsumer), + trace.WithAttributes(attribute.String("sundynix.task_id", t.ID))) + defer span.End() + if herr := h(mctx, t); herr != nil { + span.RecordError(herr) + _ = msg.NakWithDelay(time.Second) // 处理失败,延迟重投 + return + } + _ = msg.Ack() + }() }) if err != nil { return nil, fmt.Errorf("consume: %w", err) diff --git a/sundynix-shared/bus/bus_e2e_test.go b/sundynix-shared/bus/bus_e2e_test.go index bed00f8..80cbcbe 100644 --- a/sundynix-shared/bus/bus_e2e_test.go +++ b/sundynix-shared/bus/bus_e2e_test.go @@ -194,3 +194,63 @@ func TestTokenStreamRoundTrip(t *testing.T) { t.Fatal("timeout: 未收到流结束信号") } } + +// TestConcurrentConsume 验证并发消费:一个长时间阻塞的任务(模拟 HITL 待审) +// 不应阻塞后续任务——旧的串行消费下本测试会超时失败。 +func TestConcurrentConsume(t *testing.T) { + url := startEmbeddedNATS(t) + + gw, err := bus.Connect(url) + if err != nil { + t.Fatalf("gateway connect: %v", err) + } + defer gw.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + if err := gw.EnsureTaskStream(ctx); err != nil { + t.Fatalf("ensure stream: %v", err) + } + + dp, err := bus.Connect(url) + if err != nil { + t.Fatalf("dispatcher connect: %v", err) + } + defer dp.Close() + + blockA := make(chan struct{}) + doneB := make(chan string, 1) + stop, err := dp.ConsumeTasks(ctx, func(_ context.Context, task *contract.Task) error { + switch task.ID { + case "A": + <-blockA // 模拟审批长阻塞 + case "B": + doneB <- task.ID + } + return nil + }) + if err != nil { + t.Fatalf("consume: %v", err) + } + defer stop() + defer close(blockA) + + // 先发 A 并等它被消费、卡在 handler 里;再发 B。 + if _, err := gw.PublishTask(ctx, &contract.Task{ID: "A", Graph: json.RawMessage(`{}`)}); err != nil { + t.Fatalf("publish A: %v", err) + } + time.Sleep(400 * time.Millisecond) + if _, err := gw.PublishTask(ctx, &contract.Task{ID: "B", Graph: json.RawMessage(`{}`)}); err != nil { + t.Fatalf("publish B: %v", err) + } + + select { + case id := <-doneB: + if id != "B" { + t.Fatalf("unexpected task done: %q", id) + } + t.Log("✓ 并发消费生效:A 仍阻塞时 B 已完成") + case <-time.After(4 * time.Second): + t.Fatal("B 被阻塞的 A 卡住——并发消费未生效(仍是串行)") + } +} From 53669437d58be705716d733c7a5030af6d382d0d Mon Sep 17 00:00:00 2001 From: Blizzard Date: Wed, 24 Jun 2026 15:40:50 +0800 Subject: [PATCH 4/9] =?UTF-8?q?perf(tools):=20=E5=B7=A5=E5=85=B7=E6=9C=8D?= =?UTF-8?q?=E5=8A=A1=E5=8D=95=E5=AE=9E=E4=BE=8B=E5=B9=B6=E5=8F=91=EF=BC=88?= =?UTF-8?q?ServeTool=20=E5=8D=8F=E7=A8=8B=E5=8C=96=EF=BC=89+=20mcp-py=20?= =?UTF-8?q?=E5=90=8C=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 垂直扩:单个 mcp 实例内每个工具调用分发到独立 goroutine/task 并发处理, 不再受 NATS 单订阅回调串行所限;配合队列组多副本即「单实例并发 × 副本数」横向扩。 - bus.ServeTool:回调立即起 goroutine 并返回(不阻塞 NATS 投递),信号量限并发 (MCP_TOOL_CONCURRENCY,默认 16),handler panic → 回错误结果避免调用方干等超时。 - mcp-py mcp_gateway:_on_call 改 asyncio.create_task 派发 + Semaphore 限并发(同默认 16)。 - 测试 TestConcurrentToolServe:slow 工具阻塞时 fast 工具仍返回(旧串行下会超时)。 验证:单测通过;mcp-go/mcp-py live 重启工具就绪,工具往返正常(memory/history/kb_search 均正常响应,无 panic)。全模块 build+vet+test 全绿。 注:下游共享后端(Milvus/Neo4j/PG/嵌入端点)是横向扩的最终上限,届时扩这些而非 mcp 实例。 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../src/sundynix_mcp_py/mcp_gateway.py | 45 ++++++++------ sundynix-shared/bus/bus.go | 59 +++++++++++++------ sundynix-shared/bus/bus_e2e_test.go | 52 ++++++++++++++++ 3 files changed, 118 insertions(+), 38 deletions(-) diff --git a/sundynix-mcp-py/src/sundynix_mcp_py/mcp_gateway.py b/sundynix-mcp-py/src/sundynix_mcp_py/mcp_gateway.py index e4be237..0801dff 100644 --- a/sundynix-mcp-py/src/sundynix_mcp_py/mcp_gateway.py +++ b/sundynix-mcp-py/src/sundynix_mcp_py/mcp_gateway.py @@ -47,6 +47,8 @@ class McpGateway: self.interpreter = CodeInterpreter() # Docker 隔离沙箱 self._nc: nats.NATS | None = None self._tools: dict[str, callable] = {} + # 单实例工具并发上限(垂直扩);队列组多副本再叠加水平扩。 + self._sem = asyncio.Semaphore(int(os.getenv("MCP_TOOL_CONCURRENCY", "16"))) async def register_tools(self) -> None: """注册工具名 → 处理协程。""" @@ -77,25 +79,30 @@ class McpGateway: await asyncio.Event().wait() async def _on_call(self, msg) -> None: - """解析 ToolCall → 路由到工具 → Respond ToolResult。""" - try: - req = json.loads(msg.data) - except Exception as e: # noqa: BLE001 - await self._reply(msg, ok=False, error=f"bad tool call: {e}") - return - tool = req.get("tool", "") - task_id = req.get("task_id", "") - args = req.get("args") or {} - log.info("[mcp_py] tool=%s task=%s args=%s", tool, task_id, args) - fn = self._tools.get(tool) - if fn is None: - await self._reply(msg, ok=False, error=f"unknown tool: {tool}") - return - try: - content = await fn(args) - await self._reply(msg, ok=True, content=content) - except Exception as e: # noqa: BLE001 - await self._reply(msg, ok=False, error=str(e)) + """立即派发到独立 task 并发处理,不阻塞订阅消息循环(nats-py 单订阅回调本是串行 await)。""" + asyncio.create_task(self._handle(msg)) + + async def _handle(self, msg) -> None: + """解析 ToolCall → 路由到工具 → Respond ToolResult(限并发)。""" + async with self._sem: + try: + req = json.loads(msg.data) + except Exception as e: # noqa: BLE001 + await self._reply(msg, ok=False, error=f"bad tool call: {e}") + return + tool = req.get("tool", "") + task_id = req.get("task_id", "") + args = req.get("args") or {} + log.info("[mcp_py] tool=%s task=%s args=%s", tool, task_id, args) + fn = self._tools.get(tool) + if fn is None: + await self._reply(msg, ok=False, error=f"unknown tool: {tool}") + return + try: + content = await fn(args) + await self._reply(msg, ok=True, content=content) + except Exception as e: # noqa: BLE001 + await self._reply(msg, ok=False, error=str(e)) @staticmethod async def _reply(msg, *, ok: bool, content: str = "", error: str = "") -> None: diff --git a/sundynix-shared/bus/bus.go b/sundynix-shared/bus/bus.go index a33b201..5b0e1ad 100644 --- a/sundynix-shared/bus/bus.go +++ b/sundynix-shared/bus/bus.go @@ -233,27 +233,48 @@ func (b *Bus) CallTool(ctx context.Context, subject string, call *contract.ToolC // ToolHandler 处理一次工具调用并返回结果。 type ToolHandler func(ctx context.Context, call *contract.ToolCall) *contract.ToolResult -// ServeTool 以队列组订阅工具主题(可用通配 sundynix.tools.go.>), -// 对每个请求调用 h 并 Respond,队列组内多副本自动负载均衡。 -// 返回的 unsub 用于退订。 +// toolConcurrency 返回单实例工具并发处理上限(env MCP_TOOL_CONCURRENCY,默认 16)。 +func toolConcurrency() int { + if v := os.Getenv("MCP_TOOL_CONCURRENCY"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + return n + } + } + return 16 +} + +// ServeTool 以队列组订阅工具主题(可用通配 sundynix.tools.go.>),队列组内多副本自动负载均衡(水平扩)。 +// 单实例内每个请求分发到独立 goroutine 并发处理(垂直扩,上限 MCP_TOOL_CONCURRENCY)—— +// 回调立即返回不阻塞 NATS 投递;信号量约束同时执行的 handler 数。返回的 unsub 用于退订。 func (b *Bus) ServeTool(subject, queue string, h ToolHandler) (unsub func() error, err error) { + sem := make(chan struct{}, toolConcurrency()) sub, err := b.nc.QueueSubscribe(subject, queue, func(m *nats.Msg) { - var call contract.ToolCall - if err := json.Unmarshal(m.Data, &call); err != nil { - respond(m, &contract.ToolResult{OK: false, Error: "bad tool call: " + err.Error()}) - return - } - // 还原上游链路并开服务端 span(成为 dispatcher tool.call span 的子节点)。 - mctx := extractTrace(context.Background(), m.Header) - mctx, span := tracer().Start(mctx, "tool.serve "+call.Tool, - trace.WithSpanKind(trace.SpanKindServer), - trace.WithAttributes(attribute.String("sundynix.tool", call.Tool))) - res := h(mctx, &call) - if res != nil { - span.SetAttributes(attribute.Bool("sundynix.tool.ok", res.OK)) - } - span.End() - respond(m, res) + // 立即起 goroutine 并返回,让 NATS 继续投递下一条(核心 NATS 单订阅回调本是串行)。 + go func() { + sem <- struct{}{} // 限并发:超额则在此短暂排队 + defer func() { + <-sem + if r := recover(); r != nil { // handler panic → 回错误结果,调用方不必干等超时 + respond(m, &contract.ToolResult{OK: false, Error: fmt.Sprintf("tool panic: %v", r)}) + } + }() + var call contract.ToolCall + if err := json.Unmarshal(m.Data, &call); err != nil { + respond(m, &contract.ToolResult{OK: false, Error: "bad tool call: " + err.Error()}) + return + } + // 还原上游链路并开服务端 span(成为 dispatcher tool.call span 的子节点)。 + mctx := extractTrace(context.Background(), m.Header) + mctx, span := tracer().Start(mctx, "tool.serve "+call.Tool, + trace.WithSpanKind(trace.SpanKindServer), + trace.WithAttributes(attribute.String("sundynix.tool", call.Tool))) + defer span.End() + res := h(mctx, &call) + if res != nil { + span.SetAttributes(attribute.Bool("sundynix.tool.ok", res.OK)) + } + respond(m, res) + }() }) if err != nil { return nil, fmt.Errorf("serve tool %s: %w", subject, err) diff --git a/sundynix-shared/bus/bus_e2e_test.go b/sundynix-shared/bus/bus_e2e_test.go index 80cbcbe..c5f6bde 100644 --- a/sundynix-shared/bus/bus_e2e_test.go +++ b/sundynix-shared/bus/bus_e2e_test.go @@ -254,3 +254,55 @@ func TestConcurrentConsume(t *testing.T) { t.Fatal("B 被阻塞的 A 卡住——并发消费未生效(仍是串行)") } } + +// TestConcurrentToolServe 验证单实例工具服务并发:一个慢工具调用不阻塞另一个快调用。 +// 旧的串行 ServeTool 下本测试会超时失败。 +func TestConcurrentToolServe(t *testing.T) { + url := startEmbeddedNATS(t) + + srv, err := bus.Connect(url) + if err != nil { + t.Fatalf("server connect: %v", err) + } + defer srv.Close() + + blockSlow := make(chan struct{}) + unsub, err := srv.ServeTool(contract.SubjectToolsGoAll, contract.QueueToolsGo, + func(_ context.Context, call *contract.ToolCall) *contract.ToolResult { + if call.Tool == "slow" { + <-blockSlow // 模拟慢工具(如 RAG 嵌入) + } + return &contract.ToolResult{OK: true, Content: call.Tool} + }) + if err != nil { + t.Fatalf("serve tool: %v", err) + } + defer unsub() + defer close(blockSlow) + + cli, err := bus.Connect(url) + if err != nil { + t.Fatalf("client connect: %v", err) + } + defer cli.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + // 先发 slow(会阻塞在 handler 里),再发 fast——fast 应在 slow 仍阻塞时返回。 + go func() { + _, _ = cli.CallTool(ctx, contract.ToolSubjectGo("slow"), &contract.ToolCall{Tool: "slow"}) + }() + time.Sleep(300 * time.Millisecond) + + fctx, fcancel := context.WithTimeout(ctx, 3*time.Second) + defer fcancel() + res, err := cli.CallTool(fctx, contract.ToolSubjectGo("fast"), &contract.ToolCall{Tool: "fast"}) + if err != nil { + t.Fatalf("fast 工具被慢工具阻塞——单实例并发未生效: %v", err) + } + if res.Content != "fast" { + t.Fatalf("unexpected result: %q", res.Content) + } + t.Log("✓ 单实例工具并发生效:slow 仍阻塞时 fast 已返回") +} From 38a74e39110287ce2a27a0d2c87ef7bfce164a6d Mon Sep 17 00:00:00 2001 From: Blizzard Date: Wed, 24 Jun 2026 15:53:30 +0800 Subject: [PATCH 5/9] =?UTF-8?q?fix(history):=20=E7=A9=BA=E7=AD=94=E5=A4=8D?= =?UTF-8?q?=E4=B8=8D=E8=90=BD=E5=8E=86=E5=8F=B2=20+=20=E5=8F=91=E9=80=81?= =?UTF-8?q?=E5=89=8D=E8=BF=87=E6=BB=A4=E7=A9=BA=E6=B6=88=E6=81=AF=EF=BC=88?= =?UTF-8?q?=E6=A0=B9=E6=B2=BB=E4=BC=9A=E8=AF=9D=E6=AF=92=E5=8C=96=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 现象:某轮 LLM 返回空(余额不足/失败/降级)→ 空 assistant 消息被写进会话历史 → 此后该 session 每次请求都带一条空 content 的 assistant 消息 → DeepSeek 400 "Invalid assistant message: content or tool_calls must be set" → 又空 → 又写空 → 自我循环毒化。 - Fix A(断源):orchestrator.memorize 对空答复直接 return,不落历史(失败的一轮不留痕)。 - Fix B(兜底):buildMessages 发送前剔除 content 与 tool_calls 均空的历史消息, 防御任何来源的脏历史(含存量)。 验证:清理被污染的 default 会话后恢复正常;新逻辑下空答复不再写入。dispatcher build+vet+test 全绿。 注:诊断中发现激活模型 deepseek-v4-pro 是推理模型(答案在 reasoning_content,content 为空), 已在管理端切回 deepseek-chat。仍待办:LLM 调用失败应暴露为 failed 而非 done-空输出。 Co-Authored-By: Claude Opus 4.8 (1M context) --- sundynix-dispatcher/internal/eino/compile.go | 11 ++++++++++- sundynix-dispatcher/internal/eino/orchestrator.go | 6 ++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/sundynix-dispatcher/internal/eino/compile.go b/sundynix-dispatcher/internal/eino/compile.go index 82527d2..3659104 100644 --- a/sundynix-dispatcher/internal/eino/compile.go +++ b/sundynix-dispatcher/internal/eino/compile.go @@ -48,9 +48,18 @@ func buildMessages(ctx context.Context, rc *RunCtx) ([]*schema.Message, error) { sys.WriteString("\n\n以下是前序协作 agent 的产出,请在此基础上继续完成你的部分(不要重头再来):\n") sys.WriteString(strings.Join(rc.Upstream, "\n---\n")) } + // 防御:剔除空 content 的历史消息。OpenAI 兼容 API 拒绝「content 与 tool_calls 都为空」的 + // assistant 消息(400 Invalid assistant message),一条脏历史会毒化整段会话的后续请求。 + history := make([]*schema.Message, 0, len(rc.History)) + for _, m := range rc.History { + if m == nil || (strings.TrimSpace(m.Content) == "" && len(m.ToolCalls) == 0) { + continue + } + history = append(history, m) + } return chatTemplate.Format(ctx, map[string]any{ "system": sys.String(), - "history": rc.History, + "history": history, "query": rc.Query, }) } diff --git a/sundynix-dispatcher/internal/eino/orchestrator.go b/sundynix-dispatcher/internal/eino/orchestrator.go index 4019d4b..0727535 100644 --- a/sundynix-dispatcher/internal/eino/orchestrator.go +++ b/sundynix-dispatcher/internal/eino/orchestrator.go @@ -8,6 +8,7 @@ import ( "fmt" "log" "log/slog" + "strings" "sync" "time" @@ -270,6 +271,11 @@ const consolidateEveryTurns = 3 // memorize 写回阶段(异步、离热路径):落短期历史;每 N 轮做一次记忆对账(consolidate)。 func (o *Orchestrator) memorize(t *contract.Task, answer string) { + // 空答复(LLM 失败/降级/拒绝)不落历史:空 assistant 消息会被 LLM API 拒绝(400), + // 一旦写入会毒化该会话后续所有请求。失败的一轮干脆不留痕。 + if strings.TrimSpace(answer) == "" { + return + } uid, _ := t.Meta[contract.MetaUserID].(string) sid, _ := t.Meta[contract.MetaSessionID].(string) if sid != "" && o.tools != nil { From 700845d64ad071d316a171ea5e8cab4459345f01 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Wed, 24 Jun 2026 16:14:19 +0800 Subject: [PATCH 6/9] =?UTF-8?q?feat(prod):=20DB=20=E8=BF=9E=E6=8E=A5?= =?UTF-8?q?=E6=B1=A0=E4=B8=8A=E9=99=90=20+=20LLM=20=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E6=9A=B4=E9=9C=B2=E4=B8=BA=20failed=EF=BC=88=E7=94=9F=E4=BA=A7?= =?UTF-8?q?=E7=BA=A7=E5=B9=B6=E5=8F=91=E6=94=B6=E5=B0=BE=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为高并发生产做的三项收尾(配合已有的任务/工具并发消费): 1. DB 连接池上限(pgsql.go / memory/store.go):SetMaxOpenConns(默认 25, DB_MAX_OPEN_CONNS 可调)+ MaxIdleConns 5 + ConnMaxLifetime 1h。 防高并发无限开连接打爆 PG(max_connections 默认 100)。 2. LLM 失败暴露为 failed(graph.go):board.fatalErr —— agent 模型调用出错即上抛, runGraph 中止后续节点并返回错误 → Handle 判 failed(带原因),不再静默 done-空。 可观测/可告警,生产排障必需。 压测验证(dispatcher 并发=50, 池=25, deepseek-v4-pro 推理): - 平台同一秒并发收下 40 任务,全程零 DB/连接错误,平台开销≈0(裸 LLM 1.8s vs 平台 P50 1.7s)。 - 并发 10 健康 4.7/s;20+ 延迟暴涨 = DeepSeek 开发账号并发限流(外部),非平台。 - 失败注入(错误模型名)→ 任务正确判 failed 并回传原因。 结论:平台并发机制达生产级;真实吞吐上限 = 自托管模型容量(生产 Qwen,可加卡线性扩)。 Co-Authored-By: Claude Opus 4.8 (1M context) --- sundynix-dispatcher/internal/eino/graph.go | 18 +++++++++++++--- sundynix-gateway/internal/store/pgsql.go | 24 ++++++++++++++++++++++ sundynix-mcp-go/internal/memory/store.go | 14 +++++++++++++ 3 files changed, 53 insertions(+), 3 deletions(-) diff --git a/sundynix-dispatcher/internal/eino/graph.go b/sundynix-dispatcher/internal/eino/graph.go index 3729e0f..97533f0 100644 --- a/sundynix-dispatcher/internal/eino/graph.go +++ b/sundynix-dispatcher/internal/eino/graph.go @@ -33,6 +33,7 @@ type board struct { answer string // 当前成稿(多 agent 协作时 = 最近一个 agent 的产出 = 成品) agentOut []string // 各上游 agent 的产出(按序),注入下游 agent 上下文以实现接力协作 rejected bool // HITL 审批节点拒绝/超时 → 置位,runGraph 中止并返回 errRejected + fatalErr error // agent 节点 LLM 调用失败 → 置位,runGraph 中止并上抛 → 任务判 failed(而非 done-空) } // runGraph 按 DSL 图的真实拓扑与连线执行(替代旧的线性拍平 compileFlow)。 @@ -56,7 +57,7 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT b.profile = o.fetchMemory(ctx, b.uid, b.query) b.history = o.fetchHistory(ctx, b.sid) o.runConversation(ctx, t.ID, b, plan.System, tr, "agent") - return b.answer, nil + return b.answer, b.fatalErr // 模型失败 → 上抛判 failed } // 建邻接与入度(只认两端都存在的边)。保留整条边以便 branch 按 true/false 标签选路。 @@ -155,8 +156,8 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT tr.info(n.Kind+":"+n.ID, "system", labelOf(n, n.Kind), "未识别节点,跳过") } nspan.End() - if b.rejected { - break // 审批拒绝/超时:中止后续节点,整图判 rejected + if b.rejected || b.fatalErr != nil { + break // 审批拒绝/超时 或 模型失败:中止后续节点 } for _, tgt := range propagate { active[tgt] = true @@ -166,11 +167,17 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT if b.rejected { return b.answer, errRejected // 合法终态,Handle 据此判 rejected 并优雅收尾 } + if b.fatalErr != nil { + return b.answer, b.fatalErr // 上抛 → Handle 判 failed(带原因) + } // 图里无 agent 节点(纯工具/检索图)也要出一段模型答复,否则没有输出。 if b.answer == "" { o.runConversation(ctx, t.ID, b, plan.System, tr, "agent") } + if b.fatalErr != nil { + return b.answer, b.fatalErr + } return b.answer, nil } @@ -283,6 +290,11 @@ func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, sy } if err != nil { tr.emit(node, "model", "error", "模型流式推理", err.Error(), time.Since(t0).Milliseconds()) + // 未产出任何 token 即失败 → 标记致命错,让任务判 failed(暴露原因,便于监控告警), + // 而非静默 done-空。已流出部分 token 的中断也算失败(结果不完整)。 + if b.fatalErr == nil { + b.fatalErr = fmt.Errorf("agent 模型推理失败: %w", err) + } return } if redacted > 0 { diff --git a/sundynix-gateway/internal/store/pgsql.go b/sundynix-gateway/internal/store/pgsql.go index 8b011f7..c4344cc 100644 --- a/sundynix-gateway/internal/store/pgsql.go +++ b/sundynix-gateway/internal/store/pgsql.go @@ -5,6 +5,9 @@ import ( "context" "errors" "log" + "os" + "strconv" + "time" "gorm.io/driver/postgres" "gorm.io/gorm" @@ -13,6 +16,26 @@ import ( "github.com/sundynix/sundynix-shared/contract" ) +// envInt 读正整数环境变量,缺省回退 def。 +func envInt(key string, def int) int { + if v := os.Getenv(key); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + return n + } + } + return def +} + +// tunePool 给连接池设上限:高并发下不至于无限开连接打爆 PG(max_connections 默认 100)。 +// 各服务默认 25,可经 DB_MAX_OPEN_CONNS / DB_MAX_IDLE_CONNS 调整。 +func tunePool(db *gorm.DB) { + if sqlDB, err := db.DB(); err == nil { + sqlDB.SetMaxOpenConns(envInt("DB_MAX_OPEN_CONNS", 25)) + sqlDB.SetMaxIdleConns(envInt("DB_MAX_IDLE_CONNS", 5)) + sqlDB.SetConnMaxLifetime(time.Hour) + } +} + // errStoreDisabled 表示 Postgres 处于降级(未连接)模式,写操作无法进行。 var errStoreDisabled = errors.New("postgres store disabled") @@ -36,6 +59,7 @@ func OpenPostgres(dsn string) *Postgres { log.Printf("[store] postgres 不可用,降级运行(不持久化): %v", err) return &Postgres{} } + tunePool(db) // 连接池上限,防高并发打爆 PG // 一次性迁移:旧表用整型自增 id,与新雪花字符串 id 不兼容(AutoMigrate 不改主键类型)。 // 备份模型密钥(唯一不可再生的数据) → 重建全部表 → 回灌模型。其余为可重建的测试数据。 migrateLegacyIntIDs(db) diff --git a/sundynix-mcp-go/internal/memory/store.go b/sundynix-mcp-go/internal/memory/store.go index 944bd7a..05d8ceb 100644 --- a/sundynix-mcp-go/internal/memory/store.go +++ b/sundynix-mcp-go/internal/memory/store.go @@ -7,7 +7,9 @@ import ( "fmt" "log" "math" + "os" "sort" + "strconv" "strings" "time" @@ -69,6 +71,18 @@ func Open(dsn string) *Store { log.Printf("[memory] postgres 不可用,记忆降级(召回为空): %v", err) return &Store{} } + // 连接池上限,防高并发打爆 PG(max_connections 默认 100);默认 25,可经 DB_MAX_OPEN_CONNS 调。 + if sqlDB, derr := db.DB(); derr == nil { + maxOpen := 25 + if v := os.Getenv("DB_MAX_OPEN_CONNS"); v != "" { + if n, perr := strconv.Atoi(v); perr == nil && n > 0 { + maxOpen = n + } + } + sqlDB.SetMaxOpenConns(maxOpen) + sqlDB.SetMaxIdleConns(5) + sqlDB.SetConnMaxLifetime(time.Hour) + } // 一次性迁移:旧表用复合主键 (user_id,key) 无 id/时间戳,与雪花规约不兼容。 migrateLegacyProfile(db) if err := db.AutoMigrate(&Profile{}); err != nil { From e503965a83f76474a2998e29d7fdaebfcc023a15 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Wed, 24 Jun 2026 16:47:15 +0800 Subject: [PATCH 7/9] =?UTF-8?q?feat(tools):=20=E6=96=B0=E5=A2=9E=204=20?= =?UTF-8?q?=E4=B8=AA=20agent=20=E5=B7=A5=E5=85=B7=20=E2=80=94=E2=80=94=20w?= =?UTF-8?q?eb=5Fsearch=20/=20web=5Ffetch=20/=20calculator=20/=20current=5F?= =?UTF-8?q?datetime?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 均在 mcp-go,注册即 agent 可见(dispatcher 经 list_tools 动态发现,零改调度代码): - web_search 联网搜索:Tavily(有 TAVILY_API_KEY 则用,干净 JSON)/ DuckDuckGo HTML(免 key 兜底)。 - web_fetch 网页抓取:取 URL → 去脚本样式 → 去标签 → 解实体 → 收敛空白,限 8000 rune; 复用 external_api 的 SSRF 防护(拒环回/内网/元数据 + 重定向校验)。 - calculator 计算器:自研调度场算法求值(+ - * / % ^ 括号、一元负号),杜绝任意代码执行。 - current_datetime 当前时间:含星期与时区(可传 tz)。 测试:单测覆盖 evalArith(含右结合/一元负号/错误用例)、htmlToText(剔脚本样式+解实体)、 ddgUnwrap。live 自主 agent 实测:calculator (123+456)*7→4053、current_datetime 周三、 web_fetch example.com→200、web_search DDG 返回真实结果,全部端到端通过。 Co-Authored-By: Claude Opus 4.8 (1M context) --- sundynix-mcp-go/internal/mcp/gateway.go | 37 +- sundynix-mcp-go/internal/mcp/tools_extra.go | 406 ++++++++++++++++++ .../internal/mcp/tools_extra_test.go | 58 +++ 3 files changed, 496 insertions(+), 5 deletions(-) create mode 100644 sundynix-mcp-go/internal/mcp/tools_extra.go create mode 100644 sundynix-mcp-go/internal/mcp/tools_extra_test.go diff --git a/sundynix-mcp-go/internal/mcp/gateway.go b/sundynix-mcp-go/internal/mcp/gateway.go index 4e758a6..b696a58 100644 --- a/sundynix-mcp-go/internal/mcp/gateway.go +++ b/sundynix-mcp-go/internal/mcp/gateway.go @@ -81,7 +81,7 @@ func (g *Gateway) buildRegistry() map[string]toolDef { // —— 暴露给自主 agent 的工具(带参数 schema / 注入声明)—— "wiki_search": { cn: "知识检索", desc: "检索知识库,返回与查询最相关的资料片段。需要外部知识/事实依据时调用。", - agent: true, + agent: true, params: []paramSpec{{Name: "q", Type: "string", Desc: "检索查询语句", Required: true}}, inject: []string{"kb"}, handler: g.wikiSearch, }, @@ -102,6 +102,33 @@ func (g *Gateway) buildRegistry() map[string]toolDef { cn: "历史召回", desc: "取当前会话最近多轮对话,用于理解上下文。", agent: true, inject: []string{"session_id"}, handler: g.historyGet, }, + "web_search": { + cn: "联网搜索", desc: "联网搜索,返回最新网页结果(标题/链接/摘要)。需要实时/最新信息或外部事实时调用。", + agent: true, + params: []paramSpec{ + {Name: "q", Type: "string", Desc: "搜索关键词", Required: true}, + {Name: "topK", Type: "integer", Desc: "返回结果条数(默认 5,最多 10)"}, + }, + handler: g.webSearch, + }, + "web_fetch": { + cn: "网页抓取", desc: "抓取一个网页 URL 并提取正文文本。需要读取某个链接的内容时调用。", + agent: true, + params: []paramSpec{{Name: "url", Type: "string", Desc: "要抓取的网页 URL", Required: true}}, + handler: g.webFetch, + }, + "calculator": { + cn: "计算器", desc: "精确计算数学表达式(+ - * / % ^ 与括号)。涉及算术/数值计算时调用,不要心算。", + agent: true, + params: []paramSpec{{Name: "expr", Type: "string", Desc: "数学表达式,如 (3+4)*2^3", Required: true}}, + handler: g.calculator, + }, + "current_datetime": { + cn: "当前时间", desc: "获取当前日期与时间(含星期)。需要“现在/今天几号/星期几”等时间信息时调用。", + agent: true, + params: []paramSpec{{Name: "tz", Type: "string", Desc: "可选时区,如 Asia/Shanghai;缺省服务器本地时区"}}, + handler: g.currentDatetime, + }, // —— 仅内部/流水线/管理用,不暴露给自主 agent —— "kb_ingest": {cn: "知识入库", desc: "文本切块 → 向量化 → 写入 Milvus / Bleve", handler: g.kbIngest}, @@ -147,10 +174,10 @@ func (g *Gateway) listTools() *contract.ToolResult { Name string `json:"name"` CN string `json:"cn"` Desc string `json:"desc"` - Agent bool `json:"agent_exposed"` // 是否给自主 agent - AgentName string `json:"agent_name,omitempty"`// 模型可见名(空=name) - Params []paramSpec `json:"params,omitempty"` // 模型可填参数 - Inject []string `json:"inject,omitempty"` // 服务端注入参数(不暴露给模型) + Agent bool `json:"agent_exposed"` // 是否给自主 agent + AgentName string `json:"agent_name,omitempty"` // 模型可见名(空=name) + Params []paramSpec `json:"params,omitempty"` // 模型可填参数 + Inject []string `json:"inject,omitempty"` // 服务端注入参数(不暴露给模型) } out := make([]info, 0, len(g.tools)) for name, td := range g.tools { diff --git a/sundynix-mcp-go/internal/mcp/tools_extra.go b/sundynix-mcp-go/internal/mcp/tools_extra.go new file mode 100644 index 0000000..4693ff6 --- /dev/null +++ b/sundynix-mcp-go/internal/mcp/tools_extra.go @@ -0,0 +1,406 @@ +package mcp + +import ( + "context" + "fmt" + "html" + "io" + "math" + "net/http" + "net/url" + "os" + "regexp" + "strconv" + "strings" + "time" + + "github.com/sundynix/sundynix-shared/contract" +) + +// ===== current_datetime:当前日期时间(LLM 不知道"现在")===== + +var weekdaysCN = []string{"周日", "周一", "周二", "周三", "周四", "周五", "周六"} + +func (g *Gateway) currentDatetime(_ context.Context, call *contract.ToolCall) *contract.ToolResult { + loc := time.Local + if tz, _ := call.Args["tz"].(string); strings.TrimSpace(tz) != "" { + if l, err := time.LoadLocation(strings.TrimSpace(tz)); err == nil { + loc = l + } + } + now := time.Now().In(loc) + out := fmt.Sprintf("%s %s(%s,时区 %s,Unix %d)", + now.Format("2006-01-02"), now.Format("15:04:05"), + weekdaysCN[int(now.Weekday())], now.Format("MST-07:00"), now.Unix()) + return &contract.ToolResult{OK: true, Content: out} +} + +// ===== calculator:安全表达式求值(+ - * / % ^ 括号、一元负号)===== + +func (g *Gateway) calculator(_ context.Context, call *contract.ToolCall) *contract.ToolResult { + expr := strings.TrimSpace(fmt.Sprint(call.Args["expr"])) + if expr == "" || expr == "" { + return &contract.ToolResult{OK: false, Error: "calculator: expr 必填"} + } + v, err := evalArith(expr) + if err != nil { + return &contract.ToolResult{OK: false, Error: "calculator: " + err.Error()} + } + return &contract.ToolResult{OK: true, Content: strconv.FormatFloat(v, 'g', -1, 64)} +} + +// evalArith 用调度场算法把中缀表达式转 RPN 再求值。仅支持数值与 + - * / % ^ ( ),杜绝任意代码执行。 +func evalArith(s string) (float64, error) { + toks, err := tokenizeArith(s) + if err != nil { + return 0, err + } + prec := map[string]int{"+": 1, "-": 1, "*": 2, "/": 2, "%": 2, "^": 3, "u-": 4} + rightAssoc := map[string]bool{"^": true, "u-": true} + var output, ops []string + for i, t := range toks { + switch { + case isNumber(t): + output = append(output, t) + case t == "(": + ops = append(ops, t) + case t == ")": + for len(ops) > 0 && ops[len(ops)-1] != "(" { + output = append(output, ops[len(ops)-1]) + ops = ops[:len(ops)-1] + } + if len(ops) == 0 { + return 0, fmt.Errorf("括号不匹配") + } + ops = ops[:len(ops)-1] // 弹出 "(" + default: // 运算符 + op := t + // 一元负号:在表达式开头或运算符/左括号之后的 "-"。 + if op == "-" && (i == 0 || isOperator(toks[i-1]) || toks[i-1] == "(") { + op = "u-" + } + for len(ops) > 0 { + top := ops[len(ops)-1] + if top == "(" { + break + } + if prec[top] > prec[op] || (prec[top] == prec[op] && !rightAssoc[op]) { + output = append(output, top) + ops = ops[:len(ops)-1] + } else { + break + } + } + ops = append(ops, op) + } + } + for len(ops) > 0 { + if ops[len(ops)-1] == "(" { + return 0, fmt.Errorf("括号不匹配") + } + output = append(output, ops[len(ops)-1]) + ops = ops[:len(ops)-1] + } + return evalRPN(output) +} + +func evalRPN(rpn []string) (float64, error) { + var st []float64 + pop := func() (float64, error) { + if len(st) == 0 { + return 0, fmt.Errorf("表达式非法") + } + v := st[len(st)-1] + st = st[:len(st)-1] + return v, nil + } + for _, t := range rpn { + if isNumber(t) { + f, _ := strconv.ParseFloat(t, 64) + st = append(st, f) + continue + } + if t == "u-" { + a, err := pop() + if err != nil { + return 0, err + } + st = append(st, -a) + continue + } + b, err := pop() + if err != nil { + return 0, err + } + a, err := pop() + if err != nil { + return 0, err + } + switch t { + case "+": + st = append(st, a+b) + case "-": + st = append(st, a-b) + case "*": + st = append(st, a*b) + case "/": + if b == 0 { + return 0, fmt.Errorf("除以零") + } + st = append(st, a/b) + case "%": + st = append(st, math.Mod(a, b)) + case "^": + st = append(st, math.Pow(a, b)) + default: + return 0, fmt.Errorf("未知运算符 %q", t) + } + } + if len(st) != 1 { + return 0, fmt.Errorf("表达式非法") + } + return st[0], nil +} + +func tokenizeArith(s string) ([]string, error) { + var toks []string + r := []rune(s) + for i := 0; i < len(r); { + c := r[i] + switch { + case c == ' ' || c == '\t': + i++ + case strings.ContainsRune("+-*/%^()", c): + toks = append(toks, string(c)) + i++ + case (c >= '0' && c <= '9') || c == '.': + j := i + for j < len(r) && ((r[j] >= '0' && r[j] <= '9') || r[j] == '.' || r[j] == 'e' || r[j] == 'E' || + ((r[j] == '+' || r[j] == '-') && j > i && (r[j-1] == 'e' || r[j-1] == 'E'))) { + j++ + } + num := string(r[i:j]) + if _, err := strconv.ParseFloat(num, 64); err != nil { + return nil, fmt.Errorf("非法数字 %q", num) + } + toks = append(toks, num) + i = j + default: + return nil, fmt.Errorf("非法字符 %q(仅支持数字与 + - * / %% ^ ( ))", string(c)) + } + } + if len(toks) == 0 { + return nil, fmt.Errorf("空表达式") + } + return toks, nil +} + +func isNumber(t string) bool { _, err := strconv.ParseFloat(t, 64); return err == nil } +func isOperator(t string) bool { + return t == "+" || t == "-" || t == "*" || t == "/" || t == "%" || t == "^" +} + +// ===== web_fetch:抓取网页并提取正文文本 ===== + +var ( + // RE2 不支持反向引用,逐标签枚举其配对闭合。 + reScriptStyle = regexp.MustCompile(`(?is)]*>.*?|]*>.*?|]*>.*?|]*>.*?`) + reTag = regexp.MustCompile(`(?s)<[^>]+>`) + reWS = regexp.MustCompile(`[ \t\x{00a0}]+`) + reBlankLines = regexp.MustCompile(`\n\s*\n\s*\n+`) +) + +const webFetchMaxText = 8000 // 提取正文上限(rune),避免塞爆上下文 + +func (g *Gateway) webFetch(ctx context.Context, call *contract.ToolCall) *contract.ToolResult { + raw := strings.TrimSpace(fmt.Sprint(call.Args["url"])) + if raw == "" || raw == "" { + return &contract.ToolResult{OK: false, Error: "web_fetch: url 必填"} + } + if reason, ok := validateExternalURL(raw, extAllowlist()); !ok { + return &contract.ToolResult{OK: false, Error: "web_fetch: URL 被拦截 —— " + reason} + } + body, status, err := httpGet(ctx, raw) + if err != nil { + return &contract.ToolResult{OK: false, Error: "web_fetch: " + err.Error()} + } + text := htmlToText(body) + if rs := []rune(text); len(rs) > webFetchMaxText { + text = string(rs[:webFetchMaxText]) + "\n…(已截断)" + } + return &contract.ToolResult{OK: true, Content: fmt.Sprintf("URL: %s (HTTP %d)\n\n%s", raw, status, text)} +} + +// htmlToText 把 HTML 粗提取为可读文本:去脚本/样式 → 去标签 → 解实体 → 收敛空白。 +func htmlToText(h string) string { + h = reScriptStyle.ReplaceAllString(h, " ") + h = regexp.MustCompile(`(?i)<\s*(br|/p|/div|/li|/h[1-6]|/tr)\s*/?>`).ReplaceAllString(h, "\n") + h = reTag.ReplaceAllString(h, "") + h = html.UnescapeString(h) + h = reWS.ReplaceAllString(h, " ") + h = reBlankLines.ReplaceAllString(h, "\n\n") + return strings.TrimSpace(h) +} + +// ===== web_search:联网搜索(Tavily 有 key 则用,否则 DuckDuckGo HTML 兜底)===== + +const webSearchDefaultTopK = 5 + +func (g *Gateway) webSearch(ctx context.Context, call *contract.ToolCall) *contract.ToolResult { + q := strings.TrimSpace(fmt.Sprint(call.Args["q"])) + if q == "" || q == "" { + return &contract.ToolResult{OK: false, Error: "web_search: q 必填"} + } + topK := webSearchDefaultTopK + if n, ok := toInt(call.Args["topK"]); ok && n > 0 && n <= 10 { + topK = n + } + if key := strings.TrimSpace(os.Getenv("TAVILY_API_KEY")); key != "" { + if out, err := tavilySearch(ctx, key, q, topK); err == nil { + return &contract.ToolResult{OK: true, Content: out} + } + // Tavily 失败 → 落 DDG 兜底 + } + out, err := ddgSearch(ctx, q, topK) + if err != nil { + return &contract.ToolResult{OK: false, Error: "web_search: " + err.Error()} + } + return &contract.ToolResult{OK: true, Content: out} +} + +var ( + reDDGLink = regexp.MustCompile(`(?is)]+class="result__a"[^>]+href="([^"]+)"[^>]*>(.*?)`) + reDDGSnippet = regexp.MustCompile(`(?is)]+class="result__snippet"[^>]*>(.*?)`) +) + +// ddgSearch 抓 DuckDuckGo HTML 版结果(免 key)。SERP 结构变动时尽力解析,解析不到则返回提示。 +func ddgSearch(ctx context.Context, q string, topK int) (string, error) { + u := "https://html.duckduckgo.com/html/?q=" + url.QueryEscape(q) + body, _, err := httpGet(ctx, u) + if err != nil { + return "", err + } + links := reDDGLink.FindAllStringSubmatch(body, -1) + snips := reDDGSnippet.FindAllStringSubmatch(body, -1) + if len(links) == 0 { + return "", fmt.Errorf("未解析到结果(DDG 可能限流或改版)") + } + var b strings.Builder + for i := 0; i < len(links) && i < topK; i++ { + title := strings.TrimSpace(htmlToText(links[i][2])) + href := ddgUnwrap(links[i][1]) + snippet := "" + if i < len(snips) { + snippet = strings.TrimSpace(htmlToText(snips[i][1])) + } + fmt.Fprintf(&b, "%d. %s\n %s\n %s\n", i+1, title, href, snippet) + } + return strings.TrimSpace(b.String()), nil +} + +// ddgUnwrap 还原 DDG 的跳转链接 /l/?uddg=<编码真实 url>。 +func ddgUnwrap(href string) string { + if i := strings.Index(href, "uddg="); i >= 0 { + raw := href[i+len("uddg="):] + if amp := strings.IndexByte(raw, '&'); amp >= 0 { + raw = raw[:amp] + } + if dec, err := url.QueryUnescape(raw); err == nil { + return dec + } + } + if strings.HasPrefix(href, "//") { + return "https:" + href + } + return href +} + +// tavilySearch 走 Tavily Search API(需 TAVILY_API_KEY)—— 返回干净的标题/URL/摘要。 +func tavilySearch(ctx context.Context, key, q string, topK int) (string, error) { + payload := fmt.Sprintf(`{"api_key":%q,"query":%q,"max_results":%d}`, key, q, topK) + req, _ := http.NewRequestWithContext(ctx, http.MethodPost, "https://api.tavily.com/search", strings.NewReader(payload)) + req.Header.Set("Content-Type", "application/json") + resp, err := (&http.Client{Timeout: extTimeout}).Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + if resp.StatusCode >= 400 { + return "", fmt.Errorf("tavily HTTP %d", resp.StatusCode) + } + data, _ := io.ReadAll(io.LimitReader(resp.Body, extMaxBytes)) + // 轻量提取(不引 JSON 结构体):用正则挑 title/url/content。 + reItem := regexp.MustCompile(`(?is)"title"\s*:\s*"(.*?)".*?"url"\s*:\s*"(.*?)".*?"content"\s*:\s*"(.*?)"`) + items := reItem.FindAllStringSubmatch(string(data), -1) + if len(items) == 0 { + return "", fmt.Errorf("tavily 无结果") + } + var b strings.Builder + for i, m := range items { + if i >= topK { + break + } + fmt.Fprintf(&b, "%d. %s\n %s\n %s\n", i+1, jsonUnesc(m[1]), jsonUnesc(m[2]), truncateRunes(jsonUnesc(m[3]), 240)) + } + return strings.TrimSpace(b.String()), nil +} + +// ===== 公共小工具 ===== + +func httpGet(ctx context.Context, raw string) (body string, status int, err error) { + allow := extAllowlist() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, raw, nil) + if err != nil { + return "", 0, err + } + // 带个常见 UA,部分站点对空 UA 返回 403。 + req.Header.Set("User-Agent", "Mozilla/5.0 (compatible; sundynix-agentix/1.0)") + client := &http.Client{ + Timeout: extTimeout, + CheckRedirect: func(r *http.Request, via []*http.Request) error { + if len(via) >= 3 { + return fmt.Errorf("重定向过多") + } + if reason, ok := validateExternalURL(r.URL.String(), allow); !ok { + return fmt.Errorf("重定向被拦截:%s", reason) + } + return nil + }, + } + resp, err := client.Do(req) + if err != nil { + return "", 0, err + } + defer resp.Body.Close() + data, _ := io.ReadAll(io.LimitReader(resp.Body, extMaxBytes)) + return string(data), resp.StatusCode, nil +} + +func toInt(v any) (int, bool) { + switch n := v.(type) { + case float64: + return int(n), true + case int: + return n, true + case string: + if i, err := strconv.Atoi(strings.TrimSpace(n)); err == nil { + return i, true + } + } + return 0, false +} + +// truncateRunes 按 rune 截断(中文安全),超出加省略号。 +func truncateRunes(s string, n int) string { + r := []rune(s) + if len(r) <= n { + return s + } + return string(r[:n]) + "…" +} + +// jsonUnesc 还原 JSON 字符串里的常见转义(够用即可,不做完整解码)。 +func jsonUnesc(s string) string { + r := strings.NewReplacer(`\"`, `"`, `\\`, `\`, `\n`, " ", `\t`, " ", `\/`, "/") + return strings.TrimSpace(r.Replace(s)) +} diff --git a/sundynix-mcp-go/internal/mcp/tools_extra_test.go b/sundynix-mcp-go/internal/mcp/tools_extra_test.go new file mode 100644 index 0000000..96ffd04 --- /dev/null +++ b/sundynix-mcp-go/internal/mcp/tools_extra_test.go @@ -0,0 +1,58 @@ +package mcp + +import ( + "math" + "strings" + "testing" +) + +func TestEvalArith(t *testing.T) { + cases := map[string]float64{ + "1+2*3": 7, + "(1+2)*3": 9, + "2^3^2": 512, // 右结合 + "-3+5": 2, + "10/4": 2.5, + "10%3": 1, + "2*(3+4)-5": 9, + "-(2+3)*2": -10, + "3.5e1+0.5": 35.5, + } + for in, want := range cases { + got, err := evalArith(in) + if err != nil { + t.Errorf("evalArith(%q) 报错: %v", in, err) + continue + } + if math.Abs(got-want) > 1e-9 { + t.Errorf("evalArith(%q)=%v want %v", in, got, want) + } + } +} + +func TestEvalArithErrors(t *testing.T) { + for _, in := range []string{"1/0", "1+", "(1+2", "1+2)", "rm -rf /", "import os", ""} { + if _, err := evalArith(in); err == nil { + t.Errorf("evalArith(%q) 应报错但没有", in) + } + } +} + +func TestHTMLToText(t *testing.T) { + in := `x +

标题

第一段&符号

第二段
` + got := htmlToText(in) + if strings.Contains(got, "alert") || strings.Contains(got, ".a{}") { + t.Fatalf("脚本/样式未剔除: %q", got) + } + if !strings.Contains(got, "标题") || !strings.Contains(got, "第一段&符号") || !strings.Contains(got, "第二段") { + t.Fatalf("正文/实体提取不正确: %q", got) + } +} + +func TestDDGUnwrap(t *testing.T) { + got := ddgUnwrap("//duckduckgo.com/l/?uddg=https%3A%2F%2Fexample.com%2Fa&rut=x") + if got != "https://example.com/a" { + t.Fatalf("ddgUnwrap 解码错误: %q", got) + } +} From ca38dcd0c92a406c240f8aee0c3e7fb1914cd1a2 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Wed, 24 Jun 2026 17:07:34 +0800 Subject: [PATCH 8/9] =?UTF-8?q?feat(tools):=20=E6=96=B0=E5=A2=9E=20sql=5Fq?= =?UTF-8?q?uery=20=E5=8F=AA=E8=AF=BB=20SQL=20=E6=9F=A5=E8=AF=A2=E5=B7=A5?= =?UTF-8?q?=E5=85=B7=EF=BC=88agent=20=E5=8F=AF=E8=A7=81=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit agent 可对数据库执行只读查询。三重防护: - 静态校验:仅单条 SELECT/WITH,词边界拒 insert/update/delete/drop/alter/create/ truncate/grant 等写/DDL 关键字(不误伤 created_at 之类列名)。 - 只读事务(sql.TxOptions{ReadOnly:true}):Postgres 引擎级强制只读,硬兜底。 - 行数(100)+超时(10s)+单元格(200 rune)上限。 连接:SQL_QUERY_DSN 优先(生产应指向专用只读库/账号),未设回退服务已解析的平台 PG DSN (经 NewGateway 传入 g.pgDSN,不再裸读 env)。独立小连接池(8)。 测试:单测 validateReadOnlySQL(放行 SELECT/WITH/列名含 created;拒写/DDL/多语句)。 live 自主 agent 实测:查 sundynix_task=157 行、sundynix_model=2 行。 Co-Authored-By: Claude Opus 4.8 (1M context) --- sundynix-mcp-go/cmd/server/main.go | 2 +- sundynix-mcp-go/internal/mcp/gateway.go | 17 +- sundynix-mcp-go/internal/mcp/sql_query.go | 168 ++++++++++++++++++ .../internal/mcp/tools_extra_test.go | 29 +++ 4 files changed, 213 insertions(+), 3 deletions(-) create mode 100644 sundynix-mcp-go/internal/mcp/sql_query.go diff --git a/sundynix-mcp-go/cmd/server/main.go b/sundynix-mcp-go/cmd/server/main.go index 5bfd161..71b067f 100644 --- a/sundynix-mcp-go/cmd/server/main.go +++ b/sundynix-mcp-go/cmd/server/main.go @@ -90,7 +90,7 @@ func main() { go b.RequestConfigWithRetry(ctx, contract.ConfigKindEmbedding, applyEmbed) go b.RequestConfigWithRetry(ctx, contract.ConfigKindChat, applyChat) - gw := mcp.NewGateway(b, engine, mem, hist, ragEngine) + gw := mcp.NewGateway(b, engine, mem, hist, ragEngine, pgDSN) log.Println("[mcp_go] serving MCP over sundynix.tools.go.* (Ctrl-C to quit)") if err := gw.Serve(ctx); err != nil && err != context.Canceled { diff --git a/sundynix-mcp-go/internal/mcp/gateway.go b/sundynix-mcp-go/internal/mcp/gateway.go index b696a58..eaa259c 100644 --- a/sundynix-mcp-go/internal/mcp/gateway.go +++ b/sundynix-mcp-go/internal/mcp/gateway.go @@ -3,6 +3,7 @@ package mcp import ( "context" + "database/sql" "encoding/json" "fmt" "log" @@ -10,6 +11,7 @@ import ( "path/filepath" "sort" "strings" + "sync" "time" sharedbus "github.com/sundynix/sundynix-shared/bus" @@ -30,6 +32,11 @@ type Gateway struct { history *history.Store rag *rag.Engine tools map[string]toolDef // 工具注册表:唯一事实源,dispatch 与 list_tools 共用,杜绝漂移 + + pgDSN string // 平台 PG DSN(sql_query 兜底库;SQL_QUERY_DSN 未设时用它) + sqlOnce sync.Once // sql_query 只读连接懒连一次 + sqlDB *sql.DB // + sqlDBErr error // } // paramSpec 是一个工具参数的声明(供自主 agent 据此生成调用入参)。 @@ -53,8 +60,8 @@ type toolDef struct { handler func(context.Context, *contract.ToolCall) *contract.ToolResult } -func NewGateway(b *sharedbus.Bus, s *search.Hybrid, m *memory.Store, h *history.Store, r *rag.Engine) *Gateway { - g := &Gateway{bus: b, search: s, memory: m, history: h, rag: r} +func NewGateway(b *sharedbus.Bus, s *search.Hybrid, m *memory.Store, h *history.Store, r *rag.Engine, pgDSN string) *Gateway { + g := &Gateway{bus: b, search: s, memory: m, history: h, rag: r, pgDSN: pgDSN} g.tools = g.buildRegistry() return g } @@ -129,6 +136,12 @@ func (g *Gateway) buildRegistry() map[string]toolDef { params: []paramSpec{{Name: "tz", Type: "string", Desc: "可选时区,如 Asia/Shanghai;缺省服务器本地时区"}}, handler: g.currentDatetime, }, + "sql_query": { + cn: "SQL查询", desc: "对数据库执行只读 SQL 查询(仅 SELECT/WITH),返回结果表。需要查业务数据/统计时调用。", + agent: true, + params: []paramSpec{{Name: "sql", Type: "string", Desc: "只读 SQL,如 SELECT count(*) FROM sundynix_task", Required: true}}, + handler: g.sqlQuery, + }, // —— 仅内部/流水线/管理用,不暴露给自主 agent —— "kb_ingest": {cn: "知识入库", desc: "文本切块 → 向量化 → 写入 Milvus / Bleve", handler: g.kbIngest}, diff --git a/sundynix-mcp-go/internal/mcp/sql_query.go b/sundynix-mcp-go/internal/mcp/sql_query.go new file mode 100644 index 0000000..4460f36 --- /dev/null +++ b/sundynix-mcp-go/internal/mcp/sql_query.go @@ -0,0 +1,168 @@ +package mcp + +import ( + "context" + "database/sql" + "fmt" + "os" + "strings" + "time" + + "github.com/sundynix/sundynix-shared/contract" + "gorm.io/driver/postgres" + "gorm.io/gorm" + "gorm.io/gorm/logger" +) + +const ( + sqlQueryMaxRows = 100 // 结果行上限,防超大结果撑爆上下文 + sqlQueryTimeout = 10 * time.Second // 单次查询超时 + sqlCellMaxRunes = 200 // 单元格文本上限 +) + +// sqlQueryDB 懒连只读查询库:优先 SQL_QUERY_DSN(生产应指向专用只读库 / 只读账号), +// 未配置则回退本服务已解析的平台 PG DSN(g.pgDSN,仅供开发;线上务必单配,避免把平台库直接暴露给 agent)。 +func (g *Gateway) sqlQueryDB() (*sql.DB, error) { + g.sqlOnce.Do(func() { + dsn := strings.TrimSpace(os.Getenv("SQL_QUERY_DSN")) + if dsn == "" { + dsn = g.pgDSN + } + if dsn == "" { + g.sqlDBErr = fmt.Errorf("未配置 SQL_QUERY_DSN") + return + } + gdb, err := gorm.Open(postgres.New(postgres.Config{DSN: dsn}), &gorm.Config{ + Logger: logger.Default.LogMode(logger.Silent), + }) + if err != nil { + g.sqlDBErr = err + return + } + db, err := gdb.DB() + if err != nil { + g.sqlDBErr = err + return + } + db.SetMaxOpenConns(8) // 只读查询独立小池,不挤占主连接 + db.SetMaxIdleConns(2) + db.SetConnMaxLifetime(time.Hour) + g.sqlDB = db + }) + return g.sqlDB, g.sqlDBErr +} + +// sqlQuery 对数据库执行只读 SQL。三重防护:① 仅允许单条 SELECT/WITH 语句(拒多语句/写/DDL); +// ② 在「只读事务」中执行(Postgres 引擎级强制只读,是真正的兜底);③ 行数 + 超时 + 单元格上限。 +func (g *Gateway) sqlQuery(ctx context.Context, call *contract.ToolCall) *contract.ToolResult { + q := strings.TrimSpace(fmt.Sprint(call.Args["sql"])) + if q == "" || q == "" { + return &contract.ToolResult{OK: false, Error: "sql_query: sql 必填"} + } + if reason, ok := validateReadOnlySQL(q); !ok { + return &contract.ToolResult{OK: false, Error: "sql_query: " + reason} + } + db, err := g.sqlQueryDB() + if err != nil { + return &contract.ToolResult{OK: false, Error: "sql_query: " + err.Error()} + } + cctx, cancel := context.WithTimeout(ctx, sqlQueryTimeout) + defer cancel() + // 只读事务:即便上面的语句校验被绕过,Postgres 也会在引擎层拒绝任何写操作。 + tx, err := db.BeginTx(cctx, &sql.TxOptions{ReadOnly: true}) + if err != nil { + return &contract.ToolResult{OK: false, Error: "sql_query: " + err.Error()} + } + defer func() { _ = tx.Rollback() }() + + rows, err := tx.QueryContext(cctx, q) + if err != nil { + return &contract.ToolResult{OK: false, Error: "sql_query: " + err.Error()} + } + defer rows.Close() + cols, _ := rows.Columns() + + var b strings.Builder + b.WriteString(strings.Join(cols, " | ") + "\n") + b.WriteString(strings.Repeat("-", len(strings.Join(cols, " | "))) + "\n") + n := 0 + for rows.Next() { + if n >= sqlQueryMaxRows { + break + } + cells := make([]any, len(cols)) + ptrs := make([]any, len(cols)) + for i := range cells { + ptrs[i] = &cells[i] + } + if err := rows.Scan(ptrs...); err != nil { + return &contract.ToolResult{OK: false, Error: "sql_query: scan " + err.Error()} + } + strs := make([]string, len(cols)) + for i, c := range cells { + strs[i] = truncateRunes(cellToString(c), sqlCellMaxRunes) + } + b.WriteString(strings.Join(strs, " | ") + "\n") + n++ + } + if err := rows.Err(); err != nil { + return &contract.ToolResult{OK: false, Error: "sql_query: " + err.Error()} + } + b.WriteString(fmt.Sprintf("(%d 行%s)", n, map[bool]string{true: ",已截断至上限"}[n >= sqlQueryMaxRows])) + return &contract.ToolResult{OK: true, Content: b.String()} +} + +// validateReadOnlySQL 静态校验:单条语句、以 SELECT/WITH 开头、不含写/DDL 关键字。 +// 与只读事务双保险(这层挡明显误用,事务层是硬保证)。 +func validateReadOnlySQL(q string) (reason string, ok bool) { + s := strings.TrimSpace(q) + s = strings.TrimSuffix(s, ";") + if strings.Contains(s, ";") { + return "只允许单条语句", false + } + low := strings.ToLower(s) + if !strings.HasPrefix(low, "select") && !strings.HasPrefix(low, "with") { + return "只允许 SELECT / WITH 查询", false + } + // 词边界匹配写/DDL 关键字(避免误伤列名如 created_at)。 + for _, kw := range []string{"insert", "update", "delete", "drop", "alter", "create", + "truncate", "grant", "revoke", "comment", "copy", "merge", "call", "do "} { + if containsWord(low, kw) { + return "检测到写/DDL 关键字「" + strings.TrimSpace(kw) + "」,仅允许只读查询", false + } + } + return "", true +} + +func containsWord(s, word string) bool { + for i := 0; ; { + idx := strings.Index(s[i:], word) + if idx < 0 { + return false + } + j := i + idx + before := j == 0 || !isWordChar(rune(s[j-1])) + after := j+len(word) >= len(s) || !isWordChar(rune(s[j+len(word)])) + if before && after { + return true + } + i = j + len(word) + } +} + +func isWordChar(r rune) bool { + return r == '_' || (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') +} + +func cellToString(v any) string { + switch x := v.(type) { + case nil: + return "NULL" + case []byte: + return string(x) + case time.Time: + return x.Format("2006-01-02 15:04:05") + default: + return fmt.Sprint(x) + } +} diff --git a/sundynix-mcp-go/internal/mcp/tools_extra_test.go b/sundynix-mcp-go/internal/mcp/tools_extra_test.go index 96ffd04..d78808a 100644 --- a/sundynix-mcp-go/internal/mcp/tools_extra_test.go +++ b/sundynix-mcp-go/internal/mcp/tools_extra_test.go @@ -56,3 +56,32 @@ func TestDDGUnwrap(t *testing.T) { t.Fatalf("ddgUnwrap 解码错误: %q", got) } } + +func TestValidateReadOnlySQL(t *testing.T) { + okCases := []string{ + "SELECT 1", + "select count(*) from sundynix_task", + "WITH x AS (SELECT 1) SELECT * FROM x", + " select * from t where created_at > now() ", // created/update 作为列名一部分不应误伤 + } + for _, q := range okCases { + if _, ok := validateReadOnlySQL(q); !ok { + t.Errorf("应通过: %q", q) + } + } + badCases := []string{ + "insert into t values (1)", + "update t set a=1", + "delete from t", + "drop table t", + "select 1; drop table t", + "truncate t", + "SELECT 1; SELECT 2", + "grant all on t to x", + } + for _, q := range badCases { + if _, ok := validateReadOnlySQL(q); ok { + t.Errorf("应拒绝: %q", q) + } + } +} From 592b2a3d97a51b94bdbc845e388548ba58dc30ec Mon Sep 17 00:00:00 2001 From: Blizzard Date: Wed, 24 Jun 2026 17:24:41 +0800 Subject: [PATCH 9/9] =?UTF-8?q?feat(tools):=20chart=20=E5=9B=BE=E8=A1=A8?= =?UTF-8?q?=E5=B7=A5=E5=85=B7=20=E2=80=94=E2=80=94=20=E5=B7=A5=E5=85=B7?= =?UTF-8?q?=E4=BA=A7=E5=87=BA=20JSON=20spec=EF=BC=8C=E5=89=8D=E7=AB=AF=20S?= =?UTF-8?q?VG=20=E6=B8=B2=E6=9F=93=EF=BC=88=E8=81=8C=E8=B4=A3=E5=88=86?= =?UTF-8?q?=E7=A6=BB=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 按「工具只产数据、渲染交前端」设计: - 后端 chart 工具(mcp-go):校验并返回规范化图表 JSON(type=bar/line/pie + labels + series, 校验类型/长度一致/pie 取首系列)。工具说明指示 agent 用 ```chart 围栏原样包裹返回的 JSON。 - 前端:lib/chartspec.ts 从输出抽取 ```chart 块(解析失败回退为文本不丢内容); components/ChartView.tsx 自绘 SVG 柱/线/饼图(无第三方图表依赖); BottomDrawer 输出区含图表块时分段渲染(文本 + SVG),否则纯文本。 测试:前端 chartspec 单测 12 例(isChartSpec 校验、分段抽取、非法块回退、多块、hasChart); tsc 干净,vitest 48 过。live 自主 agent:chart 工具产出 {"type":"bar",...}, agent 正确用 ```chart 围栏嵌入答复,前端据此渲染。 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../frontend/src/components/ChartView.tsx | 126 ++++++++++++++++++ .../frontend/src/lib/chartspec.test.ts | 47 +++++++ .../frontend/src/lib/chartspec.ts | 60 +++++++++ .../frontend/src/shell/BottomDrawer.tsx | 37 ++++- sundynix-mcp-go/internal/mcp/chart.go | 60 +++++++++ sundynix-mcp-go/internal/mcp/gateway.go | 11 ++ 6 files changed, 336 insertions(+), 5 deletions(-) create mode 100644 sundynix-desktop/frontend/src/components/ChartView.tsx create mode 100644 sundynix-desktop/frontend/src/lib/chartspec.test.ts create mode 100644 sundynix-desktop/frontend/src/lib/chartspec.ts create mode 100644 sundynix-mcp-go/internal/mcp/chart.go diff --git a/sundynix-desktop/frontend/src/components/ChartView.tsx b/sundynix-desktop/frontend/src/components/ChartView.tsx new file mode 100644 index 0000000..fddf146 --- /dev/null +++ b/sundynix-desktop/frontend/src/components/ChartView.tsx @@ -0,0 +1,126 @@ +import type { ChartSpec } from "../lib/chartspec"; + +// ChartView:把 chart spec 自绘成 SVG(bar / line / pie),无第三方图表依赖。 +const PALETTE = ["#60a5fa", "#34d399", "#fbbf24", "#f87171", "#a78bfa", "#22d3ee", "#fb923c", "#4ade80"]; + +export function ChartView({ spec }: { spec: ChartSpec }) { + return ( +
+ {spec.title &&
{spec.title}
} + {spec.type === "pie" ? : } + {spec.type !== "pie" && spec.series.length > 1 && s.name || `系列${i + 1}`)} />} +
+ ); +} + +function Legend({ names }: { names: string[] }) { + return ( +
+ {names.map((n, i) => ( + + + {n} + + ))} +
+ ); +} + +// BarLine:柱状图 / 折线图(共用坐标系)。 +function BarLine({ spec }: { spec: ChartSpec }) { + const W = 480, H = 220, padL = 40, padB = 28, padT = 8, padR = 8; + const plotW = W - padL - padR, plotH = H - padT - padB; + const all = spec.series.flatMap((s) => s.data); + const max = Math.max(1, ...all); + const min = Math.min(0, ...all); + const span = max - min || 1; + const y = (v: number) => padT + plotH - ((v - min) / span) * plotH; + const n = spec.labels.length; + const slot = plotW / n; + + return ( + + {/* y 轴基准线 */} + + {fmt(max)} + {fmt(min)} + + {spec.type === "bar" + ? spec.series.map((s, si) => + s.data.map((v, i) => { + const bw = (slot * 0.7) / spec.series.length; + const x = padL + i * slot + slot * 0.15 + si * bw; + return ( + + {`${spec.labels[i]}: ${v}`} + + ); + }), + ) + : spec.series.map((s, si) => { + const pts = s.data.map((v, i) => `${padL + i * slot + slot / 2},${y(v)}`).join(" "); + return ( + + + {s.data.map((v, i) => ( + + {`${spec.labels[i]}: ${v}`} + + ))} + + ); + })} + + {/* x 轴标签 */} + {spec.labels.map((lb, i) => ( + + {lb.length > 6 ? lb.slice(0, 6) + "…" : lb} + + ))} + + ); +} + +// Pie:饼图(用第一条系列)。 +function Pie({ spec }: { spec: ChartSpec }) { + const data = spec.series[0]?.data ?? []; + const total = data.reduce((a, b) => a + Math.max(0, b), 0) || 1; + const cx = 110, cy = 110, r = 90; + let acc = -Math.PI / 2; // 从 12 点方向起 + const arcs = data.map((v, i) => { + const frac = Math.max(0, v) / total; + const a0 = acc; + const a1 = acc + frac * Math.PI * 2; + acc = a1; + const large = a1 - a0 > Math.PI ? 1 : 0; + const x0 = cx + r * Math.cos(a0), y0 = cy + r * Math.sin(a0); + const x1 = cx + r * Math.cos(a1), y1 = cy + r * Math.sin(a1); + const d = `M${cx},${cy} L${x0.toFixed(2)},${y0.toFixed(2)} A${r},${r} 0 ${large} 1 ${x1.toFixed(2)},${y1.toFixed(2)} Z`; + return { d, color: PALETTE[i % PALETTE.length], label: spec.labels[i], pct: Math.round(frac * 100) }; + }); + return ( +
+ + {arcs.map((a, i) => ( + + {`${a.label}: ${data[i]} (${a.pct}%)`} + + ))} + +
+ {arcs.map((a, i) => ( + + + {a.label} · {a.pct}% + + ))} +
+
+ ); +} + +function fmt(v: number): string { + if (Math.abs(v) >= 1000) return (v / 1000).toFixed(1) + "k"; + return String(Math.round(v * 100) / 100); +} diff --git a/sundynix-desktop/frontend/src/lib/chartspec.test.ts b/sundynix-desktop/frontend/src/lib/chartspec.test.ts new file mode 100644 index 0000000..8aaeaf0 --- /dev/null +++ b/sundynix-desktop/frontend/src/lib/chartspec.test.ts @@ -0,0 +1,47 @@ +import { describe, it, expect } from "vitest"; +import { extractChartBlocks, isChartSpec, hasChart } from "./chartspec"; + +const spec = { type: "bar", title: "销量", labels: ["Q1", "Q2"], series: [{ name: "销量", data: [120, 180] }] }; + +describe("isChartSpec", () => { + it("合法 spec 通过", () => expect(isChartSpec(spec)).toBe(true)); + it.each([ + {}, + { type: "x", labels: ["a"], series: [{ data: [1] }] }, + { type: "bar", labels: [], series: [{ data: [] }] }, + { type: "bar", labels: ["a"], series: [] }, + { type: "bar", labels: ["a"], series: [{ name: "x" }] }, // 无 data + ])("非法 spec 拒绝 %#", (bad) => expect(isChartSpec(bad)).toBe(false)); +}); + +describe("extractChartBlocks", () => { + it("纯文本 → 单个 text 段", () => { + const segs = extractChartBlocks("你好世界"); + expect(segs).toEqual([{ kind: "text", text: "你好世界" }]); + }); + + it("文本 + chart 块 + 文本 → 三段且顺序正确", () => { + const out = "看图:\n```chart\n" + JSON.stringify(spec) + "\n```\n以上。"; + const segs = extractChartBlocks(out); + expect(segs.map((s) => s.kind)).toEqual(["text", "chart", "text"]); + expect(segs[1].kind === "chart" && segs[1].spec.title).toBe("销量"); + }); + + it("非法 JSON 的 chart 块 → 回退为文本,不丢内容", () => { + const out = "```chart\n{坏的}\n```"; + const segs = extractChartBlocks(out); + expect(segs).toHaveLength(1); + expect(segs[0].kind).toBe("text"); + }); + + it("两个 chart 块都解析", () => { + const blk = "```chart\n" + JSON.stringify(spec) + "\n```"; + const segs = extractChartBlocks(blk + "\n中间\n" + blk); + expect(segs.filter((s) => s.kind === "chart")).toHaveLength(2); + }); +}); + +describe("hasChart", () => { + it("含 chart 围栏 → true", () => expect(hasChart("a\n```chart\n{}\n```")).toBe(true)); + it("不含 → false", () => expect(hasChart("```json\n{}\n```")).toBe(false)); +}); diff --git a/sundynix-desktop/frontend/src/lib/chartspec.ts b/sundynix-desktop/frontend/src/lib/chartspec.ts new file mode 100644 index 0000000..da07bbd --- /dev/null +++ b/sundynix-desktop/frontend/src/lib/chartspec.ts @@ -0,0 +1,60 @@ +// 图表 spec 与「从模型输出里抽取 ```chart 代码块」的解析。 +// 约定:chart 工具返回图表 JSON,agent 在答复里用 ```chart 围栏原样包裹;前端据此渲染 SVG。 + +export interface ChartSeries { + name?: string; + data: number[]; +} + +export interface ChartSpec { + type: "bar" | "line" | "pie"; + title?: string; + labels: string[]; + series: ChartSeries[]; +} + +// Segment:把一段输出拆成「普通文本」与「图表」交错的有序片段,供输出区分别渲染。 +export type Segment = { kind: "text"; text: string } | { kind: "chart"; spec: ChartSpec }; + +const chartFence = /```chart\s*\n([\s\S]*?)```/g; + +// isChartSpec 校验解析出的对象是否是合法图表 spec(防脏数据炸渲染)。 +export function isChartSpec(v: unknown): v is ChartSpec { + if (!v || typeof v !== "object") return false; + const o = v as Record; + if (o.type !== "bar" && o.type !== "line" && o.type !== "pie") return false; + if (!Array.isArray(o.labels) || o.labels.length === 0) return false; + if (!Array.isArray(o.series) || o.series.length === 0) return false; + return (o.series as unknown[]).every( + (s) => s && typeof s === "object" && Array.isArray((s as ChartSeries).data), + ); +} + +// extractChartBlocks 把输出文本拆为 text / chart 片段(保持原顺序)。 +// ```chart 块解析失败或非法 → 回退为普通文本,绝不丢内容。 +export function extractChartBlocks(input: string): Segment[] { + if (!input) return []; + const segs: Segment[] = []; + let last = 0; + for (let m = chartFence.exec(input); m; m = chartFence.exec(input)) { + if (m.index > last) segs.push({ kind: "text", text: input.slice(last, m.index) }); + let parsed: unknown; + try { + parsed = JSON.parse(m[1].trim()); + } catch { + parsed = null; + } + if (isChartSpec(parsed)) segs.push({ kind: "chart", spec: parsed }); + else segs.push({ kind: "text", text: m[0] }); // 解析失败:原样当文本,不丢 + last = m.index + m[0].length; + } + chartFence.lastIndex = 0; + if (last < input.length) segs.push({ kind: "text", text: input.slice(last) }); + return segs; +} + +// hasChart 快速判断输出里是否含图表块(决定是否走分段渲染)。 +export function hasChart(input: string): boolean { + const has = /```chart\s*\n/.test(input); + return has; +} diff --git a/sundynix-desktop/frontend/src/shell/BottomDrawer.tsx b/sundynix-desktop/frontend/src/shell/BottomDrawer.tsx index 4bdea0a..be91343 100644 --- a/sundynix-desktop/frontend/src/shell/BottomDrawer.tsx +++ b/sundynix-desktop/frontend/src/shell/BottomDrawer.tsx @@ -2,6 +2,8 @@ import { useState } from "react"; import { ChevronDown, ChevronUp, Wrench, ShieldCheck, Check, X } from "lucide-react"; import { deriveNodes, pendingApproval, type RunState } from "../lib/run"; import { ExecTrace } from "../components/ExecTrace"; +import { ChartView } from "../components/ChartView"; +import { extractChartBlocks, hasChart } from "../lib/chartspec"; import { approveTask } from "../lib/api"; import { Tabs, Badge, cn, type TabDef } from "../ui"; @@ -53,11 +55,7 @@ export function BottomDrawer({ run }: { run: RunState }) {
{open && (
- {tab === "output" && ( -
-              {run.output || "在编排页搭图 → 运行,模型注入画像与历史后流式作答,token 在此呈现。"}
-            
- )} + {tab === "output" && } {tab === "trace" && } {tab === "tools" && } {tab === "cite" &&

引用列表:RAG 答案的来源块(源文档 + 分数 + 来源徽标)。

} @@ -122,6 +120,35 @@ function ApprovalBar({ taskId, node, title, summary }: { taskId: string; node: s ); } +// OutputView:渲染模型输出。含 ```chart 块时分段渲染(文本 + SVG 图表),否则纯文本。 +function OutputView({ output }: { output: string }) { + if (!output) { + return ( +
+        在编排页搭图 → 运行,模型注入画像与历史后流式作答,token 在此呈现。
+      
+ ); + } + if (!hasChart(output)) { + return
{output}
; + } + return ( +
+ {extractChartBlocks(output).map((seg, i) => + seg.kind === "chart" ? ( + + ) : ( + seg.text.trim() && ( +
+              {seg.text}
+            
+ ) + ), + )} +
+ ); +} + // ToolCalls:从执行事件里筛出工具调用节点,逐条展示入参 → 产出 + 耗时/状态。 function ToolCalls({ run }: { run: RunState }) { const tools = deriveNodes(run.exec).filter((n) => n.kind === "tool"); diff --git a/sundynix-mcp-go/internal/mcp/chart.go b/sundynix-mcp-go/internal/mcp/chart.go new file mode 100644 index 0000000..a0aae2d --- /dev/null +++ b/sundynix-mcp-go/internal/mcp/chart.go @@ -0,0 +1,60 @@ +package mcp + +import ( + "encoding/json" + "fmt" + + "context" + + "github.com/sundynix/sundynix-shared/contract" +) + +// chartSeries 是一条数据系列。 +type chartSeries struct { + Name string `json:"name,omitempty"` + Data []float64 `json:"data"` +} + +// chartSpec 是图表的结构化规范(工具产出,前端据此渲染 SVG,不在后端出图)。 +type chartSpec struct { + Type string `json:"type"` // bar / line / pie + Title string `json:"title,omitempty"` // + Labels []string `json:"labels"` // x 轴/扇区标签 + Series []chartSeries `json:"series"` // 一条或多条数据系列(pie 取第一条) +} + +// chart 工具:只校验并返回规范化图表 JSON(渲染交前端)。职责单一、零图片传输。 +// 返回内容即一段 chart JSON;工具说明会指示 agent 在最终答复里用 ```chart 围栏原样包裹它。 +func (g *Gateway) chart(_ context.Context, call *contract.ToolCall) *contract.ToolResult { + // 用 JSON round-trip 把 args 收进类型化结构(args 里 labels/series 是 []any,手解繁琐)。 + raw, _ := json.Marshal(call.Args) + var in chartSpec + if err := json.Unmarshal(raw, &in); err != nil { + return &contract.ToolResult{OK: false, Error: "chart: 参数解析失败 —— " + err.Error()} + } + switch in.Type { + case "bar", "line", "pie": + case "": + in.Type = "bar" + default: + return &contract.ToolResult{OK: false, Error: "chart: type 仅支持 bar / line / pie"} + } + if len(in.Labels) == 0 { + return &contract.ToolResult{OK: false, Error: "chart: labels 必填"} + } + if len(in.Series) == 0 || len(in.Series[0].Data) == 0 { + return &contract.ToolResult{OK: false, Error: "chart: series 至少一条且 data 非空"} + } + for i, s := range in.Series { + if len(s.Data) != len(in.Labels) { + return &contract.ToolResult{OK: false, + Error: fmt.Sprintf("chart: 第 %d 条系列 data 长度(%d) 与 labels 长度(%d) 不一致", i+1, len(s.Data), len(in.Labels))} + } + } + if in.Type == "pie" { + in.Series = in.Series[:1] // pie 只用第一条系列 + } + out, _ := json.Marshal(in) + // 提示 agent:把这段 JSON 用 ```chart 围栏原样放进最终答复,前端会渲染成图。 + return &contract.ToolResult{OK: true, Content: string(out)} +} diff --git a/sundynix-mcp-go/internal/mcp/gateway.go b/sundynix-mcp-go/internal/mcp/gateway.go index eaa259c..78ea6bb 100644 --- a/sundynix-mcp-go/internal/mcp/gateway.go +++ b/sundynix-mcp-go/internal/mcp/gateway.go @@ -142,6 +142,17 @@ func (g *Gateway) buildRegistry() map[string]toolDef { params: []paramSpec{{Name: "sql", Type: "string", Desc: "只读 SQL,如 SELECT count(*) FROM sundynix_task", Required: true}}, handler: g.sqlQuery, }, + "chart": { + cn: "图表", desc: "把数据生成图表。返回图表 JSON——请在最终答复中用 ```chart 代码块原样包裹该 JSON,前端会渲染成图。需要可视化数据分布/趋势时调用。", + agent: true, + params: []paramSpec{ + {Name: "type", Type: "string", Desc: "图表类型:bar / line / pie", Required: true}, + {Name: "title", Type: "string", Desc: "图表标题"}, + {Name: "labels", Type: "array", Desc: "x 轴/扇区标签数组,如 [\"Q1\",\"Q2\"]", Required: true}, + {Name: "series", Type: "array", Desc: "数据系列数组,如 [{\"name\":\"销量\",\"data\":[120,180]}]", Required: true}, + }, + handler: g.chart, + }, // —— 仅内部/流水线/管理用,不暴露给自主 agent —— "kb_ingest": {cn: "知识入库", desc: "文本切块 → 向量化 → 写入 Milvus / Bleve", handler: g.kbIngest},