Split usage accounting end to end, retry by default, and take the WP6 pins (N3 WP6) #96
Vendored
+1
-1
Submodule deps/agent-native updated: 0e03cc2144...1ac21c68d4
Vendored
+1
-1
Submodule deps/agent-squad updated: bf7fe8429b...c462bc4ae5
Vendored
+1
-1
Submodule deps/agent-system-core updated: 0b5c7747a3...1b7e2c0f47
@@ -101,6 +101,10 @@ type runHandle struct {
|
||||
started time.Time
|
||||
promptTokens int
|
||||
completionTokens int
|
||||
lastPromptTokens int
|
||||
lastCompletionTokens int
|
||||
cacheReadTokens int
|
||||
cacheWriteTokens int
|
||||
}
|
||||
|
||||
type hitlAnswer struct {
|
||||
@@ -236,10 +240,15 @@ func (c *runCtx) ConversationSandbox() string {
|
||||
}
|
||||
|
||||
// AddUsage 累加本 run 的模型用量,finishRun 写入 run.Usage。
|
||||
// 计费拆分(N3 WP6):Last* 覆盖(最近一次请求)、Cache* 累计。
|
||||
func (c *runCtx) AddUsage(u core.Usage) {
|
||||
c.mgr.mu.Lock()
|
||||
c.h.promptTokens += u.PromptTokens
|
||||
c.h.completionTokens += u.CompletionTokens
|
||||
c.h.lastPromptTokens = u.LastPromptTokens
|
||||
c.h.lastCompletionTokens = u.LastCompletionTokens
|
||||
c.h.cacheReadTokens += u.CacheReadTokens
|
||||
c.h.cacheWriteTokens += u.CacheWriteTokens
|
||||
c.mgr.mu.Unlock()
|
||||
}
|
||||
func (c *runCtx) Subtask() string { return c.subTask }
|
||||
@@ -1080,6 +1089,10 @@ func (m *Manager) finishRun(rc *runCtx, status core.RunStatus, usage core.Usage,
|
||||
usage.PromptTokens = rc.h.promptTokens
|
||||
usage.CompletionTokens = rc.h.completionTokens
|
||||
usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens
|
||||
usage.LastPromptTokens = rc.h.lastPromptTokens
|
||||
usage.LastCompletionTokens = rc.h.lastCompletionTokens
|
||||
usage.CacheReadTokens = rc.h.cacheReadTokens
|
||||
usage.CacheWriteTokens = rc.h.cacheWriteTokens
|
||||
m.mu.Unlock()
|
||||
if m.shuttingDown.Load() && status == core.RunCancelled {
|
||||
status = core.RunInterrupted
|
||||
@@ -1395,7 +1408,17 @@ func (m *Manager) finishChild(rc *runCtx, runner runtime.Runner, emitResult bool
|
||||
return runner.Execute(rc)
|
||||
}()
|
||||
|
||||
// 子代理记账(N3 WP6):token 用量随子 run 入账(此前只有 DurationMs)。
|
||||
usage := core.Usage{DurationMs: float64(time.Since(rc.h.started).Milliseconds())}
|
||||
m.mu.Lock()
|
||||
usage.PromptTokens = rc.h.promptTokens
|
||||
usage.CompletionTokens = rc.h.completionTokens
|
||||
usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens
|
||||
usage.LastPromptTokens = rc.h.lastPromptTokens
|
||||
usage.LastCompletionTokens = rc.h.lastCompletionTokens
|
||||
usage.CacheReadTokens = rc.h.cacheReadTokens
|
||||
usage.CacheWriteTokens = rc.h.cacheWriteTokens
|
||||
m.mu.Unlock()
|
||||
status := core.RunCompleted
|
||||
summary := "完成"
|
||||
payload := core.AgentResultPayload{AgentRun: rc.run.ID, Status: status, Summary: summary}
|
||||
|
||||
@@ -207,3 +207,18 @@ func TestResumeSkipsCompletedMarker(t *testing.T) {
|
||||
t.Fatalf("completed 终态不得续跑:before=%v after=%v", before, ids)
|
||||
}
|
||||
}
|
||||
|
||||
// N3 WP6:子代理记账链——AddUsage 累计拆分字段,finishChild 合并进 child.Usage。
|
||||
func TestAddUsageAccumulatesSplitFields(t *testing.T) {
|
||||
c, _ := newSpawnEnv(t, &gateRunner{})
|
||||
c.h = &runHandle{}
|
||||
c.AddUsage(core.Usage{PromptTokens: 100, CompletionTokens: 10, LastPromptTokens: 100, LastCompletionTokens: 10, CacheReadTokens: 50, CacheWriteTokens: 4})
|
||||
c.AddUsage(core.Usage{PromptTokens: 60, CompletionTokens: 6, LastPromptTokens: 60, LastCompletionTokens: 6, CacheReadTokens: 20, CacheWriteTokens: 2})
|
||||
c.mgr.mu.Lock()
|
||||
defer c.mgr.mu.Unlock()
|
||||
if c.h.promptTokens != 160 || c.h.completionTokens != 16 ||
|
||||
c.h.lastPromptTokens != 60 || c.h.lastCompletionTokens != 6 ||
|
||||
c.h.cacheReadTokens != 70 || c.h.cacheWriteTokens != 6 {
|
||||
t.Fatalf("记账链:%+v", c.h)
|
||||
}
|
||||
}
|
||||
|
||||
+13
-2
@@ -205,9 +205,10 @@ func (l *LLMRunner) Execute(rc RunCtx) error {
|
||||
|
||||
stream := l.Stream
|
||||
if stream == nil {
|
||||
// 重试经库默认(WP6:runtime.llmRetries → Config.RetryAttempts 映射),
|
||||
// 此处不再显式包装(防双重)。
|
||||
client := openLLMClient(route.Base, route.Key)
|
||||
retries := int(eff.GetNumber("runtime.llmRetries", 2))
|
||||
stream = agentllm.WithRetry(client.Stream, retries, nil)
|
||||
stream = client.Stream
|
||||
}
|
||||
|
||||
// 知识注入:当前消息包含 @knowledge 引用时检索注入(最小实现:显式引用)。
|
||||
@@ -404,6 +405,16 @@ func outputTokenCap(budget core.RunBudget, eff core.EffectiveSettings, key strin
|
||||
// 组合根经 runCtx 注入,runmgr 的 storeMemory 实现之)。
|
||||
type memoryStore = agentmemory.Store
|
||||
|
||||
// retryAttemptsOf 把 runtime.llmRetries 的设置语义(0=不重试,N=N 次)
|
||||
// 映射到库 Config.RetryAttempts(0=默认 2,负值=关,正数=次数)。
|
||||
func retryAttemptsOf(eff core.EffectiveSettings) int {
|
||||
n := int(eff.GetNumber("runtime.llmRetries", 2))
|
||||
if n <= 0 {
|
||||
return -1
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// agentNameOf 是 OTel gen_ai.agent.name 的取值(画像 ID,空回退 agent-system)。
|
||||
func agentNameOf(prof core.AgentProfile) string {
|
||||
if strings.TrimSpace(prof.ID) != "" {
|
||||
|
||||
@@ -211,6 +211,7 @@ func (l *LLMRunner) runDirect(rc RunCtx, stream agentllm.StreamFunc, model strin
|
||||
Trace: l.Trace,
|
||||
AgentName: agentNameOf(rc.Profile()),
|
||||
Memory: st.memoryStore,
|
||||
RetryAttempts: retryAttemptsOf(eff),
|
||||
UnknownTool: func(call agentllm.ToolCall) agenttools.Result {
|
||||
return st.runTool(call.Name, call.Arguments)
|
||||
},
|
||||
@@ -275,6 +276,9 @@ func (st *directRun) noteToolDelta(ev agentllm.Event) {
|
||||
|
||||
func (st *directRun) onEvent(ev nruntime.Event) {
|
||||
switch ev.Type {
|
||||
case "auto_retry_start":
|
||||
// 重试可观测(N3 WP6):沿用 process part 形态。
|
||||
st.finalParts = append(st.finalParts, core.Part{Type: core.PartText, Text: "模型在输出前失败,正在重试(" + ev.Text + ")", Process: true})
|
||||
case "turn_start":
|
||||
st.responding = false
|
||||
st.acting = false
|
||||
@@ -314,6 +318,11 @@ func (st *directRun) onEvent(ev nruntime.Event) {
|
||||
st.usageTotal.PromptTokens += ev.PromptTokens
|
||||
st.usageTotal.CompletionTokens += ev.CompletionTokens
|
||||
st.usageTotal.TotalTokens += ev.TotalTokens
|
||||
// 计费拆分(N3 WP6):上一请求覆盖、缓存命中累计。
|
||||
st.usageTotal.LastPromptTokens = ev.PromptTokens
|
||||
st.usageTotal.LastCompletionTokens = ev.CompletionTokens
|
||||
st.usageTotal.CacheReadTokens += ev.CacheReadTokens
|
||||
st.usageTotal.CacheWriteTokens += ev.CacheWriteTokens
|
||||
st.queued = append([]agentllm.ToolCall(nil), ev.Message.ToolCalls...)
|
||||
hasCalls := len(ev.Message.ToolCalls) > 0
|
||||
if st.turnThink.Len() > 0 {
|
||||
|
||||
@@ -43,6 +43,7 @@ type fakeRunCtx struct {
|
||||
bgErr error
|
||||
sandbox string // 非空时 ConversationSandbox 返回之(N3 WP2 转录测试)
|
||||
memEntries []memEntry // 非空时 StoreMemory 返回之(N3 WP5 记忆注入测试)
|
||||
usageTotal core.Usage // AddUsage 累计(N3 WP6 记账断言用)
|
||||
bgCalls int
|
||||
}
|
||||
|
||||
@@ -62,6 +63,15 @@ func (f *fakeRunCtx) Emit(typ core.StreamEventType, payload any) error {
|
||||
return nil
|
||||
}
|
||||
func (f *fakeRunCtx) SetAssistantParts(parts []core.Part) error { f.assistant = parts; return nil }
|
||||
func (f *fakeRunCtx) AddUsage(u core.Usage) {
|
||||
f.usageTotal.PromptTokens += u.PromptTokens
|
||||
f.usageTotal.CompletionTokens += u.CompletionTokens
|
||||
f.usageTotal.TotalTokens += u.TotalTokens
|
||||
f.usageTotal.LastPromptTokens = u.LastPromptTokens
|
||||
f.usageTotal.LastCompletionTokens = u.LastCompletionTokens
|
||||
f.usageTotal.CacheReadTokens += u.CacheReadTokens
|
||||
f.usageTotal.CacheWriteTokens += u.CacheWriteTokens
|
||||
}
|
||||
func (f *fakeRunCtx) AskHITL(card core.HITLCard) (core.HITLResponse, error) {
|
||||
return core.HITLResponse{}, fmt.Errorf("not implemented")
|
||||
}
|
||||
|
||||
@@ -53,7 +53,11 @@ func (s *SquadRunner) Execute(rc RunCtx) error {
|
||||
toolParts = append(toolParts, core.Part{Type: core.PartToolResult, Process: true, ToolResult: &core.ToolResultPart{CallID: ev.Tool, Status: "ok", Summary: truncateRun(ev.Detail, 160)}})
|
||||
case "usage":
|
||||
if sink, ok := rc.(interface{ AddUsage(core.Usage) }); ok {
|
||||
sink.AddUsage(core.Usage{PromptTokens: ev.PromptTokens, CompletionTokens: ev.CompletionTokens, TotalTokens: ev.PromptTokens + ev.CompletionTokens})
|
||||
sink.AddUsage(core.Usage{
|
||||
PromptTokens: ev.PromptTokens, CompletionTokens: ev.CompletionTokens, TotalTokens: ev.PromptTokens + ev.CompletionTokens,
|
||||
LastPromptTokens: ev.LastPromptTokens, LastCompletionTokens: ev.LastCompletionTokens,
|
||||
CacheReadTokens: ev.CacheReadTokens, CacheWriteTokens: ev.CacheWriteTokens,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -111,7 +111,12 @@ func (l *SubLLMRunner) Execute(rc RunCtx) error {
|
||||
PermissionAsk: PermissionAskOf(rc),
|
||||
Trace: l.Trace,
|
||||
AgentName: agentNameOf(rc.Profile()),
|
||||
RetryAttempts: retryAttemptsOf(rc.Settings()),
|
||||
}, user)
|
||||
// 子代理记账(N3 WP6):token 用量入账(此前只有 DurationMs)。
|
||||
if sink, ok := rc.(interface{ AddUsage(core.Usage) }); ok {
|
||||
sink.AddUsage(st.usageTotal)
|
||||
}
|
||||
if len(st.parts) == 0 {
|
||||
st.parts = append(st.parts, core.Part{Type: core.PartText, Text: "(子代理未返回内容)"})
|
||||
}
|
||||
@@ -169,11 +174,12 @@ type subSlot struct {
|
||||
}
|
||||
|
||||
type subRun struct {
|
||||
rc RunCtx
|
||||
parts []core.Part
|
||||
queued []agentllm.ToolCall
|
||||
inflight *subSlot
|
||||
turnText strings.Builder
|
||||
rc RunCtx
|
||||
parts []core.Part
|
||||
queued []agentllm.ToolCall
|
||||
inflight *subSlot
|
||||
turnText strings.Builder
|
||||
usageTotal core.Usage
|
||||
}
|
||||
|
||||
func (st *subRun) onEvent(ev nruntime.Event) {
|
||||
@@ -184,6 +190,13 @@ func (st *subRun) onEvent(ev nruntime.Event) {
|
||||
st.turnText.WriteString(ev.Text)
|
||||
_ = st.rc.Emit(core.EvAgentProgress, core.AgentProgressPayload{AgentRun: st.rc.Run().ID, Note: truncateRun(st.turnText.String(), 80)})
|
||||
case "message_end":
|
||||
st.usageTotal.PromptTokens += ev.PromptTokens
|
||||
st.usageTotal.CompletionTokens += ev.CompletionTokens
|
||||
st.usageTotal.TotalTokens += ev.TotalTokens
|
||||
st.usageTotal.LastPromptTokens = ev.PromptTokens
|
||||
st.usageTotal.LastCompletionTokens = ev.CompletionTokens
|
||||
st.usageTotal.CacheReadTokens += ev.CacheReadTokens
|
||||
st.usageTotal.CacheWriteTokens += ev.CacheWriteTokens
|
||||
st.queued = append([]agentllm.ToolCall(nil), ev.Message.ToolCalls...)
|
||||
hasCalls := len(ev.Message.ToolCalls) > 0
|
||||
if body := strings.TrimSpace(st.turnText.String()); body != "" {
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
package runtime
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
agentllm "agent-native/runtime/llm"
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
// ---- N3 WP6:计费拆分与重试收口(backend 侧)----
|
||||
|
||||
// retryAttemptsOf 映射:设置 0=不重试→-1;N→N;默认 2→2。
|
||||
func TestRetryAttemptsMapping(t *testing.T) {
|
||||
if got := retryAttemptsOf(core.EffectiveSettings{}); got != 2 {
|
||||
t.Fatalf("默认应 2,得到 %d", got)
|
||||
}
|
||||
if got := retryAttemptsOf(core.EffectiveSettings{"runtime": map[string]any{"llmRetries": float64(0)}}); got != -1 {
|
||||
t.Fatalf("设置 0 应映射为关闭(-1),得到 %d", got)
|
||||
}
|
||||
if got := retryAttemptsOf(core.EffectiveSettings{"runtime": map[string]any{"llmRetries": float64(5)}}); got != 5 {
|
||||
t.Fatalf("设置 5 应映射为 5,得到 %d", got)
|
||||
}
|
||||
}
|
||||
|
||||
// 直连 run 的用量拆分:message_end 到达后 Last* 覆盖、cache 累计入账。
|
||||
func TestDirectRunUsageSplitOnRecord(t *testing.T) {
|
||||
rc := &fakeRunCtx{ctx: context.Background(), prof: core.AgentProfile{Model: core.ModelRoute{Chat: "test-model"}}}
|
||||
stream := func(ctx context.Context, model string, msgs []agentllm.Message, _ []agentllm.ToolDecl, _ agentllm.StreamOptions) <-chan agentllm.Event {
|
||||
ch := make(chan agentllm.Event, 1)
|
||||
ch <- agentllm.Event{Kind: agentllm.KindMessageEnd,
|
||||
Message: agentllm.Message{Role: agentllm.RoleAssistant, Content: "x", StopReason: agentllm.StopEnd},
|
||||
Usage: agentllm.Usage{PromptTokens: 120, CompletionTokens: 12, TotalTokens: 132, CacheReadTokens: 80, CacheWriteTokens: 6},
|
||||
}
|
||||
close(ch)
|
||||
return ch
|
||||
}
|
||||
r := &LLMRunner{Stream: stream}
|
||||
if err := r.Execute(rc); err != nil {
|
||||
t.Fatalf("Execute: %v", err)
|
||||
}
|
||||
u := rc.usageTotal
|
||||
if u.PromptTokens != 120 || u.CompletionTokens != 12 ||
|
||||
u.LastPromptTokens != 120 || u.LastCompletionTokens != 12 ||
|
||||
u.CacheReadTokens != 80 || u.CacheWriteTokens != 6 {
|
||||
t.Fatalf("用量拆分未入账:%+v", u)
|
||||
}
|
||||
}
|
||||
|
||||
// 子代理记账:subRun 的 message_end 累计 token 并经 AddUsage 入账。
|
||||
func TestSubRunAccountsTokens(t *testing.T) {
|
||||
rc := &fakeRunCtx{ctx: context.Background(), prof: core.AgentProfile{Model: core.ModelRoute{Chat: "test-model"}}}
|
||||
stream := func(ctx context.Context, model string, msgs []agentllm.Message, _ []agentllm.ToolDecl, _ agentllm.StreamOptions) <-chan agentllm.Event {
|
||||
ch := make(chan agentllm.Event, 1)
|
||||
ch <- agentllm.Event{Kind: agentllm.KindMessageEnd,
|
||||
Message: agentllm.Message{Role: agentllm.RoleAssistant, Content: "子结果", StopReason: agentllm.StopEnd},
|
||||
Usage: agentllm.Usage{PromptTokens: 66, CompletionTokens: 6, TotalTokens: 72, CacheReadTokens: 30},
|
||||
}
|
||||
close(ch)
|
||||
return ch
|
||||
}
|
||||
r := &SubLLMRunner{Stream: stream}
|
||||
if err := r.Execute(rc); err != nil {
|
||||
t.Fatalf("Execute: %v", err)
|
||||
}
|
||||
u := rc.usageTotal
|
||||
if u.PromptTokens != 66 || u.CompletionTokens != 6 ||
|
||||
u.LastPromptTokens != 66 || u.LastCompletionTokens != 6 || u.CacheReadTokens != 30 {
|
||||
t.Fatalf("子代理用量未入账:%+v", u)
|
||||
}
|
||||
}
|
||||
@@ -33,7 +33,10 @@ func (l *LLMRunner) runVerifier(rc RunCtx, body string, evidence []toolEvidence,
|
||||
return gates.Verdict{}, fmt.Errorf("验证器调用失败")
|
||||
}
|
||||
client := openLLMClient(route.Base, route.Key)
|
||||
stream = client.Stream
|
||||
// 验证器不经内核循环:显式接重试(WP6 同映射)与 5xx 回退链。
|
||||
stream = agentllm.WithFallback(
|
||||
agentllm.WithRetry(client.Stream, retryAttemptsOf(rc.Settings()), nil),
|
||||
ChatFallbacks(rc.Settings()), nil)
|
||||
}
|
||||
model := route.Model
|
||||
if model == "" {
|
||||
|
||||
Reference in New Issue
Block a user