feat(loop): Top5#5 上下文溢出压缩 + 每轮 assemble(对齐 OpenClaw context-engine)——Top5 全完成
- 每轮 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) 全部完成并推库。**
This commit is contained in:
1 parent
35cb2d3942
commit
648292d035
7 files changed
+709
-31
No files matched your search
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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])
|
||||
}
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
@@ -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),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in new issue
Block a user