From 55d50417a9f5f60155473049aaa8fc1f10f1431a Mon Sep 17 00:00:00 2001 From: Blizzard Date: Mon, 6 Jul 2026 12:00:13 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=A8=A1=E5=9E=8B=E5=81=A5=E5=BA=B7/?= =?UTF-8?q?=E7=86=94=E6=96=AD=E6=80=81=20surface=20=E5=88=B0=E7=AE=A1?= =?UTF-8?q?=E7=90=86=E7=AB=AF=EF=BC=88T4.F=20=E5=8F=AF=E8=A7=82=E6=B5=8B?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit failover/熔断的运行时态原来只在 dispatcher 日志、admin 看不到 —— 本次接到概览可见： - harness: CircuitBreaker.Snapshot() 只读观测访问器（state + fails，不动状态机） - llm: Pool.ModelHealth() 上报主备链每模型 {provider,model,role,state,fails}； buildWithFallbacks 把模型名↔breaker 配对（同包直接读 failoverModel.breakers）； newFailoverModel 改返回具体类型以便读 breakers - dispatcher 心跳 payload 加 models[] - gateway /admin/overview 独立超时 Ping dispatcher，合并进 models.health - admin 概览「模型路由」新增「运行时链路态(实时)」：逐模型状态点 (🟢在线/🔴熔断中+失败数/🟡半开探测/单点) - 单测：Snapshot、Pool.ModelHealth(名字↔态配对/单点/空) - live：配坏主→提交任务打熔断→概览显示 broken-demo「熔断中·失败3」，备用在线 Co-Authored-By: Claude Opus 4.8 --- DEPTH_ROADMAP.md | 9 +- sundynix-admin/src/api.ts | 18 +- sundynix-admin/src/pages/DashboardPage.tsx | 30 + sundynix-agentix-saas-refactor-plan.md | 1508 +++++++++++++++++ sundynix-dispatcher/cmd/dispatcher/main.go | 1 + .../internal/harness/circuitbreaker.go | 13 + .../internal/harness/circuitbreaker_test.go | 18 + sundynix-dispatcher/internal/llm/failover.go | 3 +- sundynix-dispatcher/internal/llm/pool.go | 66 +- .../internal/llm/pool_health_test.go | 45 + sundynix-gateway/internal/handler/admin.go | 25 + 11 files changed, 1719 insertions(+), 17 deletions(-) create mode 100644 sundynix-agentix-saas-refactor-plan.md create mode 100644 sundynix-dispatcher/internal/llm/pool_health_test.go diff --git a/DEPTH_ROADMAP.md b/DEPTH_ROADMAP.md index 1bfd335..74343ed 100644 --- a/DEPTH_ROADMAP.md +++ b/DEPTH_ROADMAP.md @@ -202,10 +202,11 @@ RBAC 未做，暂以单管理员账号代理；概览口径必须是**系统级* - [ ] `X-Session-ID` 防伪 / 开发 JWT 密钥外置 / 输入校验补强 | S（待） - [ ] 性能：ListModels 去 O(N²)（admin.go:48）、KB/Agent 列表分页、报告分章检索缓存、Milvus 批量 ensure | S~M - [ ] 可观测：panic 进 trace span + 回写通知、TTFT/token-s/各路检索耗时指标 | M -- [ ] 🔴 模型健康/熔断态 surface 到管理端（新，2026-07-02 failover demo 暴露的真实缺口）—— - failover/熔断现只在 dispatcher 日志，admin UI 看不到「哪个模型正被熔断、切了几次」。 - 补：dispatcher 经 NATS 心跳上报每模型 breaker 态(在线/熔断Open/半开 + 连续失败数 + failover 切换计数) - → gateway 聚合 → 展示到概览「模型路由」面板或状态页。| M（后期一起补） +- [x] 🔴 模型健康/熔断态 surface 到管理端 ✅（2026-07-06）—— harness `CircuitBreaker.Snapshot()` + + `llm.Pool.ModelHealth()`(主备链每模型 provider/model/role/state/fails，同包读 failoverModel.breakers) → + dispatcher 心跳带 `models[]` → gateway `/admin/overview` 独立超时 Ping dispatcher 合并 → admin 概览 + 「模型路由」新增「运行时链路态(实时)」逐模型状态点(🟢在线/🔴熔断中+失败数/🟡半开/单点)。 + 单测(Snapshot/ModelHealth 名字↔态配对) + **live：配坏主→提交任务打熔断→概览 broken-demo 显示「熔断中·失败3」**。 - [ ] 配置化：切块大小 / history 轮数 / 各并发度 收口为统一可配 | S - [x] 拆除 `search.Hybrid` 残骸 ✅ —— 该包空转(NewHybrid 返空、Query 返 nil)、构造后存进 Gateway 却从不调用(真 RAG 走 rag.Engine)。删整个 internal/search 包 + gateway/main.go 接线。build/vet 干净。（审计原说"死代码删文件"不准：它是接了线的残骸，需拆接线） diff --git a/sundynix-admin/src/api.ts b/sundynix-admin/src/api.ts index 8841a0b..88d8b7c 100644 --- a/sundynix-admin/src/api.ts +++ b/sundynix-admin/src/api.ts @@ -219,6 +219,15 @@ export async function statsOverview(): Promise { // —— 管理端系统级聚合（控制塔口径，RequireAdmin）—— // 区别于 statsOverview（桌面端个人工作台）：这里一律全平台口径——全部用户/任务/评测/模型态/提示词态/健康。 +// 单模型运行时健康态（failover 链上主/备各一条）。 +export interface ModelHealthItem { + provider: string; + model: string; + role: string; // primary / fallback + state: string; // closed(在线) / open(熔断中) / half-open(半开探测) / single(无备用链) + fails: number; +} + export interface AdminOverview { users: number; kb_count: number; // 全平台知识库数 @@ -230,7 +239,14 @@ export interface AdminOverview { eval_avg: number; faithful_avg: number; eval_count: number; - models: { chat_count: number; embedding_count: number; active_chat: string; active_embedding: string; fallbacks: number }; + models: { + chat_count: number; + embedding_count: number; + active_chat: string; + active_embedding: string; + fallbacks: number; + health: ModelHealthItem[]; // 运行时每模型 failover/熔断态（来自 dispatcher） + }; prompts: { managed: number; overrides: number }; services: Record; checked_at: string; diff --git a/sundynix-admin/src/pages/DashboardPage.tsx b/sundynix-admin/src/pages/DashboardPage.tsx index 78f6327..b5fdcba 100644 --- a/sundynix-admin/src/pages/DashboardPage.tsx +++ b/sundynix-admin/src/pages/DashboardPage.tsx @@ -14,6 +14,15 @@ const STATUS_STYLE: Record = { }; const statusStyle = (s: string) => STATUS_STYLE[s] ?? { label: s, color: "#94a3b8" }; +// 模型熔断态 → 展示样式。 +const BREAKER_STYLE: Record = { + closed: { label: "在线", dot: "bg-emerald-500", text: "text-emerald-600" }, + open: { label: "熔断中", dot: "bg-rose-500", text: "text-rose-500" }, + "half-open": { label: "半开探测", dot: "bg-amber-500", text: "text-amber-600" }, + single: { label: "单点", dot: "bg-gray-300", text: "text-gray-400" }, +}; +const breakerStyle = (s: string) => BREAKER_STYLE[s] ?? { label: s, dot: "bg-gray-300", text: "text-gray-400" }; + function last7DaysTrend(trend: { key: string; count: number }[]): { key: string; count: number }[] { const byKey = new Map(trend.map((d) => [d.key, d.count])); const out: { key: string; count: number }[] = []; @@ -169,6 +178,27 @@ export function DashboardPage() { chat {m.chat_count} · embedding {m.embedding_count} + + {/* 运行时 failover/熔断态（来自 dispatcher） */} + {m.health && m.health.length > 0 && ( +
+
运行时链路态（实时）
+
+ {m.health.map((h, i) => { + const st = breakerStyle(h.state); + return ( +
+ + {h.role === "primary" ? "主" : "备"} + {h.model} + {st.label} + {h.fails > 0 && h.state !== "closed" && 失败 {h.fails}} +
+ ); + })} +
+
+ )}
diff --git a/sundynix-agentix-saas-refactor-plan.md b/sundynix-agentix-saas-refactor-plan.md new file mode 100644 index 0000000..8966ad8 --- /dev/null +++ b/sundynix-agentix-saas-refactor-plan.md @@ -0,0 +1,1508 @@ +# sundynix-agentix · Web SaaS 架构重构实施文档 + +> 版本：v2.0-refactor +> 日期：2026-07-06 +> 基于：`ARCHITECTURE_DESIGN.md` 当前完整版 +> 目标：在保留现有事件驱动、Eino 编排、NATS 总线、MCP 工具化优势的基础上，将系统从「桌面端 + 云编排平台」升级为「可商业化的 Web SaaS Agent 平台」。 + +--- + +## 0. 重构结论 + +当前架构已经具备较强的 Agent 平台雏形： + +- Gateway 作为唯一 HTTP 接入层。 +- Dispatcher 使用 Eino 作为编排核心。 +- NATS + JetStream 承载任务、工具 RPC、token 流、控制面广播。 +- Redis Stream 支持 SSE 断点回放。 +- mcp-go / mcp-py 承载工具服务。 +- LLM Pool 支持 OpenAI-compatible Provider、failover、熔断、缓存。 +- Harness 层已经覆盖 guardrail、脱敏、预算、评测纠偏等能力。 + +但如果要做 Web SaaS 商业化，需要优先补齐以下核心能力： + +1. **Client 层重构**：从「桌面端主产品」调整为「Web App 主产品 + Desktop Connector 可选」。 +2. **Task Runtime 独立化**：把任务状态机、重试、取消、恢复、审批 checkpoint、幂等从 dispatcher/gateway 中抽象出来。 +3. **Model Gateway 独立化**：把 LLM Pool 升级为模型网关，承载模型路由、用量计量、成本核算、套餐权限、BYOK、fallback。 +4. **多租户/RBAC 前置**：全链路补 `tenant_id / workspace_id / project_id`，避免后期大规模重构。 +5. **Tool / MCP 分层**：拆分 Internal Tools、Remote MCP Gateway、Tool Policy Engine。 +6. **NATS Subject 规范化**：把 Command、Event、Stream、Tool RPC、Control、Health 分区，避免“万物总线”失控。 +7. **Billing Usage Meter 先落事件**：不必马上接 Stripe/支付，但必须先把用量事件和成本账本做出来。 +8. **生产化迁移机制**：逐步替换 GORM AutoMigrate，补充 migration、幂等、审计、数据隔离策略。 + +--- + +## 1. 重构目标 + +### 1.1 产品目标 + +将系统从当前形态： + +```text +Desktop App + Admin Console + Cloud Agent Backend +``` + +升级为： + +```text +Web SaaS Agent Platform ++ Optional Desktop / Local Connector ++ Admin Console +``` + +目标产品能力： + +- 用户可在 Web App 中创建工作区、项目、任务、Agent 图和知识库。 +- Agent 编排默认在服务端执行。 +- 远程 MCP、内部工具、模型调用、计费全部由服务端统一管控。 +- 桌面端不再是唯一用户入口，而是可选的本地能力扩展。 +- 后续可支持 Local Agent / Desktop Connector 处理本地文件、Shell、本地 MCP。 + +### 1.2 技术目标 + +- 保留现有 Go + Eino + NATS + MCP 方向。 +- 降低 Gateway 和 Dispatcher 的职责耦合。 +- 增加 SaaS 必备的租户、计费、权限、审计、模型成本能力。 +- 让 Agent 任务支持暂停、恢复、取消、重试、审批、断点回放。 +- 让模型和工具调用都可计量、可审计、可限流、可降级。 +- 为后续微服务拆分、K8s、NATS 集群、DB HA 做边界准备。 + +--- + +## 2. 当前架构问题清单 + +### 2.1 Client 层定位偏桌面端 + +当前文档中用户工作产品是 Wails 桌面端，Web 管理端仅用于运维控制台。这个定位更适合桌面 Agent，而不是 Web SaaS。 + +**风险：** + +- SaaS 用户入口不清晰。 +- 本地文件能力和云端 Agent 能力边界混合。 +- 后续做团队协作、组织计费、浏览器任务面板会被桌面端假设限制。 + +**重构方向：** + +- Web App 成为主用户产品。 +- Admin Console 保留为管理端。 +- Desktop/Wails 改为 Optional Local Connector。 + +--- + +### 2.2 NATS 承载语义过多 + +当前 NATS 承载任务、工具 RPC、token 流、执行轨迹、控制面广播、心跳、持久队列。 + +**风险：** + +- Subject 命名膨胀。 +- Command/Event/Stream/Query 混用。 +- 任务状态散落在消息链路中。 +- 调试和权限治理变复杂。 + +**重构方向：** + +- 保留 NATS 作为事件和工具总线。 +- 明确 Command Bus、Event Bus、Stream Bus、Tool RPC、Control Plane、Health 的命名规范。 +- 引入 Task Runtime 作为任务生命周期事实源。 + +--- + +### 2.3 Task Runtime 不够独立 + +当前任务生命周期分散在 Gateway、NATS、Dispatcher、Redis Stream、DB FSM、approval checkpoint 中。 + +**风险：** + +- 取消、暂停、恢复、审批重试、Worker 崩溃恢复逻辑分散。 +- 很难统一做任务幂等和任务审计。 +- 后续引入 Temporal 或多 Worker 会迁移困难。 + +**重构方向：** + +- 增加 Task Runtime Layer。 +- MVP 阶段用 PostgreSQL FSM + NATS/JetStream。 +- 生产阶段可演进到 Temporal。 + +--- + +### 2.4 LLM Pool 应升级为 Model Gateway + +当前 LLM Pool 已经具备 failover、缓存、热配置、breaker，但仍偏模型池实现，而不是 SaaS 计费核心。 + +**风险：** + +- 模型能力、价格、用户套餐、token 用量、成本账本分散。 +- 不利于接入多 Provider、BYOK、企业模型网关。 +- 不利于真实计费。 + +**重构方向：** + +- 独立 Model Gateway / Model Runtime 概念。 +- 统一 Provider Adapter、Usage Meter、Cost Calculator、Routing Policy、Capability Registry。 + +--- + +### 2.5 多租户和 RBAC 尚未进入核心数据模型 + +当前文档明确是 `owner_id` 单租户假设，多租户/RBAC 尚未启动。 + +**风险：** + +- 表结构、Redis key、NATS envelope、MinIO object path、Milvus/Neo4j namespace 后期都要重构。 +- 计费、审计、权限无法天然支持组织/团队。 + +**重构方向：** + +- 立即在核心 Contract 和数据表中引入 `tenant_id`。 +- 同步引入 `workspace_id / project_id / user_id / role / permission`。 + +--- + +### 2.6 Tool / MCP 层边界需要重拆 + +当前 mcp-go 同时承载 RAG、记忆、历史、报告、元工具；mcp-py 承载算法工具，但部分为桩。 + +**风险：** + +- 内部工具和远程 MCP 混合。 +- Tool 权限、审计、套餐限制没有独立中枢。 +- Python 算法层如果长期为桩，会影响产品真实可用性。 + +**重构方向：** + +- 拆成 Internal Tool Services、Remote MCP Gateway、Tool Policy Engine。 +- Python 工具必须从“链路已通”走向“至少一条真实可用核心能力”。 + +--- + +## 3. 目标架构总览 + +```mermaid +flowchart TB + +subgraph CLIENT["1. Client Layer"] + WEB["User Web App\nNext.js/React + React Flow\nTask Console / DSL Designer / Artifact Viewer"] + ADMIN["Admin Console\nModel / Billing / Tenant / Audit / Ops"] + LOCAL["Optional Local Connector\nWails/Tauri\nLocal FS / Shell / Local MCP"] +end + +subgraph GATEWAY["2. Gateway Layer"] + GW["Go Gateway\nAuth / RateLimit / Guardrail Tier1\nTask API / SSE / Admin API"] +end + +subgraph TASK["3. Task Runtime Layer"] + FSM["Task FSM\nsubmitted/running/waiting/done/failed"] + APPROVAL["Approval Checkpoint"] + IDEMP["Idempotency / Retry / Cancel / Resume"] + EVENTLOG["Task Event Log"] +end + +subgraph BUS["4. Event Bus Layer"] + NATS["NATS + JetStream"] + REDISSTREAM["Redis Stream\nSSE Replay"] +end + +subgraph DISPATCHER["5. Agent Dispatcher Layer"] + EINO["Eino Compose Engine"] + PLANNER["Planner / Router / Context Manager"] + TOOLRUN["Tool Executor"] + HARNESS["Harness\nGuardrail Tier2 / Eval / Budget / Desensitize"] +end + +subgraph MODEL["6. Model Gateway Layer"] + ROUTER["Routing Policy"] + ADAPTER["Provider Adapters\nOpenAI-compatible / vLLM / Ollama / Future Claude"] + USAGE["Usage Meter / Cost Calculator"] + BREAKER["Failover / Breaker / Cache"] +end + +subgraph TOOL["7. Tool & MCP Layer"] + POLICY["Tool Policy Engine\nAllow/Deny/Approval/Quota/Audit"] + GOTOOLS["Internal Go Tools\nRAG / Memory / History / Report / External API"] + PYTOOLS["Internal Python Tools\nSandbox / Parser / OCR / Code Interpreter"] + REMOTEMCP["Remote MCP Gateway\nOAuth / Registry / Remote Servers"] +end + +subgraph DATA["8. Data Layer"] + PG["PostgreSQL"] + REDIS["Redis"] + MINIO["MinIO"] + MILVUS["Milvus"] + NEO4J["Neo4j"] + JAEGER["Jaeger / OTel"] +end + +WEB --> GW +ADMIN --> GW +LOCAL --> GW +GW --> FSM +FSM --> NATS +NATS --> EINO +EINO --> ROUTER +EINO --> POLICY +POLICY --> GOTOOLS +POLICY --> PYTOOLS +POLICY --> REMOTEMCP +ROUTER --> ADAPTER +ADAPTER --> USAGE +USAGE --> PG +EINO --> NATS +NATS --> REDISSTREAM +REDISSTREAM --> GW +GW --> WEB +GW --> ADMIN +FSM --> PG +EVENTLOG --> PG +GOTOOLS --> MILVUS +GOTOOLS --> NEO4J +GOTOOLS --> MINIO +PYTOOLS --> MINIO +GW --> REDIS +``` + +--- + +## 4. 分层重构设计 + +## 4.1 Client Layer 重构 + +### 当前形态 + +```text +Desktop Wails = 用户工作产品 +Admin React = 运维控制台 +``` + +### 目标形态 + +```text +User Web App = 主产品 +Admin Console = 运维/商业后台 +Optional Local Connector = 本地能力扩展 +``` + +### User Web App 职责 + +- 工作区 / 项目 / 任务管理。 +- Agent DSL 画布编排。 +- 任务运行控制台。 +- Plan 审核、HITL 审批。 +- Token stream / exec event 实时展示。 +- Artifact / Report / Knowledge Base 预览。 +- 团队成员和权限入口。 +- 套餐、用量、账单入口。 + +### Admin Console 职责 + +- 模型配置。 +- Provider 配置。 +- 价格配置。 +- 租户管理。 +- 用户管理。 +- 审计日志。 +- Guardrail 事件。 +- 模型健康和熔断状态。 +- MCP 工具注册与状态。 +- 系统运行 overview。 + +### Optional Local Connector 职责 + +- 本地文件系统访问。 +- 本地 Shell 执行。 +- 本地 Git 操作。 +- 本地 MCP Server 连接。 +- 本地另存为、系统通知、应用打开。 +- 将本地能力以受控 Tool 形式暴露给服务端或本地 Agent。 + +### 重构建议 + +目录可调整为： + +```text +/apps + /web # 用户 Web App + /admin # 管理后台 + /desktop-connector # 可选本地连接器，原 Wails 逐步迁移 +``` + +--- + +## 4.2 Gateway Layer 重构 + +### Gateway 应保留职责 + +- HTTP API 入口。 +- Auth / JWT / Session。 +- Rate Limit。 +- CORS。 +- RequestID / OTel / Audit middleware。 +- Guardrail Tier1。 +- DSL Schema Validation。 +- Task API。 +- SSE/WebSocket Fanout。 +- Admin API。 +- Report download/export endpoint。 + +### Gateway 不应承担职责 + +- 不做 Agent 编排。 +- 不做复杂 DSL execution plan assembly。 +- 不做工具路由。 +- 不做模型路由。 +- 不直接访问内部工具实现。 + +### Gateway 推荐模块结构 + +```text +/services/gateway + /cmd + /internal + /api + /task + /workspace + /project + /billing + /admin + /sse + /middleware + auth.go + ratelimit.go + guardrail.go + audit.go + observe.go + /validator + dsl_schema.go + /client + task_runtime_client.go + event_stream_client.go +``` + +--- + +## 4.3 Task Runtime Layer 新增 + +### 为什么要新增 + +Agent SaaS 的任务不是普通 HTTP 请求，而是长生命周期对象。任务可能经历： + +```text +submitted → queued → running → waiting_approval → running → done +submitted → queued → running → failed → retrying → running → done +submitted → queued → running → cancelled +submitted → queued → running → timeout +``` + +这些状态不应该散落在 Gateway、Dispatcher 和 NATS 消息中。 + +### Task Runtime 职责 + +- Task FSM 状态机。 +- TaskRun 实例管理。 +- StepRun 状态管理。 +- Approval checkpoint。 +- Retry / Cancel / Resume / Timeout。 +- 幂等键管理。 +- Worker lease / heartbeat。 +- Event Log 落库。 +- Task 级并发控制。 +- Tenant 级并发控制。 + +### MVP 实现 + +MVP 不必马上引入 Temporal，可以用： + +```text +PostgreSQL tasks/task_runs/task_events ++ NATS JetStream durable consumer ++ Dispatcher worker heartbeat ++ Redis lock 可选 +``` + +### 生产演进 + +后续可迁移到： + +```text +Temporal Workflow ++ Go Activity Worker ++ Eino Agent Activity ++ NATS Event Fanout +``` + +### 推荐状态机 + +```mermaid +stateDiagram-v2 + [*] --> submitted + submitted --> queued + queued --> running + running --> waiting_approval + waiting_approval --> running: approved + waiting_approval --> rejected: rejected + running --> done + running --> failed + running --> timeout + running --> cancelling + cancelling --> cancelled + failed --> retrying + retrying --> queued + done --> [*] + rejected --> [*] + cancelled --> [*] + timeout --> [*] +``` + +### 核心表草案 + +```sql +CREATE TABLE sundynix_tasks ( + id UUID PRIMARY KEY, + tenant_id UUID NOT NULL, + workspace_id UUID NOT NULL, + project_id UUID, + owner_user_id UUID NOT NULL, + title TEXT NOT NULL, + status TEXT NOT NULL, + dsl JSONB NOT NULL, + current_run_id UUID, + idempotency_key TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + deleted_at TIMESTAMPTZ +); + +CREATE TABLE sundynix_task_runs ( + id UUID PRIMARY KEY, + tenant_id UUID NOT NULL, + task_id UUID NOT NULL, + status TEXT NOT NULL, + attempt INT NOT NULL DEFAULT 1, + started_at TIMESTAMPTZ, + finished_at TIMESTAMPTZ, + error_code TEXT, + error_message TEXT, + worker_id TEXT, + heartbeat_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE sundynix_task_events ( + id UUID PRIMARY KEY, + tenant_id UUID NOT NULL, + task_id UUID NOT NULL, + run_id UUID, + event_type TEXT NOT NULL, + event_payload JSONB NOT NULL, + trace_id TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE sundynix_approval_checkpoints ( + id UUID PRIMARY KEY, + tenant_id UUID NOT NULL, + task_id UUID NOT NULL, + run_id UUID NOT NULL, + node_id TEXT NOT NULL, + status TEXT NOT NULL, + request_payload JSONB NOT NULL, + response_payload JSONB, + requested_by UUID, + approved_by UUID, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + resolved_at TIMESTAMPTZ +); +``` + +--- + +## 4.4 Agent Dispatcher Layer 重构 + +### 保留能力 + +当前 dispatcher 的这些能力建议保留： + +- Eino compose 单编排引擎。 +- DSL → compose 图编译。 +- 分支、并行、map、coordinator、approval。 +- ReAct 工具调用。 +- Token stream 和 exec event。 +- Harness 治理层。 + +### 需要调整的边界 + +Dispatcher 不应该直接承担完整任务生命周期，而应该从 Task Runtime 领取任务执行： + +```text +Task Runtime 发布 RunCommand +→ Dispatcher 执行 Agent Graph +→ Dispatcher 发布 TaskEvent / StreamEvent +→ Task Runtime 更新状态 +``` + +### Dispatcher 推荐职责 + +- 编译 DSL 为 Eino compose graph。 +- 执行 graph。 +- 管理上下文。 +- 调用 Model Gateway。 +- 调用 Tool Policy Engine。 +- 产出 token stream / exec events。 +- 产出节点级执行结果。 +- 产出评测结果。 + +### Dispatcher 不应职责 + +- 不直接做计费结算。 +- 不直接管理租户权限。 +- 不直接存储模型密钥。 +- 不直接暴露 HTTP API。 + +### 推荐接口 + +```go +type AgentExecutor interface { + Execute(ctx context.Context, req ExecuteRequest) (<-chan AgentEvent, error) + Cancel(ctx context.Context, runID string) error + Resume(ctx context.Context, req ResumeRequest) error +} + +type ExecuteRequest struct { + TenantID string + WorkspaceID string + ProjectID string + TaskID string + RunID string + UserID string + DSL DSLGraph + Input map[string]any + Policy ExecutionPolicy +} +``` + +--- + +## 4.5 Model Gateway Layer 新增 + +### 为什么独立 + +模型调用是 SaaS 成本、体验和商业化的核心，不应该只是 dispatcher 内部的 `llm/pool.go`。 + +Model Gateway 要统一处理： + +- Provider Adapter。 +- 模型能力注册。 +- 模型路由。 +- 套餐可用模型。 +- BYOK。 +- Usage metering。 +- Cost calculation。 +- Failover。 +- Circuit breaker。 +- Cache。 +- Structured output 适配。 +- Tool calling 适配。 +- Prompt caching 预留。 + +### 目标结构 + +```text +Model Gateway +├── Provider Registry +├── Capability Registry +├── Routing Policy +├── Provider Adapter +├── Usage Meter +├── Cost Calculator +├── Failover Manager +├── Breaker Manager +├── Cache Layer +└── Admin Runtime State +``` + +### Provider Adapter 接口建议 + +```go +type ModelProvider interface { + Name() string + Chat(ctx context.Context, req ChatRequest) (*ChatResponse, error) + Stream(ctx context.Context, req ChatRequest) (<-chan ChatStreamEvent, error) + CountTokens(ctx context.Context, req TokenRequest) (*TokenUsage, error) + Capabilities(ctx context.Context, model string) ModelCapabilities +} + +type ModelCapabilities struct { + SupportsTools bool + SupportsVision bool + SupportsJSONMode bool + SupportsReasoning bool + SupportsEmbeddings bool + MaxContextTokens int + MaxOutputTokens int +} +``` + +### Usage Event 表 + +```sql +CREATE TABLE sundynix_model_usage_events ( + id UUID PRIMARY KEY, + tenant_id UUID NOT NULL, + user_id UUID NOT NULL, + task_id UUID, + run_id UUID, + provider TEXT NOT NULL, + model TEXT NOT NULL, + input_tokens BIGINT NOT NULL DEFAULT 0, + output_tokens BIGINT NOT NULL DEFAULT 0, + cached_tokens BIGINT NOT NULL DEFAULT 0, + cost_usd NUMERIC(18,8) NOT NULL DEFAULT 0, + latency_ms BIGINT, + status TEXT NOT NULL, + error_code TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +``` + +### 路由策略示例 + +```text +route_policy: + tenant_plan: pro + task_type: report_generation + preferred_models: + - deepseek-chat + - qwen-plus + fallback_models: + - local-vllm-qwen + max_cost_usd: 0.20 + max_latency_ms: 30000 + require_tool_calling: true +``` + +--- + +## 4.6 Tool / MCP Layer 重构 + +### 目标分层 + +```text +Tool & MCP Layer +├── Tool Policy Engine +├── Internal Go Tools +├── Internal Python Tools +└── Remote MCP Gateway +``` + +### Tool Policy Engine 职责 + +每一次工具调用前，都必须经过 Tool Policy Engine： + +```text +Dispatcher → Tool Policy Engine → Tool Service / Remote MCP +``` + +策略结果： + +```text +allow + deny + require_approval + quota_exceeded + plan_required + rate_limited +``` + +### Tool Policy 请求结构 + +```go +type ToolPolicyRequest struct { + TenantID string + WorkspaceID string + ProjectID string + UserID string + TaskID string + RunID string + ToolName string + ToolType string // internal_go, internal_py, remote_mcp, local_connector + Arguments map[string]any + RiskLevel string // low, medium, high, critical +} +``` + +### Tool Audit 表 + +```sql +CREATE TABLE sundynix_tool_call_events ( + id UUID PRIMARY KEY, + tenant_id UUID NOT NULL, + user_id UUID NOT NULL, + task_id UUID, + run_id UUID, + tool_name TEXT NOT NULL, + tool_type TEXT NOT NULL, + risk_level TEXT, + policy_decision TEXT NOT NULL, + input_hash TEXT, + output_hash TEXT, + latency_ms BIGINT, + status TEXT NOT NULL, + error_code TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +``` + +### Internal Go Tools + +适合继续由 Go 承载： + +- RAG search。 +- Wiki search。 +- Memory。 +- History。 +- Report render/export。 +- External API connectors。 +- Metadata tools。 +- Health/list_tools。 + +### Internal Python Tools + +必须从桩逐步真实化： + +- P0：Secure Code Sandbox 真实可用。 +- P0：Code Interpreter 至少支持 Python 片段执行 + 超时 + 输出限制。 +- P1：MinerU / OCR 文档解析真实可用。 +- P1：多模态解析任务异步化。 + +### Remote MCP Gateway + +Remote MCP Gateway 单独负责： + +- MCP Server Registry。 +- OAuth token vault。 +- Tool discovery。 +- Tool allowlist / denylist。 +- Per-tenant tool policy。 +- Per-plan tool availability。 +- Remote tool audit。 +- Remote tool quota。 + +--- + +## 4.7 NATS Subject 规范化 + +### 当前问题 + +当前 NATS 是统一总线，但 subject 分类需要更严格，否则后期难以运维。 + +### 推荐 Subject 命名 + +```text +# Command +sundynix.cmd.task.submit +sundynix.cmd.task.cancel +sundynix.cmd.task.resume +sundynix.cmd.task.approve +sundynix.cmd.task.reject + +# Event +sundynix.events.task.created +sundynix.events.task.queued +sundynix.events.task.started +sundynix.events.task.waiting_approval +sundynix.events.task.completed +sundynix.events.task.failed +sundynix.events.task.cancelled + +# Stream +sundynix.streams.task..token +sundynix.streams.task..exec +sundynix.streams.kb..progress + +# Tool RPC +sundynix.tools.go. +sundynix.tools.py. +sundynix.tools.remote_mcp.. + +# Control Plane +sundynix.control.model.updated +sundynix.control.prompt.activated +sundynix.control.tool.registry_updated +sundynix.control.policy.updated + +# Health +sundynix.health.gateway +sundynix.health.dispatcher +sundynix.health.mcp_go +sundynix.health.mcp_py +sundynix.health.model_gateway +``` + +### Envelope 标准 + +```go +type Envelope[T any] struct { + ID string `json:"id"` + Type string `json:"type"` + Version string `json:"version"` + TenantID string `json:"tenant_id"` + WorkspaceID string `json:"workspace_id,omitempty"` + ProjectID string `json:"project_id,omitempty"` + UserID string `json:"user_id,omitempty"` + TaskID string `json:"task_id,omitempty"` + RunID string `json:"run_id,omitempty"` + TraceID string `json:"trace_id,omitempty"` + CreatedAt time.Time `json:"created_at"` + Payload T `json:"payload"` +} +``` + +### 使用原则 + +- Command：表示“请求系统执行某动作”。 +- Event：表示“某事实已经发生”。 +- Stream：表示“高频、可回放、面向前端展示的事件流”。 +- Tool RPC：表示“内部工具请求-响应”。 +- Control：表示“运行时配置变更广播”。 +- Health：表示“探活和状态查询”。 + +--- + +## 5. 多租户与 RBAC 重构 + +## 5.1 为什么必须前置 + +Web SaaS 的最小商业单位不是 user，而是 tenant / organization。 + +如果后期再补多租户，会影响： + +- PostgreSQL 表结构。 +- Redis key。 +- NATS envelope。 +- MinIO object path。 +- Milvus collection/partition。 +- Neo4j node/edge namespace。 +- MCP token。 +- 模型 Key。 +- 审计日志。 +- 计费账本。 + +### 5.2 基础租户模型 + +```text +Tenant + └── Workspace + └── Project + └── Task + └── TaskRun +``` + +### 5.3 推荐核心表 + +```sql +CREATE TABLE sundynix_tenants ( + id UUID PRIMARY KEY, + name TEXT NOT NULL, + slug TEXT UNIQUE NOT NULL, + plan TEXT NOT NULL DEFAULT 'free', + status TEXT NOT NULL DEFAULT 'active', + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE sundynix_tenant_members ( + id UUID PRIMARY KEY, + tenant_id UUID NOT NULL, + user_id UUID NOT NULL, + role TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'active', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE(tenant_id, user_id) +); + +CREATE TABLE sundynix_workspaces ( + id UUID PRIMARY KEY, + tenant_id UUID NOT NULL, + name TEXT NOT NULL, + created_by UUID NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE sundynix_projects ( + id UUID PRIMARY KEY, + tenant_id UUID NOT NULL, + workspace_id UUID NOT NULL, + name TEXT NOT NULL, + description TEXT, + created_by UUID NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +``` + +### 5.4 RBAC 初版角色 + +```text +owner +admin +member +viewer +billing_admin +``` + +### 5.5 权限动作示例 + +```text +tenant.manage +billing.manage +workspace.create +project.create +task.create +task.run +task.cancel +task.approve +model.use +model.manage +tool.use +tool.manage +kb.read +kb.write +audit.read +``` + +### 5.6 数据隔离规则 + +- 所有核心表必须有 `tenant_id`。 +- 所有查询必须默认带 `tenant_id`。 +- 所有 Redis key 带 tenant 前缀。 +- 所有 NATS envelope 带 `tenant_id`。 +- 所有 MinIO path 带 `tenant_id/workspace_id/project_id`。 +- Milvus 推荐使用 tenant partition 或 metadata filter。 +- Neo4j 节点和关系必须带 `tenant_id`。 + +Redis Key 示例： + +```text +sundynix:{tenant_id}:task:{task_id}:stream +sundynix:{tenant_id}:rate:{user_id} +sundynix:{tenant_id}:session:{session_id} +``` + +MinIO Path 示例： + +```text +tenants/{tenant_id}/workspaces/{workspace_id}/projects/{project_id}/artifacts/{artifact_id} +``` + +--- + +## 6. Billing 与 Usage Meter 重构 + +## 6.1 原则 + +第一阶段不一定要接支付，但必须先做 Usage Meter。 + +真实计费依赖以下用量： + +- 模型 input/output/cached token。 +- 模型成本。 +- 工具调用次数。 +- 远程 MCP 调用次数。 +- Code Sandbox 执行时长。 +- 文档解析页数。 +- 向量入库 token/条数。 +- 存储空间。 +- 任务运行时长。 + +### 6.2 Billing Event 表 + +```sql +CREATE TABLE sundynix_usage_events ( + id UUID PRIMARY KEY, + tenant_id UUID NOT NULL, + user_id UUID, + task_id UUID, + run_id UUID, + event_type TEXT NOT NULL, + resource_type TEXT NOT NULL, + resource_name TEXT, + quantity NUMERIC(18,6) NOT NULL, + unit TEXT NOT NULL, + unit_price NUMERIC(18,8), + cost NUMERIC(18,8), + currency TEXT DEFAULT 'USD', + metadata JSONB, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +``` + +### 6.3 账单聚合表 + +```sql +CREATE TABLE sundynix_usage_daily_rollups ( + id UUID PRIMARY KEY, + tenant_id UUID NOT NULL, + date DATE NOT NULL, + resource_type TEXT NOT NULL, + quantity NUMERIC(18,6) NOT NULL, + cost NUMERIC(18,8) NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE(tenant_id, date, resource_type) +); +``` + +### 6.4 计费阶段 + +```text +阶段 1：仅记录 usage_events。 +阶段 2：按天 rollup，后台展示成本。 +阶段 3：套餐 quota + 超额限制。 +阶段 4：接支付系统。 +阶段 5：企业账单、发票、成本中心。 +``` + +--- + +## 7. 数据流重构 + +## 7.1 Task Run Flow + +```mermaid +sequenceDiagram + participant U as Web App + participant G as Gateway + participant T as Task Runtime + participant N as NATS + participant D as Dispatcher + participant M as Model Gateway + participant P as Tool Policy + participant Tool as Tool/MCP + participant R as Redis Stream + + U->>G: POST /tasks with DSL + G->>G: Auth + RateLimit + Guardrail Tier1 + DSL Schema Validation + G->>T: Create Task + T->>T: Create TaskRun + FSM=submitted/queued + T->>N: Publish cmd.task.run + D->>N: Subscribe cmd.task.run + D->>D: Compile DSL to Eino Graph + D->>M: Stream Chat / Tool Calling + D->>P: Check Tool Policy + P->>Tool: Allow and Call Tool + D->>N: Publish task events/token stream + N->>R: Persist stream events + G->>R: Read Redis Stream + G->>U: SSE token/exec events + D->>T: Report final status + T->>T: FSM=done/failed/timeout +``` + +--- + +## 7.2 Approval Flow + +```mermaid +sequenceDiagram + participant D as Dispatcher + participant T as Task Runtime + participant R as Redis Stream + participant G as Gateway + participant U as Web App + + D->>T: Create Approval Checkpoint + T->>T: FSM=waiting_approval + T->>R: Event approval.required + G->>R: SSE read + G->>U: Show approval card + U->>G: POST /tasks/:id/approvals/:approval_id/approve + G->>T: Approve checkpoint + T->>T: FSM=queued/running + T->>D: Resume run command +``` + +--- + +## 7.3 Model Usage Flow + +```mermaid +sequenceDiagram + participant D as Dispatcher + participant M as Model Gateway + participant P as Provider + participant B as Usage Meter + participant DB as PostgreSQL + + D->>M: Chat/Stream request + M->>M: Route model + check plan + budget + M->>P: Provider API + P-->>M: Stream/Response + usage + M->>B: Emit model_usage_event + B->>DB: Store usage/cost + M-->>D: Return response +``` + +--- + +## 7.4 Tool Call Flow + +```mermaid +sequenceDiagram + participant D as Dispatcher + participant P as Tool Policy Engine + participant A as Approval Runtime + participant Tool as Internal/Remote Tool + participant Audit as Tool Audit + + D->>P: tool_call_request + P->>P: Check tenant/plan/risk/quota + alt allow + P->>Tool: Call tool + Tool-->>P: Tool result + P->>Audit: Write allow event + P-->>D: result + else require approval + P->>A: Create approval checkpoint + P-->>D: wait approval + else deny + P->>Audit: Write deny event + P-->>D: policy denied + end +``` + +--- + +## 8. DSL 重构建议 + +### 8.1 DSL 增加版本和执行策略 + +```json +{ + "version": "2.0", + "kind": "agent_graph", + "metadata": { + "name": "Research Report Agent", + "description": "Generate report from KB and web tools" + }, + "execution_policy": { + "timeout_seconds": 1800, + "max_retries": 2, + "max_cost_usd": 0.5, + "require_approval_for_high_risk_tools": true + }, + "nodes": [], + "edges": [] +} +``` + +### 8.2 Node 增加风险和权限声明 + +```json +{ + "id": "tool_1", + "kind": "tool", + "label": "Search Wiki", + "config": { + "tool_name": "kb_search", + "risk_level": "low", + "requires_approval": false, + "timeout_seconds": 30 + } +} +``` + +### 8.3 DSL 校验分层 + +Gateway 只做： + +- JSON schema 校验。 +- 节点 ID 重复。 +- 悬挂边。 +- 节点数量限制。 +- 边数量限制。 +- 不允许的节点 kind。 + +Dispatcher 做： + +- 编译为 Eino graph。 +- execution plan assembly。 +- branch default 检查。 +- map 并发策略。 +- tool availability 检查。 +- model capability 检查。 + +--- + +## 9. 目录结构重构建议 + +```text +sundynix-agentix/ + apps/ + web/ + admin/ + desktop-connector/ + + services/ + gateway/ + task-runtime/ + dispatcher/ + model-gateway/ + tool-policy/ + mcp-go/ + mcp-py/ + + shared/ + contract/ + envelope.go + task.go + dsl.go + model.go + tool.go + billing.go + tenant.go + bus/ + nats.go + subjects.go + jetstream.go + auth/ + otel/ + secrets/ + prompts/ + errors/ + + deployments/ + docker-compose/ + k8s/ + + migrations/ + postgres/ + + docs/ + architecture/ + api/ + dsl/ + operations/ +``` + +MVP 阶段不一定物理拆成所有服务，但代码目录应按目标边界组织。 + +--- + +## 10. 分阶段实施计划 + +## Phase 0：架构边界整理，1–2 周 + +目标：不大改功能，先把概念和 contract 对齐。 + +任务： + +- [ ] 在架构文档中新增 Task Runtime Layer。 +- [ ] 在架构文档中新增 Model Gateway Layer。 +- [ ] 在架构文档中拆分 Internal Tools / Remote MCP Gateway / Tool Policy Engine。 +- [ ] 统一 NATS Subject 命名规范。 +- [ ] 定义 Envelope v1。 +- [ ] 核心 contract 增加 `tenant_id/workspace_id/project_id` 字段。 +- [ ] 明确 Web App、Admin Console、Desktop Connector 三类客户端边界。 + +交付物： + +- `docs/architecture/ARCHITECTURE_DESIGN_V2.md` +- `shared/contract/envelope.go` +- `shared/bus/subjects.go` + +--- + +## Phase 1：多租户骨架与 Task Runtime，2–4 周 + +目标：让任务运行具备 SaaS 数据隔离和统一状态机。 + +任务： + +- [ ] 新增 tenant/workspace/project 表。 +- [ ] 核心任务表补 `tenant_id/workspace_id/project_id`。 +- [ ] Redis key 加 tenant namespace。 +- [ ] NATS envelope 强制带 tenant_id。 +- [ ] 新增 task_runs/task_events/approval_checkpoints。 +- [ ] Gateway 创建任务改为写 Task Runtime。 +- [ ] Dispatcher 从 Task Runtime command 中领取任务。 +- [ ] Task 状态变更统一落 task_events。 +- [ ] SSE 从 Redis Stream + task_events 恢复历史。 + +交付物： + +- `services/task-runtime` +- `migrations/postgres/001_tenant_task_runtime.sql` +- 任务运行状态机测试。 + +--- + +## Phase 2：Model Gateway 与 Usage Meter，2–4 周 + +目标：模型调用从“能用”升级为“可商业化”。 + +任务： + +- [ ] 从 dispatcher 中抽出 model runtime。 +- [ ] 定义 ModelProvider 接口。 +- [ ] 实现 OpenAI-compatible Provider。 +- [ ] 实现 vLLM/Ollama Provider。 +- [ ] 增加 model_capabilities 表。 +- [ ] 增加 model_usage_events 表。 +- [ ] 增加价格配置表。 +- [ ] 增加 route policy。 +- [ ] Admin 展示模型健康、熔断状态、成本。 +- [ ] 每次模型调用写 usage event。 + +交付物： + +- `services/model-gateway` +- `sundynix_model_usage_events` +- Admin Model Runtime 页面。 + +--- + +## Phase 3：Tool Policy 与 Remote MCP Gateway，2–5 周 + +目标：工具调用可管控、可审批、可审计、可计费。 + +任务： + +- [ ] 新增 Tool Policy Engine。 +- [ ] 每次工具调用前经过 policy check。 +- [ ] 高风险工具支持 require_approval。 +- [ ] 工具调用写 `tool_call_events`。 +- [ ] mcp-go 工具注册增加 risk_level、required_plan、permissions。 +- [ ] mcp-py 至少真实化一个核心工具。 +- [ ] Remote MCP Gateway 独立出 registry、OAuth、audit。 +- [ ] Admin 增加 Tool Registry / Tool Audit 页面。 + +交付物： + +- `services/tool-policy` +- `sundynix_tool_call_events` +- Tool risk policy 文档。 + +--- + +## Phase 4：Web App 主产品化，3–6 周 + +目标：让 Web App 成为主产品入口。 + +任务： + +- [ ] 新建或迁移 User Web App。 +- [ ] React Flow DSL Designer Web 化。 +- [ ] Task Run Console。 +- [ ] Artifact Viewer。 +- [ ] Approval UI。 +- [ ] Usage / Billing 页面。 +- [ ] Workspace / Project / Team 页面。 +- [ ] Desktop Connector 改为可选本地能力。 + +交付物： + +- `/apps/web` +- 用户任务闭环：创建 → 执行 → 审批 → 查看结果 → 查看用量。 + +--- + +## Phase 5：生产硬化，持续 + +目标：上线准备。 + +任务： + +- [ ] AutoMigrate 替换为 migration 工具。 +- [ ] NATS 集群。 +- [ ] PostgreSQL 备份/恢复。 +- [ ] Redis HA。 +- [ ] MinIO bucket policy。 +- [ ] TLS。 +- [ ] Secret rotation。 +- [ ] OTel 指标完善。 +- [ ] Prometheus/Grafana。 +- [ ] Admin 熔断态和失败原因可视化。 +- [ ] 数据删除和租户注销流程。 + +--- + +## 11. MVP 收敛范围 + +为了避免重构过度，MVP 建议保留以下能力： + +```text +必须做： +- Web App 任务入口 +- Gateway +- Task Runtime DB FSM +- Dispatcher + Eino +- Model Gateway 基础版 +- OpenAI-compatible Provider +- NATS + Redis Stream +- mcp-go 基础工具 +- tenant_id 全链路 +- usage_events 记录 +- SSE 任务流 + +暂缓： +- Temporal +- NATS 集群 +- 完整 K8s +- 完整计费支付 +- 完整 mcp-py 算法集群 +- 完整多模型市场 +- 完整 prompt 灰度实验 +- 完整企业审计 +``` + +--- + +## 12. 风险与规避 + +| 风险 | 影响 | 规避 | +|---|---|---| +| 继续以桌面端为主产品 | Web SaaS 定位不清 | Web App 主产品化，Desktop Connector 可选 | +| NATS 语义过载 | 后期难维护 | subject 分区 + envelope 标准 | +| 任务状态散落 | 取消/恢复/审批困难 | Task Runtime 独立 | +| 模型调用不可计量 | 无法商业化 | Model Gateway + usage_events | +| 没有 tenant_id | SaaS 后期重构巨大 | 立即补全 tenant/workspace/project | +| 工具调用无策略 | 安全和套餐不可控 | Tool Policy Engine | +| mcp-py 长期为桩 | 产品能力虚 | P0 真实化 sandbox 或 parser | +| AutoMigrate 上生产 | 数据迁移不可控 | goose/atlas/migrate | +| Guardrail 只在入口 | 工具风险失控 | Tool-call 前置 policy + output guard | + +--- + +## 13. 最终建议 + +这次重构不是推翻当前架构，而是做一次 SaaS 化升级： + +```text +保留： +- Go 主后端 +- Gateway 接入层 +- Eino Dispatcher +- NATS + JetStream +- Redis Stream 回放 +- mcp-go/mcp-py 工具化 +- Harness 治理层 +- 控制面热切换 + +新增/强化： +- User Web App 主产品 +- Task Runtime +- Model Gateway +- Tenant/RBAC +- Usage Meter +- Tool Policy Engine +- Remote MCP Gateway +- NATS Subject 规范 +- Migration 生产化 +``` + +优先级最高的不是继续堆 Agent 能力，而是先把 SaaS 底座补齐： + +```text +P0：tenant_id 全链路 + Task Runtime + Model Gateway + Usage Meter +P1：Tool Policy + Remote MCP + Web App 主产品化 +P2：生产硬化 + Temporal/K8s/NATS Cluster +``` + +完成这次重构后，sundynix-agentix 将从一个能力很强的内部/桌面混合 Agent 平台，升级为一个具备商业化基础的 Web SaaS Agent 平台。 diff --git a/sundynix-dispatcher/cmd/dispatcher/main.go b/sundynix-dispatcher/cmd/dispatcher/main.go index e9d4e83..3fd4c2e 100644 --- a/sundynix-dispatcher/cmd/dispatcher/main.go +++ b/sundynix-dispatcher/cmd/dispatcher/main.go @@ -94,6 +94,7 @@ func main() { "model": pool.ModelName(), "ready": pool.Ready(), "uptime_s": int(time.Since(startedAt).Seconds()), + "models": pool.ModelHealth(), // 主备链每模型实时健康/熔断态（供管理端展示） }) return data }); herr != nil { diff --git a/sundynix-dispatcher/internal/harness/circuitbreaker.go b/sundynix-dispatcher/internal/harness/circuitbreaker.go index 6d9b715..eac9b64 100644 --- a/sundynix-dispatcher/internal/harness/circuitbreaker.go +++ b/sundynix-dispatcher/internal/harness/circuitbreaker.go @@ -139,3 +139,16 @@ func (c *CircuitBreaker) State() State { defer c.mu.Unlock() return c.state } + +// Snapshot 是熔断器的只读观测快照（供 surface 到管理端）。 +type Snapshot struct { + State State // 当前状态：closed / open / half-open + Fails int // 闭合态连续失败计数 +} + +// Snapshot 返回当前观测快照（不改状态机；管理端展示 breaker 态用）。 +func (c *CircuitBreaker) Snapshot() Snapshot { + c.mu.Lock() + defer c.mu.Unlock() + return Snapshot{State: c.state, Fails: c.fails} +} diff --git a/sundynix-dispatcher/internal/harness/circuitbreaker_test.go b/sundynix-dispatcher/internal/harness/circuitbreaker_test.go index 51bff92..7e89360 100644 --- a/sundynix-dispatcher/internal/harness/circuitbreaker_test.go +++ b/sundynix-dispatcher/internal/harness/circuitbreaker_test.go @@ -15,6 +15,24 @@ func newTestCB(threshold int, cooldown time.Duration, clock *time.Time) *Circuit return c } +// Snapshot 应如实反映当前状态与连续失败计数（管理端展示用，不改状态机）。 +func TestCircuitBreaker_Snapshot(t *testing.T) { + now := time.Unix(0, 0) + c := newTestCB(3, 10*time.Second, &now) + if s := c.Snapshot(); s.State != Closed || s.Fails != 0 { + t.Fatalf("初始应 closed/0, got %+v", s) + } + c.Report(false) + c.Report(false) + if s := c.Snapshot(); s.State != Closed || s.Fails != 2 { + t.Fatalf("2 次失败未到阈值应 closed/2, got %+v", s) + } + c.Report(false) // 第 3 次 → 熔断 + if s := c.Snapshot(); s.State != Open || s.Fails != 3 { + t.Fatalf("阈值后应 open/3, got %+v", s) + } +} + func TestCircuitBreaker_OpensAfterThreshold(t *testing.T) { now := time.Unix(0, 0) c := newTestCB(3, 10*time.Second, &now) diff --git a/sundynix-dispatcher/internal/llm/failover.go b/sundynix-dispatcher/internal/llm/failover.go index 1e1d0f9..00e8802 100644 --- a/sundynix-dispatcher/internal/llm/failover.go +++ b/sundynix-dispatcher/internal/llm/failover.go @@ -38,7 +38,8 @@ type failoverModel struct { } // newFailoverModel 建主备链。models 至少 1 个；只有 1 个时调用方应直接用该模型而非本包装。 -func newFailoverModel(models []model.ToolCallingChatModel, onFailover func(int, error)) model.ToolCallingChatModel { +// 返回具体类型 *failoverModel（仍满足接口）：同包的 buildWithFallbacks 据此读 .breakers 上报健康态。 +func newFailoverModel(models []model.ToolCallingChatModel, onFailover func(int, error)) *failoverModel { breakers := make([]*harness.CircuitBreaker, len(models)) for i := range breakers { breakers[i] = harness.NewCircuitBreakerWith(fbBreakerThreshold, fbBreakerCooldown, fbBreakerHalfOpen) diff --git a/sundynix-dispatcher/internal/llm/pool.go b/sundynix-dispatcher/internal/llm/pool.go index 57fed01..5950c81 100644 --- a/sundynix-dispatcher/internal/llm/pool.go +++ b/sundynix-dispatcher/internal/llm/pool.go @@ -19,10 +19,20 @@ import ( "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" + "github.com/sundynix/sundynix-dispatcher/internal/harness" "github.com/sundynix/sundynix-shared/contract" "github.com/sundynix/sundynix-shared/otelx" ) +// ModelHealth 是主备链上单个模型的实时健康态（供管理端展示 failover/熔断可见）。 +type ModelHealth struct { + Provider string `json:"provider"` + Model string `json:"model"` + Role string `json:"role"` // primary / fallback + State string `json:"state"` // closed(在线) / open(熔断中) / half-open(半开探测) / single(无备用链) + Fails int `json:"fails"` // 闭合态连续失败计数 +} + // requestTimeout 是单次推理请求的上限。 const requestTimeout = 120 * time.Second @@ -34,9 +44,11 @@ type ChatMessage struct { // Pool 维护当前激活的后端配置 + 据此构建的 Eino ChatModel（控制面经 NATS 下发，可热更新）。 type Pool struct { - mu sync.RWMutex - cfg *contract.ModelConfig - cm model.BaseChatModel // 由 SetConfig 用激活配置构建；未配置时为 nil + mu sync.RWMutex + cfg *contract.ModelConfig + cm model.BaseChatModel // 由 SetConfig 用激活配置构建；未配置时为 nil + health []ModelHealth // 主备链各模型静态信息（provider/model/role），与 breakers 同序 + breakers []*harness.CircuitBreaker // 与 health 一一对应；无 failover 链时为对应 nil } func NewPool() *Pool { return &Pool{} } @@ -48,12 +60,16 @@ func forceStub() bool { return os.Getenv("LLM_FORCE_STUB") == "1" } // SetConfig 热更新后端配置：用激活配置（含备用模型）重建 ChatModel（控制面变更时调用）。 func (p *Pool) SetConfig(cfg *contract.ModelConfig) { var cm model.BaseChatModel + var health []ModelHealth + var breakers []*harness.CircuitBreaker if cfg != nil && cfg.Ready() && !forceStub() { - cm = buildWithFallbacks(cfg) + cm, health, breakers = buildWithFallbacks(cfg) } p.mu.Lock() p.cfg = cfg p.cm = cm + p.health = health + p.breakers = breakers p.mu.Unlock() if cfg != nil { // 不打印 api_key。 @@ -64,19 +80,24 @@ func (p *Pool) SetConfig(cfg *contract.ModelConfig) { // buildWithFallbacks 构建主模型，并把可用的备用模型串成 failover 链（无备用则直接返回主模型）。 // 主模型构建失败 → 返回 nil（降级桩）；备用单个失败 → 跳过该备用，不影响主链。 -func buildWithFallbacks(cfg *contract.ModelConfig) model.BaseChatModel { +// 第二/三返回值为主备链的健康态元信息（provider/model/role）与对应熔断器（无 failover 链时为 nil）， +// 供 Pool.ModelHealth 上报管理端。 +func buildWithFallbacks(cfg *contract.ModelConfig) (model.BaseChatModel, []ModelHealth, []*harness.CircuitBreaker) { primary, err := buildChatModel(cfg) if err != nil { fmt.Printf("[llm] 构建主 ChatModel 失败（降级桩运行）: %v\n", err) - return nil + return nil, nil, nil } ptcm, ok := primary.(model.ToolCallingChatModel) if !ok { - return primary // 不支持 WithTools（无法包 failover/cache）→ 直接用主模型 + // 不支持 WithTools（无法包 failover/cache）→ 单模型，无独立熔断器。 + return primary, + []ModelHealth{{Provider: cfg.Provider, Model: cfg.Model, Role: "primary"}}, + []*harness.CircuitBreaker{nil} } - // 主链：主模型 +（可用的）备用模型串成 failover。 - chain := ptcm + // 主链：主模型 +（可用的）备用模型串成 failover。metas 与 models 同序。 models := []model.ToolCallingChatModel{ptcm} + metas := []ModelHealth{{Provider: cfg.Provider, Model: cfg.Model, Role: "primary"}} for i := range cfg.Fallbacks { fb := cfg.Fallbacks[i] if !fb.Ready() { @@ -89,16 +110,39 @@ func buildWithFallbacks(cfg *contract.ModelConfig) model.BaseChatModel { } if t, ok := fbm.(model.ToolCallingChatModel); ok { models = append(models, t) + metas = append(metas, ModelHealth{Provider: fb.Provider, Model: fb.Model, Role: "fallback"}) } } + var chain model.ToolCallingChatModel = ptcm + breakers := make([]*harness.CircuitBreaker, len(models)) // 无 failover 时全 nil if len(models) > 1 { fmt.Printf("[llm] 启用模型 failover：主 %s + %d 个备用\n", cfg.Model, len(models)-1) - chain = newFailoverModel(models, func(idx int, ferr error) { + fm := newFailoverModel(models, func(idx int, ferr error) { fmt.Printf("[llm] 模型 failover：第 %d 个模型失败(%v)，切下一个\n", idx, ferr) }) + copy(breakers, fm.breakers) // 每模型熔断器（同序），供上报态 + chain = fm } // 缓存包在最外层：命中直接跳过整条 failover 链（省成本+提速）。键含模型名 → 换模型自然失效。 - return withCache(chain, cfg.Model) + return withCache(chain, cfg.Model), metas, breakers +} + +// ModelHealth 返回当前主备链各模型的实时健康态（含每模型熔断状态），供管理端展示。 +func (p *Pool) ModelHealth() []ModelHealth { + p.mu.RLock() + defer p.mu.RUnlock() + out := make([]ModelHealth, len(p.health)) + for i, h := range p.health { + out[i] = h + if i < len(p.breakers) && p.breakers[i] != nil { + snap := p.breakers[i].Snapshot() + out[i].State = snap.State.String() + out[i].Fails = snap.Fails + } else { + out[i].State = "single" // 无备用链 → 该模型无独立熔断器 + } + } + return out } // buildChatModel 据 provider 归一化连接参数后构建 OpenAI 兼容 ChatModel。 diff --git a/sundynix-dispatcher/internal/llm/pool_health_test.go b/sundynix-dispatcher/internal/llm/pool_health_test.go new file mode 100644 index 0000000..0d73606 --- /dev/null +++ b/sundynix-dispatcher/internal/llm/pool_health_test.go @@ -0,0 +1,45 @@ +package llm + +import ( + "testing" + + "github.com/sundynix/sundynix-shared/contract" +) + +// Pool.ModelHealth 应把主备链各模型的名字/角色与其熔断态正确配对上报。 +func TestPool_ModelHealth(t *testing.T) { + p := NewPool() + + // 主 + 1 备 → failover 链，两模型都在线(closed)。 + p.SetConfig(&contract.ModelConfig{ + Provider: "openai-compatible", BaseURL: "http://primary", APIKey: "k", Model: "primary-m", + Fallbacks: []contract.ModelConfig{ + {Provider: "openai-compatible", BaseURL: "http://backup", APIKey: "k", Model: "backup-m"}, + }, + }) + h := p.ModelHealth() + if len(h) != 2 { + t.Fatalf("主+备应 2 条, got %d: %+v", len(h), h) + } + if h[0].Role != "primary" || h[0].Model != "primary-m" || h[0].State != "closed" { + t.Errorf("主模型态错: %+v", h[0]) + } + if h[1].Role != "fallback" || h[1].Model != "backup-m" || h[1].State != "closed" { + t.Errorf("备模型态错: %+v", h[1]) + } + + // 单模型 → 无 failover 链，state=single。 + p.SetConfig(&contract.ModelConfig{ + Provider: "openai-compatible", BaseURL: "http://only", APIKey: "k", Model: "only-m", + }) + h = p.ModelHealth() + if len(h) != 1 || h[0].State != "single" || h[0].Role != "primary" { + t.Fatalf("单模型应 1 条 single/primary, got %+v", h) + } + + // 未配置 → 空。 + p.SetConfig(nil) + if h := p.ModelHealth(); len(h) != 0 { + t.Fatalf("未配置应空, got %+v", h) + } +} diff --git a/sundynix-gateway/internal/handler/admin.go b/sundynix-gateway/internal/handler/admin.go index 222104d..a97bf0f 100644 --- a/sundynix-gateway/internal/handler/admin.go +++ b/sundynix-gateway/internal/handler/admin.go @@ -119,6 +119,30 @@ func (h *Handler) AdminOverview(c *gin.Context) { } } + // 模型运行时健康态：Ping dispatcher 心跳取每模型 failover/熔断态（在线/熔断中/半开）。 + // 权威配置来自 DB（上面 chat/emb），运行时态只有 dispatcher 知道 → 二者互补。 + modelHealth := []gin.H{} + dctx, dcancel := context.WithTimeout(ctx, 2*time.Second) // 独立超时，不与 mcp-go 探活共用 cctx（避免被挤掉） + defer dcancel() + if data, err := h.bus.Ping(dctx, contract.SubjectHealthDispatcher); err == nil && len(data) > 0 { + var dh struct { + Models []struct { + Provider string `json:"provider"` + Model string `json:"model"` + Role string `json:"role"` + State string `json:"state"` + Fails int `json:"fails"` + } `json:"models"` + } + if json.Unmarshal(data, &dh) == nil { + for _, m := range dh.Models { + modelHealth = append(modelHealth, gin.H{ + "provider": m.Provider, "model": m.Model, "role": m.Role, "state": m.State, "fails": m.Fails, + }) + } + } + } + c.JSON(http.StatusOK, gin.H{ "users": users, "kb_count": kbs, "kb_docs": docs, "tasks_today": ov.TasksToday, "tasks_total": ov.TasksTotal, @@ -127,6 +151,7 @@ func (h *Handler) AdminOverview(c *gin.Context) { "models": gin.H{ "chat_count": len(chat), "embedding_count": len(emb), "active_chat": activeChat, "active_embedding": activeEmb, "fallbacks": fallbacks, + "health": modelHealth, // 运行时每模型 failover/熔断态 }, "prompts": gin.H{"managed": len(prompts.Known), "overrides": len(overrides)}, "services": services,