Blizzard
faa1871760
token 用量估算计量（CJK≈1/字、ASCII≈1/4字，无需分词器，护栏够用）。 单任务硬上限（dispatcher）：Budget 挂 ctx 沿图透传，各 LLM 节点（对话/ReAct/compose/ 报告）入口计输入 token、出口计输出，触顶即中止整图——防失控成本（死循环/超大报告）。 报告路径优雅降级：触顶跳过剩余章节出部分稿，不整体失败。预算来源 Meta.token_budget 或 env TASK_TOKEN_BUDGET（默认 20 万）。 单用户日预算（gateway）：dispatcher 收尾经 NATS 回写 UsageEvent → 网关按用户按天累计 Redis（48h 过期自滚动）→ 提交前门控 USER_DAILY_TOKEN_BUDGET（0=不限，超额 402）。 /billing 升级为真实用量：当日已用 / 日预算 / 余额。 契约 UsageEvent + MetaTokenBudget + SubjectUsage；bus Publish/SubscribeUsage； orchestrator SetUsageSink + 预算触顶 failed（不计熔断）。harness budget 5 单测，三模块全绿。 live：单任务 budget=30 → failed(已用约689)；用户日 budget=200 → /billing remaining=0 → 402。 至此 harness 由「测温计」完成向「恒温器」的演进（评测闭环/纠偏/忠实度/脱敏/输入护栏/预算六项）。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
93 lines
3.6 KiB
Go
93 lines
3.6 KiB
Go
// Command dispatcher 启动 sundynix-dispatcher —— 第 4 层 AI Agent 调度集群。
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"log"
|
|
"os"
|
|
"os/signal"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/sundynix/sundynix-dispatcher/internal/eino"
|
|
"github.com/sundynix/sundynix-dispatcher/internal/harness"
|
|
"github.com/sundynix/sundynix-dispatcher/internal/llm"
|
|
dnats "github.com/sundynix/sundynix-dispatcher/internal/nats"
|
|
"github.com/sundynix/sundynix-shared/otelx"
|
|
"github.com/sundynix/sundynix-shared/secrets"
|
|
)
|
|
|
|
func main() {
|
|
secrets.MustHaveKeyInProd() // 生产须配 SUNDYNIX_SECRET_KEY 以解密下发的 api_key 密文
|
|
otelx.SetupSlog("sundynix-dispatcher") // 结构化日志 + 链路感知（任务日志带 trace_id）
|
|
|
|
// 链路追踪：消费 span 续上 gateway 的 trace，再向下展开节点/工具/LLM span。
|
|
shutdownTrace, _ := otelx.Init(context.Background(), "sundynix-dispatcher")
|
|
defer func() { _ = shutdownTrace(context.Background()) }()
|
|
|
|
natsURL := envOr("NATS_URL", "nats://localhost:4222")
|
|
|
|
pool := llm.NewPool() // LLM Pool: vLLM / Ollama 集群
|
|
breaker := harness.NewCircuitBreaker() // Harness: 熔断降级中心
|
|
// Harness: LLM 自动化评测（规则 + LLM-as-judge，模型就绪时启用）。
|
|
llmChat := func(ctx context.Context, sys, user string) (string, error) {
|
|
return pool.Chat(ctx, []llm.ChatMessage{{Role: "system", Content: sys}, {Role: "user", Content: user}})
|
|
}
|
|
eval := harness.NewEvaluator(pool.Ready, llmChat)
|
|
// Harness: 输入护栏 Tier2 —— 对网关标记的灰区输入做 LLM 越狱裁决（模型就绪时启用）。
|
|
guardian := harness.NewClassifier(pool.Ready, llmChat)
|
|
|
|
sub := dnats.MustConnect(natsURL)
|
|
defer sub.Close()
|
|
|
|
// 配置控制面：先订阅热更新，再后台重试拉初始配置（容忍 gateway 晚于本服务启动，
|
|
// 避免一次性请求扑空后只能干等热更新 → 降级桩跑全程）。
|
|
if _, err := sub.SubscribeModelConfigUpdated(pool.SetConfig); err != nil {
|
|
log.Printf("[dispatcher] subscribe model config: %v", err)
|
|
}
|
|
go sub.FetchModelConfigWithRetry(context.Background(), pool.SetConfig)
|
|
|
|
// sub 同时作为 Token 回流(TokenSink)、MCP 工具调用(ToolCaller)、执行事件(ExecSink)、
|
|
// 任务状态回写(StatusSink)、HITL 审批等待(ApprovalWaiter)与评测落库(EvalSink)出口。
|
|
orch, err := eino.NewOrchestrator(pool, breaker, eval, sub, sub, sub, sub, sub, sub)
|
|
if err != nil {
|
|
log.Fatalf("[dispatcher] build eino graph: %v", err)
|
|
}
|
|
orch.SetGuardian(guardian) // 输入护栏 Tier2
|
|
orch.SetUsageSink(sub) // 成本护栏：token 用量回写网关累计/计费
|
|
|
|
// 健康心跳：dispatcher 无 HTTP/工具端点，挂一个 NATS 应答让管理端「服务状态」探到它在线。
|
|
startedAt := time.Now()
|
|
if unsub, herr := sub.ServeHealth(func() []byte {
|
|
data, _ := json.Marshal(map[string]any{
|
|
"ok": true,
|
|
"service": "dispatcher",
|
|
"model": pool.ModelName(),
|
|
"ready": pool.Ready(),
|
|
"uptime_s": int(time.Since(startedAt).Seconds()),
|
|
})
|
|
return data
|
|
}); herr != nil {
|
|
log.Printf("[dispatcher] serve health: %v", herr)
|
|
} else {
|
|
defer func() { _ = unsub() }()
|
|
}
|
|
|
|
// 监听退出信号，优雅停止消费。
|
|
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
|
defer stop()
|
|
|
|
log.Println("[dispatcher] consuming sundynix.tasks.* (Ctrl-C to quit)")
|
|
if err := sub.ConsumeTasks(ctx, orch.Handle); err != nil && err != context.Canceled {
|
|
log.Fatalf("[dispatcher] exit: %v", err)
|
|
}
|
|
}
|
|
|
|
func envOr(key, def string) string {
|
|
if v := os.Getenv(key); v != "" {
|
|
return v
|
|
}
|
|
return def
|
|
}
|