Compare commits
7
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a5e3c7c817 | ||
|
|
6e972335c9 | ||
|
|
e86e74bc36 | ||
|
|
94afd71f00 | ||
|
|
c348441691 | ||
|
|
7f60baeaf8 | ||
|
|
ef2fa75006 |
@@ -39,7 +39,7 @@ func vocabScript() llm.StreamFunc {
|
||||
step++
|
||||
ch <- llm.Event{Kind: llm.KindReasoningDelta, ReasoningDelta: "思考"}
|
||||
ch <- llm.Event{Kind: llm.KindTextDelta, Text: "正文"}
|
||||
ch <- llm.Event{Kind: llm.KindToolCallDelta, ToolName: "list_dir", ArgumentsDelta: `{"path":"."}`}
|
||||
ch <- llm.Event{Kind: llm.KindToolCallDelta, ToolIndex: 2, ToolID: "c1", ToolName: "list_dir", ArgumentsDelta: `{"path":"."}`}
|
||||
ch <- llm.Event{Kind: llm.KindMessageEnd, Message: llm.Message{
|
||||
Role: llm.RoleAssistant, StopReason: llm.StopToolCalls,
|
||||
ToolCalls: []llm.ToolCall{{ID: "c1", Name: "list_dir", Arguments: `{"path":"."}`}},
|
||||
@@ -54,6 +54,25 @@ func vocabScript() llm.StreamFunc {
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoopPreservesToolCallDeltaIdentity(t *testing.T) {
|
||||
reg := tools.WorkspaceTools(t.TempDir())
|
||||
var got Event
|
||||
_, err := Run(context.Background(), Config{
|
||||
Model: "fake", Tools: reg, Stream: vocabScript(), MaxTurns: 4,
|
||||
OnEvent: func(ev Event) {
|
||||
if ev.Type == "tool_call_delta" {
|
||||
got = ev
|
||||
}
|
||||
},
|
||||
}, llm.Message{Role: llm.RoleUser, Content: "run"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got.ToolIndex != 2 || got.CallID != "c1" || got.Tool != "list_dir" || got.Detail != `{"path":"."}` {
|
||||
t.Fatalf("tool delta lost its provider identity: %#v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoopEmitsProcessVocabulary(t *testing.T) {
|
||||
reg := tools.WorkspaceTools(t.TempDir())
|
||||
var observed []string
|
||||
|
||||
+70
-9
@@ -58,6 +58,19 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
|
||||
var usage Usage
|
||||
finished := ""
|
||||
sawDone := false
|
||||
sawUsage := false
|
||||
sendTerminalFailure := func(why string) {
|
||||
message := Message{Role: RoleAssistant, Content: content.String(), StopReason: StopError}
|
||||
if reasoning.Len() > 0 {
|
||||
message.Thinking = []ThinkingBlock{{Thinking: reasoning.String()}}
|
||||
}
|
||||
failed := terminalStreamFailure(message, usage, finished, sawDone, why)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
emitAbort(out, content.String(), ctx.Err())
|
||||
case out <- failed:
|
||||
}
|
||||
}
|
||||
sc := bufio.NewScanner(r)
|
||||
sc.Buffer(make([]byte, 0, 64*1024), 4*1024*1024)
|
||||
reader := &sseEventReader{sc: sc}
|
||||
@@ -76,17 +89,26 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
|
||||
}
|
||||
var chunk sseChunk
|
||||
if err := json.Unmarshal([]byte(data), &chunk); err != nil {
|
||||
if sawUsage {
|
||||
sendTerminalFailure("malformed stream chunk after reported usage")
|
||||
return
|
||||
}
|
||||
emitError(out, content.String(), false, "malformed stream chunk: "+err.Error())
|
||||
return
|
||||
}
|
||||
if chunk.Error != nil && chunk.Error.Message != "" {
|
||||
msg := chunk.Error.Message
|
||||
if sawUsage {
|
||||
sendTerminalFailure("provider error after reported usage: " + msg)
|
||||
return
|
||||
}
|
||||
retryable := CapacityLimited(msg) && !ContextOverflow(msg)
|
||||
emitError(out, content.String(), retryable, msg)
|
||||
return
|
||||
}
|
||||
if chunk.Usage != nil {
|
||||
usage = chunk.Usage.Usage
|
||||
sawUsage = true
|
||||
if chunk.Usage.Details.Cached > 0 {
|
||||
usage.CacheReadTokens = chunk.Usage.Details.Cached
|
||||
}
|
||||
@@ -157,6 +179,10 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
|
||||
emitAbort(out, content.String(), ctx.Err())
|
||||
return
|
||||
}
|
||||
if sawUsage {
|
||||
sendTerminalFailure("stream read failed after reported usage")
|
||||
return
|
||||
}
|
||||
emitError(out, content.String(), true, err.Error())
|
||||
return
|
||||
}
|
||||
@@ -166,9 +192,38 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
|
||||
emitAbort(out, content.String(), ctx.Err())
|
||||
return
|
||||
}
|
||||
// length、空结束原因,或没有 [DONE] 的中断,都不把半截工具调用交给执行层。
|
||||
if !toolFinishExecutable(finished, sawDone) {
|
||||
order = nil
|
||||
// A finish reason alone does not prove the tool arguments arrived intact.
|
||||
// Preserve returned usage when rejecting a partial terminal, so the host
|
||||
// can charge this call and stop rather than treating it as a free retry.
|
||||
if (finished == "tool_calls" || finished == "function_call") && len(order) == 0 {
|
||||
sendTerminalFailure("tool call finish reason without a tool call")
|
||||
return
|
||||
}
|
||||
if len(order) > 0 {
|
||||
problem := ""
|
||||
switch {
|
||||
case !sawDone:
|
||||
problem = "tool call stream ended before [DONE]"
|
||||
case !toolFinishExecutable(finished, sawDone):
|
||||
problem = "tool call ended without a supported finish reason"
|
||||
default:
|
||||
for _, idx := range order {
|
||||
call := calls[idx]
|
||||
var args map[string]json.RawMessage
|
||||
if call.ID == "" || call.Name == "" {
|
||||
problem = "tool call identity is incomplete"
|
||||
break
|
||||
}
|
||||
if err := json.Unmarshal([]byte(call.Arguments), &args); err != nil || args == nil {
|
||||
problem = "tool call arguments are not a JSON object"
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
if problem != "" {
|
||||
sendTerminalFailure(problem)
|
||||
return
|
||||
}
|
||||
}
|
||||
msg := Message{Role: RoleAssistant, Content: content.String(), StopReason: mapFinish(finished, len(order) > 0)}
|
||||
if reasoning.Len() > 0 {
|
||||
@@ -183,10 +238,17 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
emitAbort(out, content.String(), ctx.Err())
|
||||
case out <- Event{Kind: KindMessageEnd, Message: msg, Usage: usage, FinishReason: finished}:
|
||||
case out <- Event{Kind: KindMessageEnd, Message: msg, Usage: usage, FinishReason: finished, StreamComplete: sawDone}:
|
||||
}
|
||||
}
|
||||
|
||||
func terminalStreamFailure(message Message, usage Usage, reason string, complete bool, why string) Event {
|
||||
err := &StreamError{Message: why, Retryable: false}
|
||||
return Event{Kind: KindMessageEnd, Message: message, Usage: usage,
|
||||
FinishReason: reason, StreamComplete: complete, Err: err,
|
||||
ErrorMessage: why, Retryable: false}
|
||||
}
|
||||
|
||||
// sseEventReader 按空行切事件,并把同一事件里的多行 data: 用换行拼回。
|
||||
// 已经是完整 JSON 的 data 行单独成事件,这样单行流不用空行也能解析。
|
||||
type sseEventReader struct {
|
||||
@@ -241,17 +303,16 @@ func (r *sseEventReader) next() (string, bool) {
|
||||
// toolFinishExecutable 只在供应商明确以工具调用或正常 stop 结束时执行。
|
||||
// length、缺失的结束原因,以及没有 [DONE] 的断开,都视为未完成。
|
||||
func toolFinishExecutable(reason string, sawDone bool) bool {
|
||||
if !sawDone {
|
||||
return false
|
||||
}
|
||||
switch reason {
|
||||
case "tool_calls", "function_call":
|
||||
return true
|
||||
case "stop":
|
||||
return true
|
||||
case "length":
|
||||
return false
|
||||
case "":
|
||||
return false
|
||||
default:
|
||||
return sawDone && reason != "error" && reason != "content_filter"
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+177
-6
@@ -93,7 +93,29 @@ func TestParseSSEToolCallIncrements(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseSSELengthDropsUnfinishedToolCall(t *testing.T) {
|
||||
func TestParseSSEToolFinishWithoutCallFails(t *testing.T) {
|
||||
for _, reason := range []string{"tool_calls", "function_call"} {
|
||||
t.Run(reason, func(t *testing.T) {
|
||||
body := strings.Join([]string{
|
||||
`data: {"choices":[{"delta":{},"finish_reason":"` + reason + `"}],"usage":{"prompt_tokens":12,"completion_tokens":5,"total_tokens":17}}`,
|
||||
`data: [DONE]`,
|
||||
"",
|
||||
}, "\n")
|
||||
var end Event
|
||||
for ev := range ParseSSE(context.Background(), strings.NewReader(body)) {
|
||||
if ev.Kind == KindMessageEnd {
|
||||
end = ev
|
||||
}
|
||||
}
|
||||
if end.Err == nil || end.Message.StopReason != StopError || end.Usage.TotalTokens != 17 ||
|
||||
!end.StreamComplete || len(end.Message.ToolCalls) != 0 {
|
||||
t.Fatalf("empty tool finish must fail and retain Usage: %#v", end)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseSSELengthFailsUnfinishedToolCall(t *testing.T) {
|
||||
body := strings.Join([]string{
|
||||
`data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"read_file","arguments":"{\"path\":"}}]}}]}`,
|
||||
`data: {"choices":[{"delta":{},"finish_reason":"length"}]}`,
|
||||
@@ -106,15 +128,15 @@ func TestParseSSELengthDropsUnfinishedToolCall(t *testing.T) {
|
||||
end = ev
|
||||
}
|
||||
}
|
||||
if end.Message.StopReason != StopEnd || len(end.Message.ToolCalls) != 0 {
|
||||
t.Fatalf("truncated tool call must not be executable: %#v", end.Message)
|
||||
if end.Message.StopReason != StopError || len(end.Message.ToolCalls) != 0 || end.Err == nil {
|
||||
t.Fatalf("truncated tool call must fail without execution: %#v", end)
|
||||
}
|
||||
if end.FinishReason != "length" {
|
||||
t.Fatalf("finish %q", end.FinishReason)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseSSEUncleanEndDropsToolCall(t *testing.T) {
|
||||
func TestParseSSEUncleanEndFailsToolCall(t *testing.T) {
|
||||
body := `data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"read_file","arguments":"{\"path\":"}}]}}]}`
|
||||
var end Event
|
||||
for ev := range ParseSSE(context.Background(), strings.NewReader(body)) {
|
||||
@@ -122,8 +144,97 @@ func TestParseSSEUncleanEndDropsToolCall(t *testing.T) {
|
||||
end = ev
|
||||
}
|
||||
}
|
||||
if end.Kind != KindMessageEnd || end.Message.StopReason != StopEnd || len(end.Message.ToolCalls) != 0 {
|
||||
t.Fatalf("unclean stream must not execute tools: %#v", end)
|
||||
if end.Kind != KindMessageEnd || end.Message.StopReason != StopError || len(end.Message.ToolCalls) != 0 || end.Err == nil {
|
||||
t.Fatalf("unclean stream must fail without executing tools: %#v", end)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseSSEToolFinishWithoutDone(t *testing.T) {
|
||||
body := strings.Join([]string{
|
||||
`data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"echo","arguments":"{}"}}]}}]}`,
|
||||
`data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":12,"completion_tokens":5,"total_tokens":17}}`,
|
||||
"",
|
||||
}, "\n\n")
|
||||
var end Event
|
||||
for ev := range ParseSSE(context.Background(), strings.NewReader(body)) {
|
||||
if ev.Kind == KindMessageEnd {
|
||||
end = ev
|
||||
}
|
||||
}
|
||||
if end.Kind != KindMessageEnd || end.Message.StopReason != StopError ||
|
||||
len(end.Message.ToolCalls) != 0 || end.Err == nil || end.Retryable ||
|
||||
end.Usage.TotalTokens != 17 {
|
||||
t.Fatalf("unsafe tool call from unfinished stream: %#v", end)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseSSEInvalidToolArguments(t *testing.T) {
|
||||
body := strings.Join([]string{
|
||||
`data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"publish_work_checkpoint","arguments":"{\"findings\":"}}]}}]}`,
|
||||
`data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":12,"completion_tokens":5,"total_tokens":17}}`,
|
||||
`data: [DONE]`,
|
||||
"",
|
||||
}, "\n\n")
|
||||
var end Event
|
||||
for ev := range ParseSSE(context.Background(), strings.NewReader(body)) {
|
||||
if ev.Kind == KindMessageEnd {
|
||||
end = ev
|
||||
}
|
||||
}
|
||||
if end.Kind != KindMessageEnd || end.Message.StopReason != StopError ||
|
||||
len(end.Message.ToolCalls) != 0 || end.Err == nil || end.Retryable ||
|
||||
end.Usage.TotalTokens != 17 {
|
||||
t.Fatalf("unsafe tool call with invalid JSON arguments: %#v", end)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseSSERejectsMissingToolIdentity(t *testing.T) {
|
||||
for _, tc := range []struct{ name, firstChunk string }{
|
||||
{"missing-id", `data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"name":"echo","arguments":"{}"}}]}}]}`},
|
||||
{"missing-name", `data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"arguments":"{}"}}]}}]}`},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
body := strings.Join([]string{
|
||||
tc.firstChunk,
|
||||
`data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":12,"completion_tokens":5,"total_tokens":17}}`,
|
||||
`data: [DONE]`,
|
||||
"",
|
||||
}, "\n\n")
|
||||
var end Event
|
||||
for ev := range ParseSSE(context.Background(), strings.NewReader(body)) {
|
||||
if ev.Kind == KindMessageEnd {
|
||||
end = ev
|
||||
}
|
||||
}
|
||||
if end.Message.StopReason != StopError || len(end.Message.ToolCalls) != 0 || end.Err == nil {
|
||||
t.Fatalf("missing tool identity reached execution: %#v", end)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseSSERejectsNonObjectToolArguments(t *testing.T) {
|
||||
for _, tc := range []struct{ name, arguments string }{
|
||||
{"array", "[]"},
|
||||
{"null", "null"},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
body := strings.Join([]string{
|
||||
`data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"echo","arguments":"` + tc.arguments + `"}}]}}]}`,
|
||||
`data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":12,"completion_tokens":5,"total_tokens":17}}`,
|
||||
`data: [DONE]`,
|
||||
"",
|
||||
}, "\n\n")
|
||||
var end Event
|
||||
for ev := range ParseSSE(context.Background(), strings.NewReader(body)) {
|
||||
if ev.Kind == KindMessageEnd {
|
||||
end = ev
|
||||
}
|
||||
}
|
||||
if end.Message.StopReason != StopError || len(end.Message.ToolCalls) != 0 || end.Err == nil {
|
||||
t.Fatalf("non-object tool arguments reached execution: %#v", end)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -149,6 +260,66 @@ func TestParseSSEMalformedIsError(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseSSEMalformedChunkAfterUsageKeepsUsage(t *testing.T) {
|
||||
body := strings.Join([]string{
|
||||
`data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"echo","arguments":"{}"}}]},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":12,"completion_tokens":5,"total_tokens":17}}`,
|
||||
`data: {not-json}`,
|
||||
"",
|
||||
}, "\n\n")
|
||||
var end Event
|
||||
for ev := range ParseSSE(context.Background(), strings.NewReader(body)) {
|
||||
end = ev
|
||||
}
|
||||
if end.Kind != KindMessageEnd || end.Message.StopReason != StopError ||
|
||||
len(end.Message.ToolCalls) != 0 || end.Err == nil || end.Retryable ||
|
||||
end.Usage.TotalTokens != 17 {
|
||||
t.Fatalf("malformed chunk dropped already returned usage: %#v", end)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseSSEProviderErrorAfterUsageKeepsUsage(t *testing.T) {
|
||||
body := strings.Join([]string{
|
||||
`data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"echo","arguments":"{}"}}]}}],"usage":{"prompt_tokens":12,"completion_tokens":5,"total_tokens":17}}`,
|
||||
`data: {"error":{"message":"server is at capacity"}}`,
|
||||
"",
|
||||
}, "\n\n")
|
||||
var end Event
|
||||
for ev := range ParseSSE(context.Background(), strings.NewReader(body)) {
|
||||
end = ev
|
||||
}
|
||||
if end.Kind != KindMessageEnd || end.Message.StopReason != StopError ||
|
||||
len(end.Message.ToolCalls) != 0 || end.Err == nil || end.Retryable ||
|
||||
end.Usage.TotalTokens != 17 {
|
||||
t.Fatalf("provider error after usage retried or lost its charge: %#v", end)
|
||||
}
|
||||
}
|
||||
|
||||
type errorAfterDataReader struct{ data *strings.Reader }
|
||||
|
||||
func (r *errorAfterDataReader) Read(p []byte) (int, error) {
|
||||
n, err := r.data.Read(p)
|
||||
if err == io.EOF {
|
||||
return 0, io.ErrUnexpectedEOF
|
||||
}
|
||||
return n, err
|
||||
}
|
||||
|
||||
func TestParseSSEReaderErrorAfterUsageKeepsUsage(t *testing.T) {
|
||||
body := strings.Join([]string{
|
||||
`data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"echo","arguments":"{}"}}]}}],"usage":{"prompt_tokens":12,"completion_tokens":5,"total_tokens":17}}`,
|
||||
"",
|
||||
}, "\n\n")
|
||||
var end Event
|
||||
for ev := range ParseSSE(context.Background(), &errorAfterDataReader{data: strings.NewReader(body)}) {
|
||||
end = ev
|
||||
}
|
||||
if end.Kind != KindMessageEnd || end.Message.StopReason != StopError ||
|
||||
len(end.Message.ToolCalls) != 0 || end.Err == nil || end.Retryable ||
|
||||
end.Usage.TotalTokens != 17 {
|
||||
t.Fatalf("reader error after usage retried or lost its charge: %#v", end)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseSSECancel(t *testing.T) {
|
||||
pr, pw := io.Pipe()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
@@ -113,9 +113,12 @@ type Event struct {
|
||||
// FinishReason 是供应商原始结束原因(stop、length、tool_calls 等)。
|
||||
// 空表示这条事件不携带结束原因。StopReason 的映射不受此字段影响。
|
||||
FinishReason string
|
||||
Err error
|
||||
Retryable bool
|
||||
ErrorMessage string
|
||||
// StreamComplete records whether the OpenAI-compatible SSE stream supplied
|
||||
// its explicit [DONE] marker. Tool calls require this before execution.
|
||||
StreamComplete bool
|
||||
Err error
|
||||
Retryable bool
|
||||
ErrorMessage string
|
||||
// HTTPStatus 是提供方的 HTTP 状态。0 表示这条错误不是 HTTP 响应。
|
||||
HTTPStatus int
|
||||
// RetryAfter is the provider's requested wait. HasRetryAfter distinguishes
|
||||
|
||||
+53
-9
@@ -38,8 +38,10 @@ type Event struct {
|
||||
Message llm.Message
|
||||
Tool string
|
||||
Text string
|
||||
// ToolIndex preserves the provider's call slot for tool_call_delta.
|
||||
ToolIndex int
|
||||
// CallID 在 tool_execution_start / tool_execution_end 时携带工具调用的
|
||||
// ID:并行执行下同回合多个调用并发进出,消费方按 ID 配对而非按顺序。
|
||||
// ID;tool_call_delta 也携带同一 ID。消费方按 ID 配对而非按顺序。
|
||||
CallID string
|
||||
// Detail carries tool arguments/results or reasoning deltas for
|
||||
// process-visualization consumers; log writers may ignore it.
|
||||
@@ -69,8 +71,11 @@ type Config struct {
|
||||
Trace func(name string, attrs map[string]string) func()
|
||||
MaxTurns int
|
||||
ToolExecution string
|
||||
GetSteering func() []llm.Message
|
||||
GetFollowUp func() []llm.Message
|
||||
// SerialTools 是并行执行下的顺序敏感工具名集合:这些调用按模型给出的
|
||||
// 原顺序在调用方 goroutine 逐个执行(互斥且保序),其余调用并发执行。
|
||||
SerialTools map[string]bool
|
||||
GetSteering func() []llm.Message
|
||||
GetFollowUp func() []llm.Message
|
||||
// OnEvent 观察循环事件。同一次 Run 内不会并发调用
|
||||
// (与 BeforeTool、AfterTool 共用一把锁串行)。
|
||||
OnEvent func(Event)
|
||||
@@ -224,9 +229,9 @@ func Run(ctx context.Context, cfg Config, user llm.Message) (Result, error) {
|
||||
}
|
||||
cfg.traceCtx = newTraceContext()
|
||||
endRun := startSpan(cfg, "invoke_agent "+agentName(cfg), map[string]string{
|
||||
"gen_ai.system": providerName(cfg),
|
||||
"gen_ai.agent.name": agentName(cfg),
|
||||
"gen_ai.request.model": cfg.Model,
|
||||
"gen_ai.system": providerName(cfg),
|
||||
"gen_ai.agent.name": agentName(cfg),
|
||||
"gen_ai.request.model": cfg.Model,
|
||||
}, true)
|
||||
defer endRun()
|
||||
var messages []llm.Message
|
||||
@@ -351,6 +356,12 @@ func Run(ctx context.Context, cfg Config, user llm.Message) (Result, error) {
|
||||
usage.LastPromptTokens = end.Usage.PromptTokens
|
||||
usage.LastCompletionTokens = end.Usage.CompletionTokens
|
||||
usage.LastTotalTokens = end.Usage.TotalTokens
|
||||
if end.Err != nil {
|
||||
cfg.logger().Error("runtime terminal stream failed", end.Err)
|
||||
emitUsage(cfg, usage)
|
||||
return Result{Text: last.Content, Messages: messages, StopReason: llm.StopError,
|
||||
Usage: usage, Err: end.Err, Retryable: false}, nil
|
||||
}
|
||||
if err := persist(cfg, last); err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
@@ -526,7 +537,8 @@ func consume(ctx context.Context, cfg Config, messages []llm.Message) (llm.Event
|
||||
case llm.KindReasoningDelta:
|
||||
emit(cfg, Event{Type: "reasoning_delta", Text: ev.ReasoningDelta})
|
||||
case llm.KindToolCallDelta:
|
||||
emit(cfg, Event{Type: "tool_call_delta", Tool: ev.ToolName, Detail: ev.ArgumentsDelta})
|
||||
emit(cfg, Event{Type: "tool_call_delta", ToolIndex: ev.ToolIndex,
|
||||
CallID: ev.ToolID, Tool: ev.ToolName, Detail: ev.ArgumentsDelta})
|
||||
case llm.KindError:
|
||||
return ev, ev.Err
|
||||
case llm.KindMessageEnd:
|
||||
@@ -597,7 +609,10 @@ func serializeHostCallbacks(cfg *Config) {
|
||||
|
||||
func executeTools(ctx context.Context, cfg Config, assistant llm.Message) []llm.Message {
|
||||
if cfg.ToolExecution == ExecParallel && len(assistant.ToolCalls) > 1 {
|
||||
return executeParallel(ctx, cfg, assistant)
|
||||
if len(cfg.SerialTools) == 0 {
|
||||
return executeParallel(ctx, cfg, assistant)
|
||||
}
|
||||
return executeMixed(ctx, cfg, assistant)
|
||||
}
|
||||
out := make([]llm.Message, 0, len(assistant.ToolCalls))
|
||||
for _, call := range assistant.ToolCalls {
|
||||
@@ -610,6 +625,35 @@ func executeTools(ctx context.Context, cfg Config, assistant llm.Message) []llm.
|
||||
return out
|
||||
}
|
||||
|
||||
// executeMixed 并行执行非串行调用,串行名(SerialTools)按模型给出的原顺序
|
||||
// 在调用方 goroutine 逐个执行——互斥且保序(卡片流等有先后依赖的工具不能只
|
||||
// 靠互斥:并发派发下先抢到锁的不一定是先声明的调用)。
|
||||
func executeMixed(ctx context.Context, cfg Config, assistant llm.Message) []llm.Message {
|
||||
out := make([]llm.Message, len(assistant.ToolCalls))
|
||||
var wg sync.WaitGroup
|
||||
serialIdx := make([]int, 0, len(assistant.ToolCalls))
|
||||
for i, call := range assistant.ToolCalls {
|
||||
if cfg.SerialTools[call.Name] {
|
||||
serialIdx = append(serialIdx, i)
|
||||
continue
|
||||
}
|
||||
wg.Add(1)
|
||||
go func(i int, call llm.ToolCall) {
|
||||
defer wg.Done()
|
||||
out[i] = runOne(ctx, cfg, call)
|
||||
}(i, call)
|
||||
}
|
||||
for _, i := range serialIdx {
|
||||
if err := ctx.Err(); err != nil {
|
||||
out[i] = cancelledToolMessage(assistant.ToolCalls[i], err)
|
||||
continue
|
||||
}
|
||||
out[i] = runOne(ctx, cfg, assistant.ToolCalls[i])
|
||||
}
|
||||
wg.Wait()
|
||||
return out
|
||||
}
|
||||
|
||||
func executeParallel(ctx context.Context, cfg Config, assistant llm.Message) []llm.Message {
|
||||
out := make([]llm.Message, len(assistant.ToolCalls))
|
||||
var wg sync.WaitGroup
|
||||
@@ -1121,7 +1165,7 @@ func drain(fn func() []llm.Message) []llm.Message {
|
||||
|
||||
func emitUsage(cfg Config, usage llm.Usage) {
|
||||
emit(cfg, Event{
|
||||
Type: "usage",
|
||||
Type: "usage",
|
||||
PromptTokens: usage.PromptTokens, CompletionTokens: usage.CompletionTokens, TotalTokens: usage.TotalTokens,
|
||||
LastPromptTokens: usage.LastPromptTokens, LastCompletionTokens: usage.LastCompletionTokens, LastTotalTokens: usage.LastTotalTokens,
|
||||
CacheReadTokens: usage.CacheReadTokens, CacheWriteTokens: usage.CacheWriteTokens,
|
||||
|
||||
@@ -0,0 +1,72 @@
|
||||
package runtime
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"agent-native/runtime/llm"
|
||||
"agent-native/runtime/tools"
|
||||
)
|
||||
|
||||
// orderTool 记录自己的执行序号(并发安全)。
|
||||
type orderTool struct {
|
||||
name string
|
||||
mu sync.Mutex
|
||||
seq []string
|
||||
}
|
||||
|
||||
func (o *orderTool) Name() string { return o.name }
|
||||
func (o *orderTool) Description() string { return o.name }
|
||||
func (o *orderTool) Parameters() json.RawMessage { return json.RawMessage(`{"type":"object"}`) }
|
||||
func (o *orderTool) Execute(ctx context.Context, _ json.RawMessage) (tools.Result, error) {
|
||||
id := o.name
|
||||
if call, ok := ToolCallFromContext(ctx); ok {
|
||||
id = call.ID
|
||||
}
|
||||
o.mu.Lock()
|
||||
o.seq = append(o.seq, id)
|
||||
o.mu.Unlock()
|
||||
return tools.Result{Content: "ran " + id}, nil
|
||||
}
|
||||
|
||||
// TestLoopSerialToolsKeepOrder 并行模式下 SerialTools 命中的调用按声明顺序执行
|
||||
// (互斥不保序不够:卡片流等有先后依赖),未命中的并发执行。
|
||||
func TestLoopSerialToolsKeepOrder(t *testing.T) {
|
||||
cards := &orderTool{name: "card"}
|
||||
reads := &orderTool{name: "read"}
|
||||
reg := tools.NewRegistry(t.TempDir())
|
||||
reg.Register(cards)
|
||||
reg.Register(reads)
|
||||
res, err := Run(context.Background(), Config{
|
||||
Model: "fake", Tools: reg,
|
||||
ToolExecution: ExecParallel,
|
||||
SerialTools: map[string]bool{"card": true},
|
||||
Stream: script(
|
||||
llm.Message{Role: llm.RoleAssistant, StopReason: llm.StopToolCalls, ToolCalls: []llm.ToolCall{
|
||||
{ID: "s1", Name: "card", Arguments: `{}`},
|
||||
{ID: "p1", Name: "read", Arguments: `{}`},
|
||||
{ID: "s2", Name: "card", Arguments: `{}`},
|
||||
{ID: "s3", Name: "card", Arguments: `{}`},
|
||||
}},
|
||||
llm.Message{Role: llm.RoleAssistant, Content: "done", StopReason: llm.StopEnd},
|
||||
),
|
||||
}, llm.Message{Role: llm.RoleUser, Content: "go"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cards.mu.Lock()
|
||||
got := append([]string(nil), cards.seq...)
|
||||
cards.mu.Unlock()
|
||||
if len(got) != 3 || got[0] != "s1" || got[1] != "s2" || got[2] != "s3" {
|
||||
t.Fatalf("card executions must follow declaration order s1→s2→s3, got %v", got)
|
||||
}
|
||||
reads.mu.Lock()
|
||||
nRead := len(reads.seq)
|
||||
reads.mu.Unlock()
|
||||
if nRead != 1 {
|
||||
t.Fatalf("read executions = %d", nRead)
|
||||
}
|
||||
_ = res
|
||||
}
|
||||
@@ -2,6 +2,7 @@ package runtime
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
@@ -12,6 +13,54 @@ import (
|
||||
"agent-native/runtime/tools"
|
||||
)
|
||||
|
||||
func TestRunTerminalToolFailureKeepsUsage(t *testing.T) {
|
||||
hits, calls := 0, 0
|
||||
reg := tools.NewRegistry(t.TempDir())
|
||||
reg.Register(hitTool{hits: &hits})
|
||||
stream := func(context.Context, string, []llm.Message, []llm.ToolDecl, llm.StreamOptions) <-chan llm.Event {
|
||||
calls++
|
||||
out := make(chan llm.Event, 1)
|
||||
out <- llm.Event{
|
||||
Kind: llm.KindMessageEnd,
|
||||
Message: llm.Message{Role: llm.RoleAssistant, StopReason: llm.StopError},
|
||||
Usage: llm.Usage{PromptTokens: 12, CompletionTokens: 5, TotalTokens: 17},
|
||||
Err: errors.New("incomplete tool call"),
|
||||
}
|
||||
close(out)
|
||||
return out
|
||||
}
|
||||
res, err := Run(context.Background(), Config{Model: "fixture", Tools: reg, Stream: stream},
|
||||
llm.Message{Role: llm.RoleUser, Content: "organize materials"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if res.Err == nil || res.StopReason != llm.StopError || res.Usage.TotalTokens != 17 ||
|
||||
hits != 0 || calls != 1 {
|
||||
t.Fatalf("terminal failure retried, executed a tool, or lost usage: result=%#v hits=%d calls=%d", res, hits, calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunTerminalErrorCannotMasqueradeAsSuccess(t *testing.T) {
|
||||
stream := func(context.Context, string, []llm.Message, []llm.ToolDecl, llm.StreamOptions) <-chan llm.Event {
|
||||
out := make(chan llm.Event, 1)
|
||||
out <- llm.Event{Kind: llm.KindMessageEnd,
|
||||
Message: llm.Message{Role: llm.RoleAssistant, Content: "incomplete", StopReason: llm.StopEnd},
|
||||
Usage: llm.Usage{PromptTokens: 12, CompletionTokens: 5, TotalTokens: 17},
|
||||
Err: errors.New("terminal failed"),
|
||||
}
|
||||
close(out)
|
||||
return out
|
||||
}
|
||||
res, err := Run(context.Background(), Config{Model: "fixture", Stream: stream},
|
||||
llm.Message{Role: llm.RoleUser, Content: "go"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if res.Err == nil || res.StopReason != llm.StopError || res.Usage.TotalTokens != 17 {
|
||||
t.Fatalf("error-bearing terminal was accepted as success: %#v", res)
|
||||
}
|
||||
}
|
||||
|
||||
func script(turns ...llm.Message) llm.StreamFunc {
|
||||
var mu sync.Mutex
|
||||
i := 0
|
||||
|
||||
Reference in New Issue
Block a user