Sakurasan
d5dac5a6fc
llm（三协议 Provider）/ session（JSONL 持久化）/ tools（Registry、围栏、 builtin、computer）/ cdp（browser）/ skills / mcp / agent（循环、两级压缩、 编排、审批门控）+ 底座 jsonl/textx/execx/diag/scratch/spill/imagecache。 产品装配代码（config 桥接、schedule 工具、memory/plan/retro/prompt） 留在 AutoGo，宿主接入方式见 MIGRATION.md。
330 lines
12 KiB
Go
330 lines
12 KiB
Go
package agent
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"strings"
|
||
"sync"
|
||
"testing"
|
||
"time"
|
||
|
||
"git.gopher.ink/mirrors2/AutoZero/llm"
|
||
"git.gopher.ink/mirrors2/AutoZero/session"
|
||
"git.gopher.ink/mirrors2/AutoZero/tools"
|
||
)
|
||
|
||
// noWait 把重试的退避压到毫秒级、预算压到一小段，免得用例白等。预算刻意只够几十次
|
||
// 1ms 退避：既足够证明"会一直重试"，又不会让"始终失败"那条用例跑上一分钟。
|
||
func noWait(t *testing.T) {
|
||
t.Helper()
|
||
f0, c0, b0 := streamRetryFirst, streamRetryCeil, streamRetryBudget
|
||
streamRetryFirst, streamRetryCeil, streamRetryBudget = time.Millisecond, 2*time.Millisecond, 50*time.Millisecond
|
||
t.Cleanup(func() {
|
||
streamRetryFirst, streamRetryCeil, streamRetryBudget = f0, c0, b0
|
||
})
|
||
}
|
||
|
||
func newEmptyTestAgent(prov llm.Provider) *Agent {
|
||
return &Agent{
|
||
Provider: prov,
|
||
Tools: tools.NewRegistry(),
|
||
Session: &session.Session{},
|
||
System: "sys",
|
||
Opt: Options{Stdout: io.Discard, Stderr: io.Discard},
|
||
}
|
||
}
|
||
|
||
func countAssistant(msgs []llm.Message) int {
|
||
n := 0
|
||
for _, m := range msgs {
|
||
if m.Role == llm.RoleAssistant {
|
||
n++
|
||
}
|
||
}
|
||
return n
|
||
}
|
||
|
||
// TestAgentRetriesEmptyCompletion：空回复由 agent 自己重试掉，不该让人去按「继续」。
|
||
// 真实取证：会话 20261009-213751 里 18 个空回合对应 23 次人工「继续」。
|
||
func TestAgentRetriesEmptyCompletion(t *testing.T) {
|
||
noWait(t)
|
||
var mu sync.Mutex
|
||
calls := 0
|
||
prov := &scriptedProvider{fn: func(_ int, _ string, _ []llm.Message, _ []llm.ToolDef) (llm.Message, error) {
|
||
mu.Lock()
|
||
calls++
|
||
n := calls
|
||
mu.Unlock()
|
||
if n <= 2 {
|
||
return llm.Message{}, fmt.Errorf("openai: %w", llm.ErrEmptyCompletion)
|
||
}
|
||
return llm.Message{Role: llm.RoleAssistant, Content: "终于说出来了"}, nil
|
||
}}
|
||
a := newEmptyTestAgent(prov)
|
||
if err := a.Run(context.Background(), "hi"); err != nil {
|
||
t.Fatalf("重试后该成功：%v", err)
|
||
}
|
||
if calls != 3 {
|
||
t.Fatalf("模型调用 %d 次，want 3（两次空 + 一次成功）", calls)
|
||
}
|
||
// 会话里绝不能留下空 assistant 消息：它会一直挂在历史里，还会被后面的
|
||
// "没有 tool_calls 就是最终回答"判定误读。
|
||
if n := countAssistant(a.Session.Messages); n != 1 {
|
||
t.Fatalf("会话里有 %d 条 assistant，want 1", n)
|
||
}
|
||
}
|
||
|
||
// TestAgentEmptyCompletionFailsLoudly：预算见底之后必须**报错**，而不是静默"成功
|
||
// 收尾"——静默正是原来那个体验：一个字不出，用户不知道发生了什么。同时它得比原来
|
||
// 那"2 次 400ms"多熬很久，否则网关一簇几十秒的抖动照样会中断整轮。
|
||
func TestAgentEmptyCompletionFailsLoudly(t *testing.T) {
|
||
noWait(t)
|
||
prov := &scriptedProvider{fn: func(_ int, _ string, _ []llm.Message, _ []llm.ToolDef) (llm.Message, error) {
|
||
return llm.Message{}, fmt.Errorf("anthropic: %w", llm.ErrEmptyCompletion)
|
||
}}
|
||
a := newEmptyTestAgent(prov)
|
||
err := a.Run(context.Background(), "hi")
|
||
if !errors.Is(err, llm.ErrEmptyCompletion) {
|
||
t.Fatalf("该把空回复作为失败返回，得到 %v", err)
|
||
}
|
||
if prov.calls <= 1+2 {
|
||
t.Fatalf("只重试了 %d 次就放弃：按时间的预算没有生效（want 远大于 3）", prov.calls)
|
||
}
|
||
if n := countAssistant(a.Session.Messages); n != 0 {
|
||
t.Fatalf("失败的回合留下了 %d 条 assistant，want 0", n)
|
||
}
|
||
}
|
||
|
||
// TestStreamRetryCoversStallAndNetwork：90 秒空闲看门狗与连接被掐这两类失败，
|
||
// 以前**一次都不重试**（判据只认空补全与截断流），看门狗一响整轮就判死。
|
||
func TestStreamRetryCoversStallAndNetwork(t *testing.T) {
|
||
noWait(t)
|
||
cases := []struct {
|
||
name string
|
||
fail func() error
|
||
}{
|
||
{"流卡死", func() error { return &llm.StalledError{Idle: 90 * time.Second} }},
|
||
{"连接被掐", func() error { return errors.New("read tcp 10.0.0.5:52314->1.2.3.4:443: connection reset by peer") }},
|
||
{"网关过载", func() error { return errors.New("openai: HTTP 503: upstream busy") }},
|
||
}
|
||
for _, c := range cases {
|
||
t.Run(c.name, func(t *testing.T) {
|
||
calls := 0
|
||
prov := &scriptedProvider{fn: func(_ int, _ string, _ []llm.Message, _ []llm.ToolDef) (llm.Message, error) {
|
||
calls++
|
||
if calls <= 2 {
|
||
return llm.Message{}, c.fail()
|
||
}
|
||
return llm.Message{Role: llm.RoleAssistant, Content: "熬过来了"}, nil
|
||
}}
|
||
a := newEmptyTestAgent(prov)
|
||
if err := a.Run(context.Background(), "hi"); err != nil {
|
||
t.Fatalf("重试后该交付：%v", err)
|
||
}
|
||
if calls != 3 {
|
||
t.Fatalf("调用 %d 次，want 3", calls)
|
||
}
|
||
})
|
||
}
|
||
}
|
||
|
||
// TestDeterministicFailureIsNotRetried：HTTP 400 这类"网关点名说请求本身不合法"是
|
||
// 确定性失败，重发只会再撞一次。它该一次就交出去（大上下文时走压缩那条自救，见
|
||
// oversized_test.go），而不是在重试环里烧预算。
|
||
func TestDeterministicFailureIsNotRetried(t *testing.T) {
|
||
noWait(t)
|
||
prov := &scriptedProvider{fn: func(_ int, _ string, _ []llm.Message, _ []llm.ToolDef) (llm.Message, error) {
|
||
return llm.Message{}, errors.New(`openai: HTTP 400: {"error":{"message":"inference request is invalid","type":"invalid_request_error","code":"400002"}}`)
|
||
}}
|
||
a := newEmptyTestAgent(prov)
|
||
if err := a.Run(context.Background(), "hi"); err == nil {
|
||
t.Fatal("该以失败结束")
|
||
}
|
||
if prov.calls != 1 {
|
||
t.Fatalf("确定性失败重试了 %d 次，want 1", prov.calls)
|
||
}
|
||
}
|
||
|
||
// TestStreamRetryIsVisible：重试不能是静默的——用户得知道"还在试、为什么还没好"。
|
||
// 第一次重试发一句系统说明进界面时间线，日志里那句照旧。
|
||
func TestStreamRetryIsVisible(t *testing.T) {
|
||
noWait(t)
|
||
var mu sync.Mutex
|
||
calls := 0
|
||
prov := &scriptedProvider{fn: func(_ int, _ string, _ []llm.Message, _ []llm.ToolDef) (llm.Message, error) {
|
||
mu.Lock()
|
||
calls++
|
||
n := calls
|
||
mu.Unlock()
|
||
if n <= 3 {
|
||
return llm.Message{}, fmt.Errorf("openai: %w", llm.ErrEmptyCompletion)
|
||
}
|
||
return llm.Message{Role: llm.RoleAssistant, Content: "好了"}, nil
|
||
}}
|
||
a := newEmptyTestAgent(prov)
|
||
var notices []string
|
||
a.Opt.OnNotice = func(text string, failed bool) {
|
||
if failed {
|
||
notices = append(notices, text)
|
||
}
|
||
}
|
||
if err := a.Run(context.Background(), "hi"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if len(notices) == 0 {
|
||
t.Fatal("一次重试都没告诉用户")
|
||
}
|
||
if !strings.Contains(notices[0], "正在重发这一步") {
|
||
t.Errorf("第一句该说清在重试：%q", notices[0])
|
||
}
|
||
}
|
||
|
||
// TestEmptyCompletionNotRetriedWhenCancelled：用户喊停之后不再多打一次请求。
|
||
// 重试是给"网关抽风"用的，不是给"我已经取消了"用的。
|
||
func TestEmptyCompletionNotRetriedWhenCancelled(t *testing.T) {
|
||
noWait(t)
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
prov := &scriptedProvider{fn: func(c int, _ string, _ []llm.Message, _ []llm.ToolDef) (llm.Message, error) {
|
||
if c == 1 {
|
||
cancel() // 第一次空回复的同时用户取消
|
||
}
|
||
return llm.Message{}, fmt.Errorf("openai: %w", llm.ErrEmptyCompletion)
|
||
}}
|
||
a := newEmptyTestAgent(prov)
|
||
if err := a.Run(ctx, "hi"); err == nil {
|
||
t.Fatal("取消后该以失败结束")
|
||
}
|
||
if prov.calls != 1 {
|
||
t.Fatalf("取消后仍重试了：调用 %d 次，want 1", prov.calls)
|
||
}
|
||
}
|
||
|
||
// TestEmptyCompletionRetryKeepsWorkingAfterRecovery：重试不是"这步作废"——
|
||
// 恢复之后回合要能继续往下走（工具调用照常执行）。
|
||
func TestEmptyCompletionRetryKeepsWorkingAfterRecovery(t *testing.T) {
|
||
noWait(t)
|
||
calls := 0
|
||
prov := &scriptedProvider{fn: func(_ int, _ string, _ []llm.Message, _ []llm.ToolDef) (llm.Message, error) {
|
||
calls++
|
||
switch calls {
|
||
case 1:
|
||
return llm.Message{}, fmt.Errorf("openai: %w", llm.ErrEmptyCompletion)
|
||
case 2:
|
||
return llm.Message{Role: llm.RoleAssistant,
|
||
ToolCalls: []llm.ToolCall{{ID: "c1", Name: "noop", Input: "{}"}}}, nil
|
||
default:
|
||
return llm.Message{Role: llm.RoleAssistant, Content: "收尾"}, nil
|
||
}
|
||
}}
|
||
reg := tools.NewRegistry()
|
||
if err := reg.Register(&tools.Tool{Name: "noop", Description: "d",
|
||
Parameters: map[string]any{"type": "object"},
|
||
Run: func(context.Context, string) (tools.Result, error) { return tools.TextResult("ok"), nil }}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
a := newEmptyTestAgent(prov)
|
||
a.Tools = reg
|
||
if err := a.Run(context.Background(), "hi"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if calls != 3 {
|
||
t.Fatalf("调用 %d 次，want 3", calls)
|
||
}
|
||
if got := a.Session.Messages[len(a.Session.Messages)-1].Content; got != "收尾" {
|
||
t.Fatalf("最后一条 = %q", got)
|
||
}
|
||
}
|
||
|
||
// switchStub 是一个能切模型的 provider：primary 前 failsFirst 次都回空回复，
|
||
// 换到 backup 之后一次成功。seen 记录每次调用时生效的模型名。
|
||
type switchStub struct {
|
||
model string
|
||
failsLeft int
|
||
seen []string
|
||
}
|
||
|
||
func (p *switchStub) Stream(context.Context, string, []llm.Message, []llm.ToolDef,
|
||
llm.StreamOptions, func(llm.StreamDelta)) (llm.Message, error) {
|
||
p.seen = append(p.seen, p.model)
|
||
if p.failsLeft > 0 {
|
||
p.failsLeft--
|
||
return llm.Message{}, fmt.Errorf("openai: %w", llm.ErrEmptyCompletion)
|
||
}
|
||
return llm.Message{Role: llm.RoleAssistant, Content: "ok from " + p.model}, nil
|
||
}
|
||
func (p *switchStub) SetModel(m string) { p.model = m }
|
||
func (p *switchStub) CurrentModel() string { return p.model }
|
||
|
||
// TestStreamRetrySwitchesModelAfterStreak：同一个模型连试 3 次还没答上就换备选模型。
|
||
// 一簇抖动通常头几次就过去；连到第 3 次更像"这条上游在抽风"，换人比继续耗着值钱。
|
||
// 换模型是粘性的（与 fallbackTo 的既有语义一致），且必须让用户看得见。
|
||
func TestStreamRetrySwitchesModelAfterStreak(t *testing.T) {
|
||
noWait(t)
|
||
p := &switchStub{model: "primary", failsLeft: 5}
|
||
a := newEmptyTestAgent(p)
|
||
a.FallbackModel = "backup"
|
||
var notices []string
|
||
a.Opt.OnNotice = func(text string, failed bool) {
|
||
if failed {
|
||
notices = append(notices, text)
|
||
}
|
||
}
|
||
if err := a.Run(context.Background(), "hi"); err != nil {
|
||
t.Fatalf("该靠换模型交付，得到 %v", err)
|
||
}
|
||
// failsLeft=5：前三次落在 primary（第 3 次之后才换人），后两次落在 backup，
|
||
// 第六次才交付——换模型不是"换一次就算了"，换完照样按预算继续退避重试。
|
||
if strings.Join(p.seen, ",") != "primary,primary,primary,backup,backup,backup" {
|
||
t.Fatalf("调用落在哪些模型上 = %v", p.seen)
|
||
}
|
||
if p.CurrentModel() != "backup" {
|
||
t.Errorf("切换该是粘性的，当前模型 = %s", p.CurrentModel())
|
||
}
|
||
joined := strings.Join(notices, " | ")
|
||
if !strings.Contains(joined, "备选模型 backup") {
|
||
t.Errorf("换模型没告诉用户：%q", joined)
|
||
}
|
||
}
|
||
|
||
// TestStreamRetryKeepsPrimaryForASingleBlip：只抖一次就不该把整轮从用户选的模型上
|
||
// 拽走——换模型要等到连试 3 次之后。
|
||
func TestStreamRetryKeepsPrimaryForASingleBlip(t *testing.T) {
|
||
noWait(t)
|
||
p := &switchStub{model: "primary", failsLeft: 2}
|
||
a := newEmptyTestAgent(p)
|
||
a.FallbackModel = "backup"
|
||
if err := a.Run(context.Background(), "hi"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if p.CurrentModel() != "primary" {
|
||
t.Errorf("抖两次就换模型了，当前 = %s，want 仍在 primary", p.CurrentModel())
|
||
}
|
||
if len(p.seen) != 3 {
|
||
t.Fatalf("调用 %d 次，want 3（两次空 + 一次成功）", len(p.seen))
|
||
}
|
||
}
|
||
|
||
// TestStreamRetryUnlimitedBudget：预算设成负数＝不限时长，一直重试到交付为止
|
||
// （只有停止/取消能打断）。用例里靠 provider 第 30 次成功来收尾。
|
||
func TestStreamRetryUnlimitedBudget(t *testing.T) {
|
||
noWait(t)
|
||
calls := 0
|
||
prov := &scriptedProvider{fn: func(_ int, _ string, _ []llm.Message, _ []llm.ToolDef) (llm.Message, error) {
|
||
calls++
|
||
if calls < 30 {
|
||
return llm.Message{}, fmt.Errorf("openai: %w", llm.ErrEmptyCompletion)
|
||
}
|
||
return llm.Message{Role: llm.RoleAssistant, Content: "熬到第 30 次"}, nil
|
||
}}
|
||
a := newEmptyTestAgent(prov)
|
||
a.StreamRetryBudget = -time.Second // 不限
|
||
if err := a.Run(context.Background(), "hi"); err != nil {
|
||
t.Fatalf("不限预算时该一直重试到交付：%v", err)
|
||
}
|
||
if calls != 30 {
|
||
t.Fatalf("调用 %d 次，want 30", calls)
|
||
}
|
||
}
|