From 648292d035177e3840a55672601a3645c340dd86 Mon Sep 17 00:00:00 2001 From: Aska Date: Sun, 6 Sep 2026 18:08:55 +0800 Subject: [PATCH] =?UTF-8?q?feat(loop):=20Top5#5=20=E4=B8=8A=E4=B8=8B?= =?UTF-8?q?=E6=96=87=E6=BA=A2=E5=87=BA=E5=8E=8B=E7=BC=A9=20+=20=E6=AF=8F?= =?UTF-8?q?=E8=BD=AE=20assemble=EF=BC=88=E5=AF=B9=E9=BD=90=20OpenClaw=20co?= =?UTF-8?q?ntext-engine=EF=BC=89=E2=80=94=E2=80=94Top5=20=E5=85=A8?= =?UTF-8?q?=E5=AE=8C=E6=88=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 每轮 assemble(transformContext 等价)：run.go 每 step 前 assembleContext——依次 Inject(每轮 重新 Retriever.Inject 不缓存首轮)、WorkingMemory 前置 Summarize、可插拔 TransformContext 投影/ 过滤；生成的 view 交 step，msgs 保持纯对话供工具结果追加，记忆不重复注入、不写回 checkpoint。 - Compactor：结构分支感知压缩——estimateContextTokens 启发式超 MaxContextTokens 且消息≥3 才动； findKeepStart 按 user 消息切整轮从后往前累加(至少保留最后一整轮→尾随工具结果绝不切片/不跨轮)； 旧轮 user/assistant 对压成一条 RoleSystem 摘要置顶，既有摘要前缀识别合并防叠；capSummary 上限 4000 rune；ctx 即 abortSignal(取消返回 ctx.Err，压缩失败仅记日志不中断 run)。 - 选项：Options 增 MaxContextTokens/Compactor/TransformContext，均 nil-tolerant(默认=现有行为， 零值不压缩)；DefaultMaxContextTokens=32768。Compactor 接口预留注入 LLM 摘要。 go build/vet/test -count=1 全绿 + windows/darwin 交叉编译。12 断言覆盖尾随工具结果保留/不跨轮切割/ 摘要上限/摘要合并/ctx 取消/每轮重检索/TransformContext 逐轮/超限压缩/默认关闭。-race 受宿主 TSan VMA 限制无法运行(与代码无关)。 **至此 Top5(1~5) 全部完成并推库。** --- internal/runtime/loop/assemble_test.go | 145 ++++++++++++ internal/runtime/loop/compactor.go | 283 ++++++++++++++++++++++++ internal/runtime/loop/compactor_test.go | 165 ++++++++++++++ internal/runtime/loop/loop.go | 49 ++-- internal/runtime/loop/memory.go | 55 ++++- internal/runtime/loop/options.go | 23 +- internal/runtime/loop/run.go | 20 +- 7 files changed, 709 insertions(+), 31 deletions(-) create mode 100644 internal/runtime/loop/assemble_test.go create mode 100644 internal/runtime/loop/compactor.go create mode 100644 internal/runtime/loop/compactor_test.go diff --git a/internal/runtime/loop/assemble_test.go b/internal/runtime/loop/assemble_test.go new file mode 100644 index 0000000..b5b2aa7 --- /dev/null +++ b/internal/runtime/loop/assemble_test.go @@ -0,0 +1,145 @@ +package loop + +import ( + "context" + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// TestAssembleContext_ReRetrievesEachRound proves memory is re-retrieved every +// round rather than cached from the first turn: an empty store yields no memory +// block, then after seeding the store a later round picks it up. +func TestAssembleContext_ReRetrievesEachRound(t *testing.T) { + t.Parallel() + store := newTestMemoryStore(t) + ret := memory.NewRetriever(store) + l := New(Options{Retriever: ret, PersonaID: "persona-a"}) + + msgs := []model.Message{model.TextMessage(model.RoleUser, "do the thing")} + + // Round 1: the store is empty, so no memory is injected. + if view := l.assembleContext(context.Background(), msgs, 1); len(view) != 1 { + t.Fatalf("round 1 view = %d messages, want 1 (not memory)", len(view)) + } + + // Seed a memory as if it was learned between rounds. + if _, err := store.Add(context.Background(), memory.MemoryItem{ + PersonaID: "persona-a", Content: "她喜欢冰美式", + Scope: memory.ScopePersona, Kind: memory.KindRelationship, + }); err != nil { + t.Fatalf("seed memory: %v", err) + } + + // Round 2: the retrieval is re-run and the memory now appears. + view := l.assembleContext(context.Background(), msgs, 2) + if len(view) != 2 { + t.Fatalf("round 2 view = %d messages, want 2 (memory injected)", len(view)) + } + if view[0].Role != model.RoleSystem || !strings.Contains(view[0].Content[0].Text, "冰美式") { + t.Fatalf("round 2 memory not injected: %+v", view[0]) + } +} + +// TestCompactIfNeeded_OverBudget_Compacts drives a multi-turn conversation over a +// small budget and verifies the loop compacts it into a leading summary. +func TestCompactIfNeeded_OverBudget_Compacts(t *testing.T) { + t.Parallel() + l := New(Options{MaxContextTokens: 1000}) + big := buildTurns(6, 600, false) + if estimateContextTokens(big) <= 1000 { + t.Fatalf("test precondition: want > 1000 tokens, got %d", estimateContextTokens(big)) + } + got := l.compactIfNeeded(context.Background(), big) + if len(got) >= len(big) { + t.Fatalf("compactIfNeeded did not shrink: got %d >= %d", len(got), len(big)) + } + if got[0].Role != model.RoleSystem || !strings.HasPrefix(messageText(got[0]), summaryMarker) { + t.Fatalf("expected a leading compaction summary, got %+v", got[0]) + } +} + +// TestCompactIfNeeded_Disabled_NoOp keeps the pre-existing behavior when +// MaxContextTokens is left at its zero value. +func TestCompactIfNeeded_Disabled_NoOp(t *testing.T) { + t.Parallel() + l := New(Options{}) // MaxContextTokens == 0 → disabled + big := buildTurns(6, 300, false) + got := l.compactIfNeeded(context.Background(), big) + if len(got) != len(big) { + t.Fatalf("disabled compaction should be a no-op: got %d, want %d", len(got), len(big)) + } +} + +// TestRun_TransformContext_InvokedEachRound verifies the TransformContext hook is +// called once per model step (per round), bridging per-round assemble into the +// Run path. +func TestRun_TransformContext_InvokedEachRound(t *testing.T) { + t.Parallel() + call := `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}` + ad := &mockAdapter{chat: []mgResp{ + {resp: model.Response{Content: call}}, + {resp: model.Response{Content: "done"}}, + }} + td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"}) + a := newTestAgent(t, ad, td) + + var rounds []int + var sizes []int + l := New(Options{ + Agent: a, + TransformContext: func(_ context.Context, msgs []model.Message, round int) []model.Message { + rounds = append(rounds, round) + sizes = append(sizes, len(msgs)) + return msgs + }, + MaxRounds: 8, + }) + + if _, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "go")}); err != nil { + t.Fatalf("Run: %v", err) + } + if len(rounds) != 2 { + t.Fatalf("TransformContext called %d times, want 2 (one per round)", len(rounds)) + } + if rounds[0] != 1 || rounds[1] != 2 { + t.Fatalf("rounds passed to TransformContext = %v, want [1 2]", rounds) + } + // Round 1 has just the user message; round 2 has the fed-back tool result. + if sizes[0] != 1 || sizes[1] != 2 { + t.Fatalf("sizes = %v, want [1 2]", sizes) + } +} + +// TestRun_ContextOverflow_CompactsHistory passes a long multi-turn history and a +// small budget into Run; the first model step should see a compacted conversation +// carrying a leading summary. +func TestRun_ContextOverflow_CompactsHistory(t *testing.T) { + t.Parallel() + ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "final answer"}}}} + a := newTestAgent(t, ad, nil) + l := New(Options{Agent: a, MaxContextTokens: 600, MaxRounds: 8}) + + history := buildTurns(8, 200, false) + if estimateContextTokens(history) <= 600 { + t.Fatalf("test precondition: want > 600 tokens, got %d", estimateContextTokens(history)) + } + if _, err := l.Run(context.Background(), "sess", history); err != nil { + t.Fatalf("Run: %v", err) + } + if len(ad.reqs) == 0 { + t.Fatal("no request captured") + } + found := false + for _, m := range ad.reqs[0].Messages { + if m.Role == model.RoleSystem && strings.HasPrefix(messageText(m), summaryMarker) { + found = true + break + } + } + if !found { + t.Fatalf("expected a compaction summary in the first request, got %+v", ad.reqs[0].Messages) + } +} diff --git a/internal/runtime/loop/compactor.go b/internal/runtime/loop/compactor.go new file mode 100644 index 0000000..d1553a9 --- /dev/null +++ b/internal/runtime/loop/compactor.go @@ -0,0 +1,283 @@ +package loop + +import ( + "context" + "fmt" + "strings" + "unicode/utf8" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// Compaction tuning constants. These bound the compactor so it stays +// deterministic and offline: the summary is capped, the recent tail is a share +// of the budget, and very short conversations are left alone even when they +// exceed the budget (design §1.3 上下文溢出压缩). +const ( + // DefaultMaxContextTokens is the context budget a caller can assign to + // Options.MaxContextTokens to enable overflow compaction (design §1.3). + DefaultMaxContextTokens = 32768 + + // keepRecentFraction is the share of the budget reserved for the recent tail + // during compaction; the remainder is left for the summary message and prompt. + keepRecentFraction = 0.5 + + // minKeepTokens is the floor for the recent-tail budget so a tiny + // MaxContextTokens does not force every turn into the summary. + minKeepTokens = 128 + + // summaryMarker tags a compaction summary system message so the compactor can + // recognize an existing summary prefix and preserve (merge into) it rather + // than re-summarize it on a later compaction. + summaryMarker = "[前文摘要 · history summary]" + + // maxSummaryChars caps the compaction summary so repeated compactions cannot + // inflate the summary without bound (design §1.3 摘要容量上限). + maxSummaryChars = 4000 + + // minCompactMessages is the smallest conversation the compactor will touch; a + // shorter one is left alone even when it is over budget. + minCompactMessages = 3 + + // messageOverheadTokens is the per-message token overhead charged for the role + // and separator tokens the provider adds around each message. + messageOverheadTokens = 4 + + // imageTokens is the rough token charge for a single image part. + imageTokens = 1000 + + // charsPerToken is the character-to-token divisor used by the offline estimate. + charsPerToken = 4 + + // summarySampleChars bounds the per-message text sampled into a naive summary. + summarySampleChars = 120 +) + +// Compactor compacts an over-budget conversation by summarizing older turns into +// a single leading summary message while preserving the recent tail — including +// the trailing tool results — unmodified (design §1.3, aligned with OpenClaw's +// context-engine). It is injected through Options.Compactor so a caller can swap +// in an LLM-backed summarizer; the loop ships a structural default that never +// calls a model. +type Compactor interface { + // Compact returns messages with older turns compressed into a leading summary + // when the estimated tokens exceed budget. It returns the input unchanged when + // the messages already fit within budget or there is nothing safe to + // summarize. ctx is the abort signal: a canceled context stops compaction and + // returns a wrapped context.Canceled so the caller can bail out. + Compact(ctx context.Context, messages []model.Message, budget int) ([]model.Message, error) +} + +// defaultCompactor is the built-in structural compactor. It is stateless and +// safe for concurrent use. +type defaultCompactor struct{} + +// Compact implements Compactor. +func (defaultCompactor) Compact(ctx context.Context, messages []model.Message, budget int) ([]model.Message, error) { + if ctx == nil { + ctx = context.Background() + } + if err := ctx.Err(); err != nil { + return nil, fmt.Errorf("loop: compact: %w", err) + } + return compactMessages(ctx, messages, budget) +} + +// compactMessages is the shared compaction implementation. It never mutates the +// input slices; a compacted result is always freshly allocated. +func compactMessages(ctx context.Context, messages []model.Message, budget int) ([]model.Message, error) { + if budget <= 0 || len(messages) < minCompactMessages { + return messages, nil + } + if estimateContextTokens(messages) <= budget { + return messages, nil + } + + prefixEnd := summaryPrefixEnd(messages) + keepTokens := max(int(float64(budget)*keepRecentFraction), minKeepTokens) + keepStart := findKeepStart(messages, prefixEnd, keepTokens) + if keepStart <= prefixEnd { + // Everything is being kept; there is no safe prefix to summarize. + return messages, nil + } + + if err := ctx.Err(); err != nil { + return nil, fmt.Errorf("loop: compact: %w", err) + } + + existing := "" + if prefixEnd > 0 { + existing = strings.TrimPrefix(messageText(messages[0]), summaryMarker+"\n\n") + } + fresh := buildSummary(messages[prefixEnd:keepStart]) + if fresh == "" && existing == "" { + return messages, nil + } + full := fresh + if existing != "" { + full = existing + "\n\n" + fresh + } + + summary := model.TextMessage(model.RoleSystem, summaryMarker+"\n\n"+capSummary(full, maxSummaryChars)) + out := make([]model.Message, 0, len(messages)-(keepStart-prefixEnd)+1) + out = append(out, summary) + out = append(out, messages[keepStart:]...) + return out, nil +} + +// summaryPrefixEnd returns the count of leading compaction-summary messages, +// which are never re-summarized. It is 0 (no summary) or 1 (a single summary +// prefix), since compaction always emits exactly one leading summary. +func summaryPrefixEnd(messages []model.Message) int { + if len(messages) == 0 || messages[0].Role != model.RoleSystem { + return 0 + } + if strings.HasPrefix(messageText(messages[0]), summaryMarker) { + return 1 + } + return 0 +} + +// findKeepStart returns the index of the first message to keep given that the +// recent tail should fit within keepTokens. It always keeps whole turns: a turn +// starts at a user message and runs to the next user message. It scores turns +// from the back, so the most recent turns survive and a trailing tool result is +// never sliced apart. When no user turn exists (unusual), it keeps nothing extra +// and returns from, so nothing is summarized. +func findKeepStart(messages []model.Message, from, keepTokens int) int { + n := len(messages) + turns := make([]int, 0, 8) + for i := from; i < n; i++ { + if messages[i].Role == model.RoleUser { + turns = append(turns, i) + } + } + if len(turns) == 0 { + return from + } + + acc := 0 + keptStart := turns[len(turns)-1] + for i := len(turns) - 1; i >= 0; i-- { + to := n + if i+1 < len(turns) { + to = turns[i+1] + } + if i == len(turns)-1 || acc+estimateContextTokens(messages[turns[i]:to]) <= keepTokens { + acc += estimateContextTokens(messages[turns[i]:to]) + keptStart = turns[i] + } else { + break + } + } + return keptStart +} + +// estimateContextTokens returns a conservative token estimate for messages. It is +// a heuristic (chars ÷ charsPerToken with a per-message and per-image overhead) +// so it stays dependency-free and offline, but it is monotonic enough to drive +// the compact decision. +func estimateContextTokens(messages []model.Message) int { + total := 0 + for _, m := range messages { + total += messageOverheadTokens + for _, part := range m.Content { + switch part.Type { + case model.ContentTypeText: + total += estimateStringTokens(part.Text) + case model.ContentTypeImage: + total += imageTokens + } + } + } + return total +} + +// estimateStringTokens is the character-level token estimate for one text string, +// weighting wide (CJK) runes double and dividing by charsPerToken. An empty string +// costs nothing; a non-empty one costs at least one token. +func estimateStringTokens(s string) int { + if s == "" { + return 0 + } + runes := utf8.RuneCountInString(s) + wide := 0 + for _, r := range s { + if r >= 0x2E80 { + wide++ + } + } + t := (runes + wide) / charsPerToken + if t < 1 { + t = 1 + } + return t +} + +// buildSummary renders a compact structural summary of the given messages: a role +// distribution plus the most recent user and assistant text. It is the default +// (offline, no-model) summarizer; a caller that wants a semantic digest injects +// its own Compactor instead. +func buildSummary(messages []model.Message) string { + if len(messages) == 0 { + return "" + } + counts := map[model.Role]int{} + lastUser, lastAssistant := "", "" + for _, m := range messages { + counts[m.Role]++ + text := messageText(m) + if text == "" { + continue + } + switch m.Role { + case model.RoleUser: + lastUser = text + case model.RoleAssistant: + lastAssistant = text + } + } + + var b strings.Builder + fmt.Fprintf(&b, "已压缩 %d 条历史消息（", len(messages)) + first := true + for _, role := range []model.Role{model.RoleSystem, model.RoleUser, model.RoleAssistant, model.RoleTool} { + if c := counts[role]; c > 0 { + if !first { + b.WriteString("，") + } + first = false + fmt.Fprintf(&b, "%s %d", role, c) + } + } + b.WriteString("）") + if lastUser != "" { + fmt.Fprintf(&b, "。最近用户: %s", truncateRunes(lastUser, summarySampleChars)) + } + if lastAssistant != "" { + fmt.Fprintf(&b, "。最近助手: %s", truncateRunes(lastAssistant, summarySampleChars)) + } + return b.String() +} + +// capSummary truncates s to at most max runes, appending a "..." marker when it +// had to cut. It is rune-safe so it never slices a multi-byte rune in half. +func capSummary(s string, max int) string { + runes := []rune(s) + if len(runes) <= max { + return s + } + if max < 3 { + return string(runes[:max]) + } + return string(runes[:max-3]) + "..." +} + +// truncateRunes shortens s to at most n runes, Unicode-safe. +func truncateRunes(s string, n int) string { + runes := []rune(s) + if len(runes) <= n { + return s + } + return string(runes[:n]) +} diff --git a/internal/runtime/loop/compactor_test.go b/internal/runtime/loop/compactor_test.go new file mode 100644 index 0000000..4874a9c --- /dev/null +++ b/internal/runtime/loop/compactor_test.go @@ -0,0 +1,165 @@ +package loop + +import ( + "context" + "errors" + "reflect" + "strconv" + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// buildTurns builds a multi-turn conversation: n user/assistant turns followed by +// a trailing tool result, each message padded to around pad chars so the token +// estimate can be pushed over a small budget. +func buildTurns(n, pad int, withTool bool) []model.Message { + msgs := make([]model.Message, 0, n*2+1) + for i := 0; i < n; i++ { + user := model.TextMessage(model.RoleUser, strings.Repeat("u", pad)+" turn-"+strconv.Itoa(i)) + asst := model.TextMessage(model.RoleAssistant, strings.Repeat("a", pad)+" reply-"+strconv.Itoa(i)) + msgs = append(msgs, user, asst) + if withTool && i == n-1 { + msgs = append(msgs, model.TextMessage(model.RoleTool, strings.Repeat("r", pad*4)+" tool-result")) + } + } + return msgs +} + +func TestCompact_UnderBudget_Unchanged(t *testing.T) { + t.Parallel() + msgs := buildTurns(2, 4, false) + got, err := (defaultCompactor{}).Compact(context.Background(), msgs, 1_000_000) + if err != nil { + t.Fatalf("Compact: %v", err) + } + if len(got) != len(msgs) { + t.Fatalf("under budget should be unchanged: got %d messages, want %d", len(got), len(msgs)) + } + for i := range msgs { + if !reflect.DeepEqual(got[i], msgs[i]) { + t.Fatalf("under budget mutated message %d: got %+v want %+v", i, got[i], msgs[i]) + } + } +} + +func TestCompact_OverBudget_PreservesTrailingToolResult(t *testing.T) { + t.Parallel() + msgs := buildTurns(3, 400, true) + c := defaultCompactor{} + if estimateContextTokens(msgs) <= 1000 { + t.Fatalf("test precondition: estimated %d tokens, want > 1000", estimateContextTokens(msgs)) + } + got, err := c.Compact(context.Background(), msgs, 1000) + if err != nil { + t.Fatalf("Compact: %v", err) + } + if len(got) >= len(msgs) { + t.Fatalf("compaction should shrink the conversation: got %d, want < %d", len(got), len(msgs)) + } + // A leading summary system message marks the compaction. + if got[0].Role != model.RoleSystem || !strings.HasPrefix(messageText(got[0]), summaryMarker) { + t.Fatalf("first message is not a compaction summary: %+v", got[0]) + } + // The trailing tool result survives at the very end, unmodified. + last := got[len(got)-1] + if last.Role != model.RoleTool || !strings.Contains(last.Content[0].Text, "tool-result") { + t.Fatalf("trailing tool result was not preserved: %+v", last) + } + // Whenever a summary is emitted, the first kept message is a whole-turn + // boundary (a user message), so compaction never splits a turn. + if got[1].Role != model.RoleUser { + t.Fatalf("first kept message role = %q, want a turn-start user message", got[1].Role) + } +} + +func TestCompact_NeverSplitsTurn_MidConversation(t *testing.T) { + t.Parallel() + // Two turns of very different sizes: the second turn is huge, the first is + // small. Compaction must keep the whole huge last turn and summarize the + // small earlier turn(s), not slice the last turn apart. + big := strings.Repeat("b", 5000) + msgs := []model.Message{ + model.TextMessage(model.RoleUser, "question one"), + model.TextMessage(model.RoleAssistant, "answer one"), + model.TextMessage(model.RoleUser, "question two"), + model.TextMessage(model.RoleAssistant, big), + } + got, err := (defaultCompactor{}).Compact(context.Background(), msgs, 400) + if err != nil { + t.Fatalf("Compact: %v", err) + } + if len(got) >= len(msgs) { + t.Fatalf("expected compaction, got %d >= %d messages", len(got), len(msgs)) + } + if got[0].Role != model.RoleSystem { + t.Fatalf("expected a leading summary, got %+v", got[0]) + } + // The kept region begins at the second user turn and its assistant reply are + // both intact. + if got[1].Role != model.RoleUser || got[1].Content[0].Text != "question two" { + t.Fatalf("kept region should start a whole turn: %+v", got[1]) + } + if !strings.Contains(got[len(got)-1].Content[0].Text, big) { + t.Fatalf("last assistant reply was not preserved: %+v", got[len(got)-1]) + } +} + +func TestCompact_SummaryCap(t *testing.T) { + t.Parallel() + // A huge prefix so the naive summary would try to sample long text, but each + // sampled message is truncated; verify the capped length stays bounded. + msgs := buildTurns(8, 5000, false) + got, err := (defaultCompactor{}).Compact(context.Background(), msgs, 5000) + if err != nil { + t.Fatalf("Compact: %v", err) + } + if got[0].Role != model.RoleSystem { + t.Fatalf("expected a leading summary") + } + if n := len([]rune(got[0].Content[0].Text)); n > maxSummaryChars+len(summaryMarker)+2 { + t.Fatalf("summary too long: %d runes, want <= %d", n, maxSummaryChars+len(summaryMarker)+2) + } +} + +func TestCompact_ExistingSummaryMerged(t *testing.T) { + t.Parallel() + old := summaryMarker + "\n\n早前摘要内容" + msgs := append([]model.Message{model.TextMessage(model.RoleSystem, old)}, buildTurns(4, 300, false)...) + got, err := (defaultCompactor{}).Compact(context.Background(), msgs, 1000) + if err != nil { + t.Fatalf("Compact: %v", err) + } + // Only one leading summary remains and it merges the prior summary. + if got[0].Role != model.RoleSystem { + t.Fatalf("expected a leading summary") + } + text := messageText(got[0]) + if !strings.Contains(text, "早前摘要内容") { + t.Fatalf("existing summary was dropped: %q", text) + } + if summaryPrefixEnd(got) != 1 { + t.Fatalf("expected exactly one leading summary message") + } +} + +func TestCompact_CanceledContext_ReturnsError(t *testing.T) { + t.Parallel() + msgs := buildTurns(4, 300, false) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, err := (defaultCompactor{}).Compact(ctx, msgs, 1000) + if !errors.Is(err, context.Canceled) { + t.Fatalf("Compact error = %v, want wrapped context.Canceled", err) + } +} + +func TestEstimateContextTokens_Monotonic(t *testing.T) { + t.Parallel() + small := []model.Message{model.TextMessage(model.RoleUser, "hi")} + big := []model.Message{model.TextMessage(model.RoleUser, strings.Repeat("x", 1000))} + if estimateContextTokens(big) <= estimateContextTokens(small) { + t.Fatalf("token estimate not monotonic: big=%d small=%d", estimateContextTokens(big), estimateContextTokens(small)) + } +} diff --git a/internal/runtime/loop/loop.go b/internal/runtime/loop/loop.go index 7661df7..95d6286 100644 --- a/internal/runtime/loop/loop.go +++ b/internal/runtime/loop/loop.go @@ -12,6 +12,7 @@ package loop import ( + "context" "errors" "io" "log/slog" @@ -91,9 +92,12 @@ type Loop struct { maxRounds int toolExec string toolParallelism int - hooks ToolLoopHooks - onStream func(model.StreamEvent) - log *slog.Logger + hooks ToolLoopHooks + onStream func(model.StreamEvent) + maxContextTokens int + compactor Compactor + transformContext func(context.Context, []model.Message, int) []model.Message + log *slog.Logger } // New builds a Loop from opts, applying defaults. A non-positive MaxRounds @@ -114,23 +118,30 @@ func New(opts Options) *Loop { if toolParallelism <= 0 { toolParallelism = DefaultToolParallelism } + compactor := opts.Compactor + if compactor == nil { + compactor = defaultCompactor{} + } return &Loop{ - agent: opts.Agent, - tools: opts.Tools, - memory: opts.Memory, - retriever: opts.Retriever, - working: opts.WorkingMemory, - extractor: opts.Extractor, - checkpoint: opts.Checkpoint, - tracer: opts.Tracer, - guards: opts.Guards, - personaID: opts.PersonaID, - maxRounds: maxRounds, - toolExec: toolExec, - toolParallelism: toolParallelism, - hooks: opts.Hooks, - onStream: opts.OnStream, - log: sourcedLog(opts.Log), + agent: opts.Agent, + tools: opts.Tools, + memory: opts.Memory, + retriever: opts.Retriever, + working: opts.WorkingMemory, + extractor: opts.Extractor, + checkpoint: opts.Checkpoint, + tracer: opts.Tracer, + guards: opts.Guards, + personaID: opts.PersonaID, + maxRounds: maxRounds, + toolExec: toolExec, + toolParallelism: toolParallelism, + hooks: opts.Hooks, + onStream: opts.OnStream, + maxContextTokens: opts.MaxContextTokens, + compactor: compactor, + transformContext: opts.TransformContext, + log: sourcedLog(opts.Log), } } diff --git a/internal/runtime/loop/memory.go b/internal/runtime/loop/memory.go index 4eb817b..d5daebd 100644 --- a/internal/runtime/loop/memory.go +++ b/internal/runtime/loop/memory.go @@ -7,11 +7,12 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" ) -// injectMemory retrieves the persona's relevant long-term memories and injects -// them into the first-turn context as a leading system message, so the model can -// ground its answer in recalled context (design §6 / §2). It returns the input -// messages unchanged when no retriever/persona is configured, retrieval fails, or -// nothing relevant is found. +// injectMemory retrieves the persona's relevant long-term memories and prepends +// them as a leading system message, so the model can ground its answer in recalled +// context (design §6 / §2). It is re-run every round via assembleContext so the +// retrieval reflects the latest conversation rather than a first-turn snapshot. It +// returns the input messages unchanged when no retriever/persona is configured, +// retrieval fails, or nothing relevant is found. func (l *Loop) injectMemory(ctx context.Context, messages []model.Message) []model.Message { if l.retriever == nil || l.personaID == "" { return messages @@ -30,6 +31,50 @@ func (l *Loop) injectMemory(ctx context.Context, messages []model.Message) []mod return out } +// assembleContext builds the model-visible message list for one round (design +// §1.3 每轮 assemble, the loop's equivalent of OpenClaw's transformContext): it +// re-injects the persona's relevant long-term memory — retrieval is re-run every +// round from the current conversation, not cached from the first turn, so it +// supports cross-round refresh — folds in the per-session working-memory summary +// when present, and applies the optional TransformContext hook last so a caller +// can project/filter what the model actually sees. It returns the input unchanged +// when nothing is configured. +func (l *Loop) assembleContext(ctx context.Context, messages []model.Message, round int) []model.Message { + var leading []model.Message + if l.working != nil { + if s := l.working.Summarize(); s != "" { + leading = append(leading, model.TextMessage(model.RoleSystem, "工作记忆: "+s)) + } + } + leading = append(leading, l.injectMemory(ctx, messages)...) + + out := leading + if l.transformContext != nil { + out = l.transformContext(ctx, leading, round) + } + return out +} + +// compactIfNeeded compacts the conversation when its estimated context exceeds +// the configured MaxContextTokens. It is a no-op when compaction is disabled +// (non-positive MaxContextTokens), the compactor is nil, the estimate is within +// budget, or the compactor returns an error — a compaction failure is logged and +// the conversation is kept as-is rather than failing the run. +func (l *Loop) compactIfNeeded(ctx context.Context, messages []model.Message) []model.Message { + if l.maxContextTokens <= 0 || l.compactor == nil { + return messages + } + compacted, err := l.compactor.Compact(ctx, messages, l.maxContextTokens) + if err != nil { + l.log.Warn("context compact", "error", err, "tokens", l.maxContextTokens) + return messages + } + if compacted == nil { + return messages + } + return compacted +} + // extractMemory runs the extraction hook (design §6) over the current // conversation and persists any derived items. Errors are logged, never fatal, so // a bad extraction cannot abort the run. diff --git a/internal/runtime/loop/options.go b/internal/runtime/loop/options.go index a35ab42..492507d 100644 --- a/internal/runtime/loop/options.go +++ b/internal/runtime/loop/options.go @@ -1,6 +1,7 @@ package loop import ( + "context" "log/slog" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" @@ -27,7 +28,8 @@ type Options struct { Tools *tool.Dispatcher // Memory is the long-term memory store used to persist extracted items. Memory *memory.Store - // Retriever injects relevant memories into the first-turn context. + // Retriever injects relevant memories into the context, re-evaluated every + // round (see assembleContext) so retrieval tracks the latest conversation. Retriever *memory.Retriever // WorkingMemory is the per-session sliding window, provided for the // orchestrator to inspect; the loop does not mutate it. @@ -62,6 +64,25 @@ type Options struct { // nil-tolerant; these are reserved for "敏感操作代审" (design/11) and the // future asynchronous-tool path. A zero Hooks value is a no-op. Hooks ToolLoopHooks + // MaxContextTokens is the context budget that triggers overflow compaction. + // Non-positive (the zero value) disables compaction, preserving the loop's + // previous behavior. Setting it to a positive value (e.g. + // DefaultMaxContextTokens) enables compaction: when the loop's estimated + // context exceeds the budget it summarizes older turns into a leading summary + // message while keeping the recent tail (including trailing tool results) + // intact (design §1.3). Nil-tolerant: leaving it unset changes nothing. + MaxContextTokens int + // Compactor compacts an over-budget conversation. Nil selects the built-in + // structural compactor (no model call); a non-nil value lets a caller inject + // an LLM-backed or custom summarizer. It is only invoked when + // MaxContextTokens is positive. + Compactor Compactor + // TransformContext, when non-nil, is called immediately before every model + // step with the assembled message list (after memory injection and any + // compaction) and the current round number (1-based). It may project, filter + // or re-order the messages the model sees. It is the loop's equivalent of + // OpenClaw's transformContext (design §1.3). Nil = no projection. + TransformContext func(ctx context.Context, messages []model.Message, round int) []model.Message // Log is the base logger. Nil is replaced with a discard logger tagged with // source "runtime/loop". Log *slog.Logger diff --git a/internal/runtime/loop/run.go b/internal/runtime/loop/run.go index ab8caad..584474f 100644 --- a/internal/runtime/loop/run.go +++ b/internal/runtime/loop/run.go @@ -9,10 +9,12 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" ) -// Run drives the agent loop: inject relevant memory into the first-turn context, -// then run up to MaxRounds steps. Each step runs the model (streaming through -// OnStream when configured) and any requested tool calls are executed under a -// per-invoke timeout and fed back as RoleTool messages so the model can continue. +// Run drives the agent loop: assemble a fresh model-visible context every round +// (re-inject relevant memory from the current conversation, optionally compacting +// the older turns when the context exceeds the configured budget), then run up to +// MaxRounds steps. Each step runs the model (streaming through OnStream when +// configured) and any requested tool calls are executed under a per-invoke +// timeout and fed back as RoleTool messages so the model can continue. // A checkpoint is persisted at every step, every step is traced as a // state_transition span (with an llm_call child and tool_call spans for each // invocation), and after each step the memory Extractor runs with errors logged @@ -29,7 +31,7 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa return nil, fmt.Errorf("loop: run: %w", ErrNilAgent) } - msgs := l.injectMemory(ctx, messages) + msgs := messages toolState := &toolLoopState{prevFP: make(map[string]bool)} res := &Result{} for step := 1; step <= l.maxRounds; step++ { @@ -37,8 +39,14 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa return nil, fmt.Errorf("loop: run: step %d: %w", step, err) } + // Compact older turns when the conversation exceeds the context budget, + // then assemble the model-visible context for this round (fresh memory + // injection + transform hook), so every step sees an up-to-date context. + msgs = l.compactIfNeeded(ctx, msgs) + view := l.assembleContext(ctx, msgs, step) + rc, roundSpan := l.span(ctx, "loop.step", trace.SpanKindStateTransition) - outcome, err := l.step(rc, sessionID, msgs) + outcome, err := l.step(rc, sessionID, view) if err != nil { l.spanProp(roundSpan, "error", err.Error()) l.finishSpan(roundSpan, trace.SpanStatusError)