From 1ddd2e547ac62ef1d86f4e8b603f7345d6aba3d9 Mon Sep 17 00:00:00 2001 From: viettranx Date: Tue, 10 Mar 2026 14:27:15 +0700 Subject: [PATCH] feat(compaction): configurable provider, model & max_tokens for pending message compaction Add Provider, Model, MaxTokens to PendingCompactionConfig so users can override the LLM used for pending message summarization via the config UI. Falls back to agent's provider/model when not set. Increase default max_tokens from 512 to 4096. Add allowEmpty prop to ProviderModelSelect to prevent auto-selecting first provider when empty means "use default". --- cmd/gateway.go | 11 +++- internal/channels/history_compaction.go | 10 +++- internal/channels/instance_loader.go | 57 ++++++++++++------- internal/config/config_channels.go | 7 ++- internal/http/pending_messages.go | 34 +++++++++-- .../shared/provider-model-select.tsx | 9 ++- ui/web/src/i18n/locales/en/config.json | 8 +++ ui/web/src/i18n/locales/vi/config.json | 8 +++ ui/web/src/i18n/locales/zh/config.json | 8 +++ .../behavior-pending-compaction-card.tsx | 50 +++++++++++++++- 10 files changed, 167 insertions(+), 35 deletions(-) diff --git a/cmd/gateway.go b/cmd/gateway.go index d094b591..eb29f780 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -14,6 +14,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/agent" "github.com/nextlevelbuilder/goclaw/internal/bootstrap" "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/cache" "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/channels/discord" "github.com/nextlevelbuilder/goclaw/internal/channels/feishu" @@ -618,6 +619,8 @@ func runGateway() { if pendingMessagesH != nil { if pc := cfg.Channels.PendingCompaction; pc != nil { pendingMessagesH.SetKeepRecent(pc.KeepRecent) + pendingMessagesH.SetMaxTokens(pc.MaxTokens) + pendingMessagesH.SetProviderModel(pc.Provider, pc.Model) } server.SetPendingMessagesHandler(pendingMessagesH) } @@ -852,7 +855,13 @@ func runGateway() { webFetchTool.UpdatePolicy(updatedCfg.Tools.WebFetch.Policy, updatedCfg.Tools.WebFetch.AllowedDomains, updatedCfg.Tools.WebFetch.BlockedDomains) }) - go consumeInboundMessages(ctx, msgBus, agentRouter, cfg, sched, channelMgr, consumerTeamStore, quotaChecker, delegateMgr, pgStores.Sessions, pgStores.Agents) + // Contact collector: auto-collect user info from channels with in-memory dedup cache. + var contactCollector *store.ContactCollector + if pgStores.Contacts != nil { + contactCollector = store.NewContactCollector(pgStores.Contacts, cache.NewInMemoryCache[bool]()) + } + + go consumeInboundMessages(ctx, msgBus, agentRouter, cfg, sched, channelMgr, consumerTeamStore, quotaChecker, delegateMgr, pgStores.Sessions, pgStores.Agents, contactCollector) go func() { sig := <-sigCh diff --git a/internal/channels/history_compaction.go b/internal/channels/history_compaction.go index 735632ca..a5812177 100644 --- a/internal/channels/history_compaction.go +++ b/internal/channels/history_compaction.go @@ -17,6 +17,7 @@ import ( type CompactionConfig struct { Threshold int // trigger compaction when entries exceed this (default 50) KeepRecent int // keep this many recent raw messages (default 15) + MaxTokens int // max output tokens for summarization (default 4096) Provider providers.Provider // LLM provider for summarization Model string // model to use for summarization } @@ -44,10 +45,13 @@ func (ph *PendingHistory) MaybeCompact(historyKey string, currentCount int, cfg // CompactGroup performs LLM-based compaction on a pending message group. // Reused by both auto-compact (channel) and HTTP compact endpoint. // Returns the number of entries remaining after compaction. -func CompactGroup(ctx context.Context, s store.PendingMessageStore, channelName, historyKey string, provider providers.Provider, model string, keepRecent int) (int, error) { +func CompactGroup(ctx context.Context, s store.PendingMessageStore, channelName, historyKey string, provider providers.Provider, model string, keepRecent, maxTokens int) (int, error) { if keepRecent <= 0 { keepRecent = 15 } + if maxTokens <= 0 { + maxTokens = 4096 + } // Step 1: Read entries from DB entries, err := s.ListByKey(ctx, channelName, historyKey) @@ -86,7 +90,7 @@ func CompactGroup(ctx context.Context, s store.PendingMessageStore, channelName, Content: "Summarize these group chat messages concisely, preserving key topics, decisions, names, and important context:\n\n" + sb.String(), }}, Model: model, - Options: map[string]any{"max_tokens": 512, "temperature": 0.3}, + Options: map[string]any{"max_tokens": maxTokens, "temperature": 0.3}, }) if err != nil { return 0, fmt.Errorf("llm summarize: %w", err) @@ -140,7 +144,7 @@ func (ph *PendingHistory) runCompaction(historyKey string, cfg *CompactionConfig return } - _, err = CompactGroup(ctx, ph.store, ph.channelName, historyKey, cfg.Provider, cfg.Model, cfg.KeepRecent) + _, err = CompactGroup(ctx, ph.store, ph.channelName, historyKey, cfg.Provider, cfg.Model, cfg.KeepRecent, cfg.MaxTokens) if err != nil { slog.Warn("compaction.failed", "channel", ph.channelName, "key", historyKey, "error", err) return diff --git a/internal/channels/instance_loader.go b/internal/channels/instance_loader.go index 7aded053..79f38109 100644 --- a/internal/channels/instance_loader.go +++ b/internal/channels/instance_loader.go @@ -235,28 +235,47 @@ func (l *InstanceLoader) loadInstance(ctx context.Context, inst store.ChannelIns base.SetType(inst.ChannelType) } - // Wire pending message auto-compaction using the agent's configured provider+model. - if pc, ok := ch.(PendingCompactable); ok && ag != nil && l.providerReg != nil && ag.Provider != "" { - if p, err := l.providerReg.Get(ag.Provider); err == nil { - model := ag.Model - if model == "" { - model = p.DefaultModel() - } - if model != "" { - cc := &CompactionConfig{ - Provider: p, - Model: model, + // Wire pending message auto-compaction. + // Priority: config provider/model > agent's provider/model > fallback. + if pc, ok := ch.(PendingCompactable); ok && l.providerReg != nil { + var p providers.Provider + var model string + + // Try config-level provider/model first. + if l.pendingCompactCfg != nil && l.pendingCompactCfg.Provider != "" { + if cp, err := l.providerReg.Get(l.pendingCompactCfg.Provider); err == nil { + p = cp + model = l.pendingCompactCfg.Model + if model == "" { + model = cp.DefaultModel() } - // Apply global threshold/keepRecent from config if set. - if l.pendingCompactCfg != nil { - cc.Threshold = l.pendingCompactCfg.Threshold - cc.KeepRecent = l.pendingCompactCfg.KeepRecent - } - pc.SetPendingCompaction(cc) - slog.Debug("pending compaction configured", "channel", inst.Name, "provider", ag.Provider, "model", model, - "threshold", cc.Threshold, "keep_recent", cc.KeepRecent) } } + // Fallback: agent's provider/model. + if p == nil && ag != nil && ag.Provider != "" { + if ap, err := l.providerReg.Get(ag.Provider); err == nil { + p = ap + model = ag.Model + if model == "" { + model = ap.DefaultModel() + } + } + } + + if p != nil && model != "" { + cc := &CompactionConfig{ + Provider: p, + Model: model, + } + if l.pendingCompactCfg != nil { + cc.Threshold = l.pendingCompactCfg.Threshold + cc.KeepRecent = l.pendingCompactCfg.KeepRecent + cc.MaxTokens = l.pendingCompactCfg.MaxTokens + } + pc.SetPendingCompaction(cc) + slog.Debug("pending compaction configured", "channel", inst.Name, "provider", p.Name(), "model", model, + "threshold", cc.Threshold, "keep_recent", cc.KeepRecent, "max_tokens", cc.MaxTokens) + } } l.manager.RegisterChannel(inst.Name, ch) l.loaded[inst.Name] = struct{}{} diff --git a/internal/config/config_channels.go b/internal/config/config_channels.go index e6634a4d..68c5e5e7 100644 --- a/internal/config/config_channels.go +++ b/internal/config/config_channels.go @@ -4,8 +4,11 @@ package config // When a group accumulates more than Threshold pending messages, older messages are // summarized by an LLM and replaced with a compact summary, keeping KeepRecent raw messages. type PendingCompactionConfig struct { - Threshold int `json:"threshold,omitempty"` // trigger compaction when entries exceed this (default 50) - KeepRecent int `json:"keep_recent,omitempty"` // keep this many recent raw messages after compaction (default 15) + Threshold int `json:"threshold,omitempty"` // trigger compaction when entries exceed this (default 50) + KeepRecent int `json:"keep_recent,omitempty"` // keep this many recent raw messages after compaction (default 15) + MaxTokens int `json:"max_tokens,omitempty"` // max output tokens for LLM summarization (default 4096) + Provider string `json:"provider,omitempty"` // LLM provider name (e.g. "openai"); empty = use agent's provider + Model string `json:"model,omitempty"` // model for summarization; empty = use agent's model } // ChannelsConfig contains per-channel configuration. diff --git a/internal/http/pending_messages.go b/internal/http/pending_messages.go index 27766ddb..3b15b76d 100644 --- a/internal/http/pending_messages.go +++ b/internal/http/pending_messages.go @@ -19,7 +19,10 @@ type PendingMessagesHandler struct { agentStore store.AgentStore token string providerReg *providers.Registry - keepRecent int // global keepRecent from config (0 = use default 15) + keepRecent int // global keepRecent from config (0 = use default 15) + maxTokens int // max output tokens for LLM summarization (0 = use default) + cfgProvider string // config-level provider override (empty = resolve from agent) + cfgModel string // config-level model override (empty = resolve from agent) } func NewPendingMessagesHandler(s store.PendingMessageStore, agentStore store.AgentStore, token string, providerReg *providers.Registry) *PendingMessagesHandler { @@ -29,6 +32,15 @@ func NewPendingMessagesHandler(s store.PendingMessageStore, agentStore store.Age // SetKeepRecent sets the global keepRecent value from config. func (h *PendingMessagesHandler) SetKeepRecent(n int) { h.keepRecent = n } +// SetMaxTokens sets the global maxTokens value from config. +func (h *PendingMessagesHandler) SetMaxTokens(n int) { h.maxTokens = n } + +// SetProviderModel sets explicit provider/model from config. +func (h *PendingMessagesHandler) SetProviderModel(provider, model string) { + h.cfgProvider = provider + h.cfgModel = model +} + func (h *PendingMessagesHandler) RegisterRoutes(mux *http.ServeMux) { mux.HandleFunc("GET /v1/pending-messages", h.authMiddleware(h.handleListGroups)) mux.HandleFunc("GET /v1/pending-messages/messages", h.authMiddleware(h.handleListMessages)) @@ -146,7 +158,7 @@ func (h *PendingMessagesHandler) handleCompact(w http.ResponseWriter, r *http.Re if keepRecent <= 0 { keepRecent = 15 } - remaining, err := channels.CompactGroup(ctx, h.store, req.ChannelName, req.HistoryKey, provider, model, keepRecent) + remaining, err := channels.CompactGroup(ctx, h.store, req.ChannelName, req.HistoryKey, provider, model, keepRecent, h.maxTokens) if err != nil { slog.Warn("compact.failed", "channel", req.ChannelName, "key", req.HistoryKey, "error", err) writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) @@ -156,14 +168,26 @@ func (h *PendingMessagesHandler) handleCompact(w http.ResponseWriter, r *http.Re } // resolveProviderAndModel resolves the LLM provider+model for pending message compaction. -// Uses the default agent's configured provider and model so compaction uses the same -// LLM as the agent that processes these messages. +// Priority: config provider/model > default agent's provider/model > first available provider. func (h *PendingMessagesHandler) resolveProviderAndModel() (providers.Provider, string) { if h.providerReg == nil { return nil, "" } - // Use the default agent's provider+model + // Config-level provider/model override. + if h.cfgProvider != "" { + if p, err := h.providerReg.Get(h.cfgProvider); err == nil { + model := h.cfgModel + if model == "" { + model = p.DefaultModel() + } + if model != "" { + return p, model + } + } + } + + // Fallback: default agent's provider+model. if h.agentStore != nil { if ag, err := h.agentStore.GetDefault(context.Background()); err == nil && ag.Provider != "" { if p, err := h.providerReg.Get(ag.Provider); err == nil { diff --git a/ui/web/src/components/shared/provider-model-select.tsx b/ui/web/src/components/shared/provider-model-select.tsx index 2dd67b52..bc7551df 100644 --- a/ui/web/src/components/shared/provider-model-select.tsx +++ b/ui/web/src/components/shared/provider-model-select.tsx @@ -34,6 +34,8 @@ interface ProviderModelSelectProps { savedModel?: string; /** Called when verification status changes. True = save should be blocked (changed but not verified). */ onSaveBlockedChange?: (blocked: boolean) => void; + /** When true, skip auto-selecting the first provider when none is set. Useful when empty means "use default". */ + allowEmpty?: boolean; } export function ProviderModelSelect({ @@ -51,17 +53,18 @@ export function ProviderModelSelect({ savedProvider, savedModel, onSaveBlockedChange, + allowEmpty, }: ProviderModelSelectProps) { const { t } = useTranslation("common"); const { providers } = useProviders(); const enabledProviders = providers.filter((p) => p.enabled); - // Auto-select first enabled provider when none is set + // Auto-select first enabled provider when none is set (unless allowEmpty) useEffect(() => { - if (!provider && enabledProviders.length > 0) { + if (!allowEmpty && !provider && enabledProviders.length > 0) { onProviderChange(enabledProviders[0]!.name); } - }, [provider, enabledProviders, onProviderChange]); + }, [allowEmpty, provider, enabledProviders, onProviderChange]); const selectedProvider = useMemo( () => enabledProviders.find((p) => p.name === provider), diff --git a/ui/web/src/i18n/locales/en/config.json b/ui/web/src/i18n/locales/en/config.json index 08a9c8e6..7c58cf9f 100644 --- a/ui/web/src/i18n/locales/en/config.json +++ b/ui/web/src/i18n/locales/en/config.json @@ -130,6 +130,14 @@ "behavior.pendingCompactionThresholdTip": "Trigger compaction when a group accumulates more than this many pending messages. Set to 0 to use the default (50).", "behavior.pendingCompactionKeepRecent": "Keep Recent", "behavior.pendingCompactionKeepRecentTip": "Number of recent raw messages to keep after compaction. Older messages are replaced with an LLM summary. Default is 15.", + "behavior.pendingCompactionMaxTokens": "Max Tokens", + "behavior.pendingCompactionMaxTokensTip": "Maximum output tokens for LLM summarization. Default is 4096.", + "behavior.pendingCompactionProvider": "Provider", + "behavior.pendingCompactionProviderTip": "LLM provider for summarization. Leave empty to use the channel agent's provider.", + "behavior.pendingCompactionProviderPlaceholder": "Use agent's provider", + "behavior.pendingCompactionModel": "Model", + "behavior.pendingCompactionModelTip": "Model for summarization. Leave empty to use the channel agent's model.", + "behavior.pendingCompactionModelPlaceholder": "Use agent's model", "behavior.pendingCompactionInfo": "When a group chat exceeds the threshold, the LLM automatically summarizes older messages into a single compact entry, keeping the most recent ones intact for context.", "behavior.rateLimitTitle": "Rate Limiting", "behavior.rateLimitDescription": "Inbound message size limits and per-user request throttling", diff --git a/ui/web/src/i18n/locales/vi/config.json b/ui/web/src/i18n/locales/vi/config.json index 692a868e..6e044dde 100644 --- a/ui/web/src/i18n/locales/vi/config.json +++ b/ui/web/src/i18n/locales/vi/config.json @@ -130,6 +130,14 @@ "behavior.pendingCompactionThresholdTip": "Kích hoạt nén khi nhóm tích lũy nhiều hơn số tin nhắn chờ này. Đặt 0 để dùng mặc định (50).", "behavior.pendingCompactionKeepRecent": "Giữ gần đây", "behavior.pendingCompactionKeepRecentTip": "Số tin nhắn gần đây giữ lại sau khi nén. Tin cũ hơn được thay bằng bản tóm tắt LLM. Mặc định là 15.", + "behavior.pendingCompactionMaxTokens": "Token tối đa", + "behavior.pendingCompactionMaxTokensTip": "Số token tối đa cho kết quả tóm tắt LLM. Mặc định 4096.", + "behavior.pendingCompactionProvider": "Nhà cung cấp", + "behavior.pendingCompactionProviderTip": "Nhà cung cấp LLM cho tóm tắt. Để trống sẽ dùng provider của agent kênh.", + "behavior.pendingCompactionProviderPlaceholder": "Dùng provider của agent", + "behavior.pendingCompactionModel": "Mô hình", + "behavior.pendingCompactionModelTip": "Mô hình LLM cho tóm tắt. Để trống sẽ dùng model của agent kênh.", + "behavior.pendingCompactionModelPlaceholder": "Dùng model của agent", "behavior.pendingCompactionInfo": "Khi nhóm chat vượt ngưỡng, LLM tự động tóm tắt các tin nhắn cũ thành một mục rút gọn, giữ nguyên các tin gần đây nhất làm ngữ cảnh.", "behavior.rateLimitTitle": "Giới hạn tốc độ", "behavior.rateLimitDescription": "Giới hạn kích thước tin nhắn đến và kiểm soát tốc độ yêu cầu mỗi người dùng", diff --git a/ui/web/src/i18n/locales/zh/config.json b/ui/web/src/i18n/locales/zh/config.json index e0dfe327..dd7efa3a 100644 --- a/ui/web/src/i18n/locales/zh/config.json +++ b/ui/web/src/i18n/locales/zh/config.json @@ -130,6 +130,14 @@ "behavior.pendingCompactionThresholdTip": "当群组累积超过此数量的待处理消息时触发压缩。设为 0 使用默认值（50）。", "behavior.pendingCompactionKeepRecent": "保留最近", "behavior.pendingCompactionKeepRecentTip": "压缩后保留的最近原始消息数量。较旧的消息将替换为 LLM 摘要。默认为 15。", + "behavior.pendingCompactionMaxTokens": "最大 Token 数", + "behavior.pendingCompactionMaxTokensTip": "LLM 摘要的最大输出 token 数。默认 4096。", + "behavior.pendingCompactionProvider": "提供商", + "behavior.pendingCompactionProviderTip": "用于摘要的 LLM 提供商。留空将使用频道 Agent 的提供商。", + "behavior.pendingCompactionProviderPlaceholder": "使用 Agent 的提供商", + "behavior.pendingCompactionModel": "模型", + "behavior.pendingCompactionModelTip": "用于摘要的模型。留空将使用频道 Agent 的模型。", + "behavior.pendingCompactionModelPlaceholder": "使用 Agent 的模型", "behavior.pendingCompactionInfo": "当群聊超过阈值时，LLM 自动将旧消息总结为一条精简条目，保留最近的消息作为上下文。", "behavior.rateLimitTitle": "速率限制", "behavior.rateLimitDescription": "入站消息大小限制和每用户请求限流", diff --git a/ui/web/src/pages/config/sections/behavior-pending-compaction-card.tsx b/ui/web/src/pages/config/sections/behavior-pending-compaction-card.tsx index c1e013fd..22aad0b6 100644 --- a/ui/web/src/pages/config/sections/behavior-pending-compaction-card.tsx +++ b/ui/web/src/pages/config/sections/behavior-pending-compaction-card.tsx @@ -1,4 +1,4 @@ -import { Archive, Clock, Info } from "lucide-react"; +import { Archive, Clock, Hash, Info } from "lucide-react"; import { useTranslation } from "react-i18next"; import { Input } from "@/components/ui/input"; import { Label } from "@/components/ui/label"; @@ -9,10 +9,14 @@ import { CardHeader, CardTitle, } from "@/components/ui/card"; +import { ProviderModelSelect } from "@/components/shared/provider-model-select"; export interface PendingCompactionValues { threshold?: number; keep_recent?: number; + max_tokens?: number; + provider?: string; + model?: string; } interface Props { @@ -38,6 +42,23 @@ export function BehaviorPendingCompactionCard({ value, onChange }: Props) { + {/* Provider & Model */} +
+ update({ provider: v })} + model={value.model ?? ""} + onModelChange={(v) => update({ model: v })} + providerLabel={t("behavior.pendingCompactionProvider")} + modelLabel={t("behavior.pendingCompactionModel")} + providerTip={t("behavior.pendingCompactionProviderTip")} + modelTip={t("behavior.pendingCompactionModelTip")} + providerPlaceholder={t("behavior.pendingCompactionProviderPlaceholder")} + modelPlaceholder={t("behavior.pendingCompactionModelPlaceholder")} + allowEmpty + /> +
+ {/* Threshold */}
@@ -64,7 +85,7 @@ export function BehaviorPendingCompactionCard({ value, onChange }: Props) {
{/* Keep Recent */} -
+
@@ -88,6 +109,31 @@ export function BehaviorPendingCompactionCard({ value, onChange }: Props) {
+ {/* Max Tokens */} +
+
+
+ +
+ +

+ {t("behavior.pendingCompactionMaxTokensTip")} +

+
+
+ update({ max_tokens: Number(e.target.value) })} + placeholder="4096" + min={256} + className="w-24 shrink-0" + /> +
+
+ {/* Info banner */}