Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7f60baeaf8 | ||
|
|
ef2fa75006 | ||
|
|
a5da18b710 | ||
|
|
d6267fba15 | ||
|
|
1f25049492 | ||
|
|
76b6171de5 | ||
|
|
3b92cb63b4 | ||
|
|
37c5e6fbd6 |
@@ -284,7 +284,7 @@ func parseAnthropicSSE(ctx context.Context, r io.Reader, out chan<- Event) {
|
|||||||
}
|
}
|
||||||
case "content_block_start":
|
case "content_block_start":
|
||||||
var v struct {
|
var v struct {
|
||||||
Index int `json:"index"`
|
Index int `json:"index"`
|
||||||
ContentBlock struct {
|
ContentBlock struct {
|
||||||
Type string `json:"type"`
|
Type string `json:"type"`
|
||||||
ID string `json:"id"`
|
ID string `json:"id"`
|
||||||
|
|||||||
+29
-1
@@ -20,6 +20,12 @@ type Client struct {
|
|||||||
HTTPClient *http.Client
|
HTTPClient *http.Client
|
||||||
// Protocol 空 = openai-chat(零值兼容所有直接构造点)。
|
// Protocol 空 = openai-chat(零值兼容所有直接构造点)。
|
||||||
Protocol Protocol
|
Protocol Protocol
|
||||||
|
// ReasoningSplit 在 openai-chat 请求体加 "reasoning_split": true:
|
||||||
|
// MiniMax 等 interleaved-thinking 提供商据此把思考放进独立的
|
||||||
|
// reasoning_content 字段而不是混在 content 的 <think> 块里(分流后
|
||||||
|
// SSE 的 reasoning_content 捕获照常生效)。对不认识该字段的服务器
|
||||||
|
// 属未知字段,多数按忽略处理;宿主按模型名自行决定是否置位。
|
||||||
|
ReasoningSplit bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// Stream implements StreamFunc. Transport and provider failures are encoded as
|
// Stream implements StreamFunc. Transport and provider failures are encoded as
|
||||||
@@ -67,6 +73,9 @@ func (c *Client) run(ctx context.Context, model string, messages []Message, tool
|
|||||||
if effort := OpenAIReasoningEffort(opts.ThinkingLevel); effort != "" {
|
if effort := OpenAIReasoningEffort(opts.ThinkingLevel); effort != "" {
|
||||||
body["reasoning_effort"] = effort
|
body["reasoning_effort"] = effort
|
||||||
}
|
}
|
||||||
|
if c.ReasoningSplit {
|
||||||
|
body["reasoning_split"] = true
|
||||||
|
}
|
||||||
if len(tools) > 0 {
|
if len(tools) > 0 {
|
||||||
body["tools"] = toProviderTools(tools)
|
body["tools"] = toProviderTools(tools)
|
||||||
body["tool_choice"] = "auto"
|
body["tool_choice"] = "auto"
|
||||||
@@ -121,6 +130,18 @@ func (c *Client) run(ctx context.Context, model string, messages []Message, tool
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// replayThink 拼接一条 assistant 消息捕获到的全部思考块。
|
||||||
|
func replayThink(m Message) string {
|
||||||
|
var b strings.Builder
|
||||||
|
for _, t := range m.Thinking {
|
||||||
|
if t.Thinking != "" {
|
||||||
|
b.WriteString(t.Thinking)
|
||||||
|
b.WriteString("\n")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return strings.TrimRight(b.String(), "\n")
|
||||||
|
}
|
||||||
|
|
||||||
func toProviderMessages(messages []Message) []map[string]any {
|
func toProviderMessages(messages []Message) []map[string]any {
|
||||||
out := make([]map[string]any, 0, len(messages))
|
out := make([]map[string]any, 0, len(messages))
|
||||||
for _, m := range messages {
|
for _, m := range messages {
|
||||||
@@ -132,7 +153,14 @@ func toProviderMessages(messages []Message) []map[string]any {
|
|||||||
"content": m.Content,
|
"content": m.Content,
|
||||||
})
|
})
|
||||||
case RoleAssistant:
|
case RoleAssistant:
|
||||||
item := map[string]any{"role": "assistant", "content": m.Content}
|
// 回放捕获的思考(Message.Thinking 由 SSE 的 reasoning_content 捕获而来):
|
||||||
|
// 交错思考模型要求历史 assistant 原样带回 <think> 块。没有思考块的
|
||||||
|
// 消息零改动——非思考模型的上下文不受影响。
|
||||||
|
content := m.Content
|
||||||
|
if think := replayThink(m); think != "" {
|
||||||
|
content = "<think>\n" + think + "\n</think>\n\n" + content
|
||||||
|
}
|
||||||
|
item := map[string]any{"role": "assistant", "content": content}
|
||||||
if len(m.ToolCalls) > 0 {
|
if len(m.ToolCalls) > 0 {
|
||||||
calls := make([]map[string]any, 0, len(m.ToolCalls))
|
calls := make([]map[string]any, 0, len(m.ToolCalls))
|
||||||
for _, tc := range m.ToolCalls {
|
for _, tc := range m.ToolCalls {
|
||||||
|
|||||||
@@ -0,0 +1,47 @@
|
|||||||
|
package llm
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"io"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestOpenAIChatBodyReasoningSplit 验证 Client.ReasoningSplit 置位时
|
||||||
|
// openai-chat 请求体带 "reasoning_split": true,未置位时不带(未知字段
|
||||||
|
// 不外溢给别的提供商)。
|
||||||
|
func TestOpenAIChatBodyReasoningSplit(t *testing.T) {
|
||||||
|
var bodies []string
|
||||||
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
raw, _ := io.ReadAll(r.Body)
|
||||||
|
bodies = append(bodies, string(raw))
|
||||||
|
w.Header().Set("Content-Type", "text/event-stream")
|
||||||
|
_, _ = io.WriteString(w, "data: [DONE]\n\n")
|
||||||
|
}))
|
||||||
|
defer srv.Close()
|
||||||
|
|
||||||
|
drain := func(c *Client) {
|
||||||
|
for range c.Stream(context.Background(), "MiniMax-M3",
|
||||||
|
[]Message{{Role: RoleUser, Content: "hi"}}, nil, StreamOptions{}) {
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
drain(&Client{BaseURL: srv.URL, APIKey: "k"})
|
||||||
|
drain(&Client{BaseURL: srv.URL, APIKey: "k", ReasoningSplit: true})
|
||||||
|
if len(bodies) != 2 {
|
||||||
|
t.Fatalf("requests = %d", len(bodies))
|
||||||
|
}
|
||||||
|
if strings.Contains(bodies[0], "reasoning_split") {
|
||||||
|
t.Fatalf("flag off must not leak the field: %s", bodies[0])
|
||||||
|
}
|
||||||
|
var got map[string]any
|
||||||
|
if err := json.Unmarshal([]byte(bodies[1]), &got); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if v, ok := got["reasoning_split"].(bool); !ok || !v {
|
||||||
|
t.Fatalf("flag on must set reasoning_split=true: %s", bodies[1])
|
||||||
|
}
|
||||||
|
}
|
||||||
+9
-1
@@ -52,6 +52,7 @@ func ParseSSE(ctx context.Context, r io.Reader) <-chan Event {
|
|||||||
|
|
||||||
func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
|
func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
|
||||||
var content strings.Builder
|
var content strings.Builder
|
||||||
|
var reasoning strings.Builder
|
||||||
calls := map[int]*ToolCall{}
|
calls := map[int]*ToolCall{}
|
||||||
order := []int{}
|
order := []int{}
|
||||||
var usage Usage
|
var usage Usage
|
||||||
@@ -108,8 +109,12 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
|
|||||||
}
|
}
|
||||||
// reasoning_content(国产 OpenAI 兼容模型的思考链)与正文分流:
|
// reasoning_content(国产 OpenAI 兼容模型的思考链)与正文分流:
|
||||||
// 只作为 reasoning_delta 事件外发,不计入 Message.Content(与 runner
|
// 只作为 reasoning_delta 事件外发,不计入 Message.Content(与 runner
|
||||||
// stripThinkBlocks"正文无 think"语义对齐)。
|
// stripThinkBlocks"正文无 think"语义对齐)。同时累计进 reasoning——
|
||||||
|
// 交错思考模型(MiniMax/DeepSeek R1 等)要求历史 assistant 原样带回
|
||||||
|
// 思考内容,丢弃会降低后续轮质量;捕获进 Message.Thinking 随会话
|
||||||
|
// 转录持久化,请求序列化侧按需回放。
|
||||||
if choice.Delta.Reasoning != "" {
|
if choice.Delta.Reasoning != "" {
|
||||||
|
reasoning.WriteString(choice.Delta.Reasoning)
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
emitAbort(out, content.String(), ctx.Err())
|
emitAbort(out, content.String(), ctx.Err())
|
||||||
@@ -166,6 +171,9 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
|
|||||||
order = nil
|
order = nil
|
||||||
}
|
}
|
||||||
msg := Message{Role: RoleAssistant, Content: content.String(), StopReason: mapFinish(finished, len(order) > 0)}
|
msg := Message{Role: RoleAssistant, Content: content.String(), StopReason: mapFinish(finished, len(order) > 0)}
|
||||||
|
if reasoning.Len() > 0 {
|
||||||
|
msg.Thinking = []ThinkingBlock{{Thinking: reasoning.String()}}
|
||||||
|
}
|
||||||
for _, idx := range order {
|
for _, idx := range order {
|
||||||
msg.ToolCalls = append(msg.ToolCalls, *calls[idx])
|
msg.ToolCalls = append(msg.ToolCalls, *calls[idx])
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,45 @@
|
|||||||
|
package llm
|
||||||
|
|
||||||
|
import (
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
// 交错思考模型(MiniMax 等)要求历史 assistant 原样带回思考内容:
|
||||||
|
// SSE 把 reasoning_content 捕获进 Message.Thinking;序列化把它作为
|
||||||
|
// <think> 前缀回放。没有思考块的消息保持零改动。
|
||||||
|
func TestReasoningCaptureAndReplay(t *testing.T) {
|
||||||
|
// 回放侧。
|
||||||
|
msgs := []Message{
|
||||||
|
{Role: RoleUser, Content: "总结"},
|
||||||
|
{Role: RoleAssistant, Content: "结论。", Thinking: []ThinkingBlock{{Thinking: "先读材料。"}}},
|
||||||
|
{Role: RoleAssistant, Content: "没有思考的轮次。"},
|
||||||
|
}
|
||||||
|
out := toProviderMessages(msgs)
|
||||||
|
if got := out[1]["content"].(string); !strings.HasPrefix(got, "<think>\n先读材料。\n</think>") || !strings.HasSuffix(got, "结论。") {
|
||||||
|
t.Fatalf("assistant replay malformed: %q", got)
|
||||||
|
}
|
||||||
|
if got := out[2]["content"].(string); got != "没有思考的轮次。" {
|
||||||
|
t.Fatalf("non-thinking assistant must be untouched: %q", got)
|
||||||
|
}
|
||||||
|
if _, ok := out[0]["tool_calls"]; ok {
|
||||||
|
t.Fatal("user messages carry no tool_calls")
|
||||||
|
}
|
||||||
|
// 捕获侧:直接构造 SSE 流。
|
||||||
|
sse := "data: {\"choices\":[{\"delta\":{\"reasoning_content\":\"思考\"}}]}\n\n" +
|
||||||
|
"data: {\"choices\":[{\"delta\":{\"content\":\"正文\"}}]}\n\n" +
|
||||||
|
"data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n" +
|
||||||
|
"data: [DONE]\n\n"
|
||||||
|
var end Event
|
||||||
|
for ev := range ParseSSE(t.Context(), strings.NewReader(sse)) {
|
||||||
|
if ev.Kind == KindMessageEnd {
|
||||||
|
end = ev
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if end.Message.Content != "正文" {
|
||||||
|
t.Fatalf("content polluted by reasoning: %q", end.Message.Content)
|
||||||
|
}
|
||||||
|
if len(end.Message.Thinking) != 1 || end.Message.Thinking[0].Thinking != "思考" {
|
||||||
|
t.Fatalf("reasoning not captured: %+v", end.Message.Thinking)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -52,9 +52,9 @@ type Message struct {
|
|||||||
Images []Image `json:"images,omitempty"`
|
Images []Image `json:"images,omitempty"`
|
||||||
ToolCallID string `json:"tool_call_id,omitempty"`
|
ToolCallID string `json:"tool_call_id,omitempty"`
|
||||||
ToolCalls []ToolCall `json:"tool_calls,omitempty"`
|
ToolCalls []ToolCall `json:"tool_calls,omitempty"`
|
||||||
// Thinking 是原生 Anthropic 路径的思考块(含签名),随 assistant 消息
|
// Thinking 是思考块:Anthropic 原生路径含签名、在请求里原样往返;
|
||||||
// 在请求里原样往返(Anthropic 要求思考块在内容块最前且签名不失)。
|
// OpenAI 兼容路径由 SSE 的 reasoning_content 捕获(不含签名),随会话
|
||||||
// OpenAI 路径不写这个字段(reasoning 不落盘,见 sse.go 注释)。
|
// 转录持久化,请求序列化时按 <think> 块回放(见 client.go replayThink)。
|
||||||
Thinking []ThinkingBlock `json:"thinking,omitempty"`
|
Thinking []ThinkingBlock `json:"thinking,omitempty"`
|
||||||
StopReason StopReason `json:"stop_reason,omitempty"`
|
StopReason StopReason `json:"stop_reason,omitempty"`
|
||||||
IsError bool `json:"is_error,omitempty"`
|
IsError bool `json:"is_error,omitempty"`
|
||||||
|
|||||||
+63
-8
@@ -38,6 +38,9 @@ type Event struct {
|
|||||||
Message llm.Message
|
Message llm.Message
|
||||||
Tool string
|
Tool string
|
||||||
Text string
|
Text string
|
||||||
|
// CallID 在 tool_execution_start / tool_execution_end 时携带工具调用的
|
||||||
|
// ID:并行执行下同回合多个调用并发进出,消费方按 ID 配对而非按顺序。
|
||||||
|
CallID string
|
||||||
// Detail carries tool arguments/results or reasoning deltas for
|
// Detail carries tool arguments/results or reasoning deltas for
|
||||||
// process-visualization consumers; log writers may ignore it.
|
// process-visualization consumers; log writers may ignore it.
|
||||||
Detail string
|
Detail string
|
||||||
@@ -66,6 +69,9 @@ type Config struct {
|
|||||||
Trace func(name string, attrs map[string]string) func()
|
Trace func(name string, attrs map[string]string) func()
|
||||||
MaxTurns int
|
MaxTurns int
|
||||||
ToolExecution string
|
ToolExecution string
|
||||||
|
// SerialTools 是并行执行下的顺序敏感工具名集合:这些调用按模型给出的
|
||||||
|
// 原顺序在调用方 goroutine 逐个执行(互斥且保序),其余调用并发执行。
|
||||||
|
SerialTools map[string]bool
|
||||||
GetSteering func() []llm.Message
|
GetSteering func() []llm.Message
|
||||||
GetFollowUp func() []llm.Message
|
GetFollowUp func() []llm.Message
|
||||||
// OnEvent 观察循环事件。同一次 Run 内不会并发调用
|
// OnEvent 观察循环事件。同一次 Run 内不会并发调用
|
||||||
@@ -594,7 +600,10 @@ func serializeHostCallbacks(cfg *Config) {
|
|||||||
|
|
||||||
func executeTools(ctx context.Context, cfg Config, assistant llm.Message) []llm.Message {
|
func executeTools(ctx context.Context, cfg Config, assistant llm.Message) []llm.Message {
|
||||||
if cfg.ToolExecution == ExecParallel && len(assistant.ToolCalls) > 1 {
|
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))
|
out := make([]llm.Message, 0, len(assistant.ToolCalls))
|
||||||
for _, call := range assistant.ToolCalls {
|
for _, call := range assistant.ToolCalls {
|
||||||
@@ -607,6 +616,35 @@ func executeTools(ctx context.Context, cfg Config, assistant llm.Message) []llm.
|
|||||||
return out
|
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 {
|
func executeParallel(ctx context.Context, cfg Config, assistant llm.Message) []llm.Message {
|
||||||
out := make([]llm.Message, len(assistant.ToolCalls))
|
out := make([]llm.Message, len(assistant.ToolCalls))
|
||||||
var wg sync.WaitGroup
|
var wg sync.WaitGroup
|
||||||
@@ -706,22 +744,38 @@ func noteToolHook(cfg Config, call llm.ToolCall, msg llm.Message) {
|
|||||||
cfg.Hooks.Emit(hooks.Call{Event: ev, Tool: call.Name, Args: call.Arguments, Result: msg.Content, IsError: msg.IsError})
|
cfg.Hooks.Emit(hooks.Call{Event: ev, Tool: call.Name, Args: call.Arguments, Result: msg.Content, IsError: msg.IsError})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// toolCallKey 是把当前工具调用注入 Execute ctx 的私有键:并行执行下工具实现
|
||||||
|
// 需要知道自己服务于哪次调用(配对过程事件、按调用记账),经
|
||||||
|
// ToolCallFromContext 读取;旧宿主/直接构造的 ctx 返回 ok=false。
|
||||||
|
type toolCallKey struct{}
|
||||||
|
|
||||||
|
// WithToolCall 把 call 挂进 ctx(内核 runOne 在执行前注入)。
|
||||||
|
func WithToolCall(ctx context.Context, call llm.ToolCall) context.Context {
|
||||||
|
return context.WithValue(ctx, toolCallKey{}, call)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ToolCallFromContext 取出当前正在执行的调用;不存在时 ok=false。
|
||||||
|
func ToolCallFromContext(ctx context.Context) (llm.ToolCall, bool) {
|
||||||
|
call, ok := ctx.Value(toolCallKey{}).(llm.ToolCall)
|
||||||
|
return call, ok
|
||||||
|
}
|
||||||
|
|
||||||
func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message {
|
func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message {
|
||||||
if err := ctx.Err(); err != nil {
|
if err := ctx.Err(); err != nil {
|
||||||
return cancelledToolMessage(call, err)
|
return cancelledToolMessage(call, err)
|
||||||
}
|
}
|
||||||
if why := rejectIncompleteToolCall(call); why != "" {
|
if why := rejectIncompleteToolCall(call); why != "" {
|
||||||
msg := llm.Message{Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: why, IsError: true}
|
msg := llm.Message{Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: why, IsError: true}
|
||||||
emit(cfg, Event{Type: "tool_execution_start", Tool: call.Name, Detail: call.Arguments})
|
emit(cfg, Event{Type: "tool_execution_start", Tool: call.Name, CallID: call.ID, Detail: call.Arguments})
|
||||||
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: why})
|
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, CallID: call.ID, Text: why})
|
||||||
if cfg.AfterTool != nil {
|
if cfg.AfterTool != nil {
|
||||||
cfg.AfterTool(call, msg)
|
cfg.AfterTool(call, msg)
|
||||||
}
|
}
|
||||||
return msg
|
return msg
|
||||||
}
|
}
|
||||||
emit(cfg, Event{Type: "tool_execution_start", Tool: call.Name, Detail: call.Arguments})
|
emit(cfg, Event{Type: "tool_execution_start", Tool: call.Name, CallID: call.ID, Detail: call.Arguments})
|
||||||
if gated, msg, blocked := gateTool(ctx, cfg, call); blocked {
|
if gated, msg, blocked := gateTool(ctx, cfg, call); blocked {
|
||||||
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: msg.Content})
|
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, CallID: call.ID, Text: msg.Content})
|
||||||
noteToolHook(cfg, call, msg)
|
noteToolHook(cfg, call, msg)
|
||||||
if cfg.AfterTool != nil {
|
if cfg.AfterTool != nil {
|
||||||
cfg.AfterTool(call, msg)
|
cfg.AfterTool(call, msg)
|
||||||
@@ -733,7 +787,7 @@ func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message {
|
|||||||
if cfg.BeforeTool != nil {
|
if cfg.BeforeTool != nil {
|
||||||
if err := cfg.BeforeTool(call); err != nil {
|
if err := cfg.BeforeTool(call); err != nil {
|
||||||
msg := llm.Message{Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: err.Error(), IsError: true}
|
msg := llm.Message{Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: err.Error(), IsError: true}
|
||||||
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: msg.Content})
|
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, CallID: call.ID, Text: msg.Content})
|
||||||
noteToolHook(cfg, call, msg)
|
noteToolHook(cfg, call, msg)
|
||||||
if cfg.AfterTool != nil {
|
if cfg.AfterTool != nil {
|
||||||
cfg.AfterTool(call, msg)
|
cfg.AfterTool(call, msg)
|
||||||
@@ -744,7 +798,7 @@ func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message {
|
|||||||
emit(cfg, Event{Type: "tool_execution_update", Tool: call.Name, Detail: call.Arguments})
|
emit(cfg, Event{Type: "tool_execution_update", Tool: call.Name, Detail: call.Arguments})
|
||||||
if err := commitToolIntent(cfg, call); err != nil {
|
if err := commitToolIntent(cfg, call); err != nil {
|
||||||
msg := llm.Message{Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: err.Error(), IsError: true}
|
msg := llm.Message{Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: err.Error(), IsError: true}
|
||||||
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: msg.Content})
|
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, CallID: call.ID, Text: msg.Content})
|
||||||
return msg
|
return msg
|
||||||
}
|
}
|
||||||
endSpan := startSpan(cfg, "execute_tool "+call.Name, map[string]string{
|
endSpan := startSpan(cfg, "execute_tool "+call.Name, map[string]string{
|
||||||
@@ -754,6 +808,7 @@ func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message {
|
|||||||
defer endSpan()
|
defer endSpan()
|
||||||
res := tools.Result{Content: "no tools configured", IsError: true}
|
res := tools.Result{Content: "no tools configured", IsError: true}
|
||||||
known := cfg.Tools != nil && cfg.Tools.Has(call.Name)
|
known := cfg.Tools != nil && cfg.Tools.Has(call.Name)
|
||||||
|
ctx = WithToolCall(ctx, call)
|
||||||
switch {
|
switch {
|
||||||
case known:
|
case known:
|
||||||
res = cfg.Tools.Execute(ctx, call.Name, json.RawMessage(call.Arguments))
|
res = cfg.Tools.Execute(ctx, call.Name, json.RawMessage(call.Arguments))
|
||||||
@@ -769,7 +824,7 @@ func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message {
|
|||||||
Content: res.Content,
|
Content: res.Content,
|
||||||
IsError: res.IsError,
|
IsError: res.IsError,
|
||||||
}
|
}
|
||||||
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: res.Content})
|
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, CallID: call.ID, Text: res.Content})
|
||||||
noteToolHook(cfg, call, msg)
|
noteToolHook(cfg, call, msg)
|
||||||
if cfg.AfterTool != nil {
|
if cfg.AfterTool != nil {
|
||||||
cfg.AfterTool(call, msg)
|
cfg.AfterTool(call, msg)
|
||||||
|
|||||||
@@ -0,0 +1,69 @@
|
|||||||
|
package runtime
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"sync"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"agent-native/runtime/llm"
|
||||||
|
"agent-native/runtime/tools"
|
||||||
|
)
|
||||||
|
|
||||||
|
// ctxEchoTool 把自己从 ctx 读到的调用 ID 记下来(并发安全),验证内核
|
||||||
|
// 在 Execute 前注入了当前调用。
|
||||||
|
type ctxEchoTool struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
ids []string
|
||||||
|
bad int
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *ctxEchoTool) Name() string { return "echo" }
|
||||||
|
func (c *ctxEchoTool) Description() string { return "echo" }
|
||||||
|
func (c *ctxEchoTool) Parameters() json.RawMessage { return json.RawMessage(`{"type":"object"}`) }
|
||||||
|
func (c *ctxEchoTool) Execute(ctx context.Context, _ json.RawMessage) (tools.Result, error) {
|
||||||
|
call, ok := ToolCallFromContext(ctx)
|
||||||
|
c.mu.Lock()
|
||||||
|
defer c.mu.Unlock()
|
||||||
|
if !ok || call.ID == "" || call.Name != "echo" {
|
||||||
|
c.bad++
|
||||||
|
return tools.Result{Content: "call missing from context", IsError: true}, nil
|
||||||
|
}
|
||||||
|
c.ids = append(c.ids, call.ID)
|
||||||
|
return tools.Result{Content: "ok " + call.ID}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestLoopToolSeesOwnCallInContext 并行执行两个调用,各自读到自己的 ID。
|
||||||
|
func TestLoopToolSeesOwnCallInContext(t *testing.T) {
|
||||||
|
tool := &ctxEchoTool{}
|
||||||
|
reg := tools.NewRegistry(t.TempDir())
|
||||||
|
reg.Register(tool)
|
||||||
|
res, err := Run(context.Background(), Config{
|
||||||
|
Model: "fake",
|
||||||
|
Tools: reg,
|
||||||
|
ToolExecution: ExecParallel,
|
||||||
|
Stream: script(
|
||||||
|
llm.Message{Role: llm.RoleAssistant, StopReason: llm.StopToolCalls, ToolCalls: []llm.ToolCall{
|
||||||
|
{ID: "pa", Name: "echo", Arguments: `{}`},
|
||||||
|
{ID: "pb", Name: "echo", Arguments: `{}`},
|
||||||
|
}},
|
||||||
|
llm.Message{Role: llm.RoleAssistant, Content: "done", StopReason: llm.StopEnd},
|
||||||
|
),
|
||||||
|
}, llm.Message{Role: llm.RoleUser, Content: "go"})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
tool.mu.Lock()
|
||||||
|
defer tool.mu.Unlock()
|
||||||
|
if tool.bad != 0 || len(tool.ids) != 2 {
|
||||||
|
t.Fatalf("bad=%d ids=%v", tool.bad, tool.ids)
|
||||||
|
}
|
||||||
|
seen := map[string]bool{}
|
||||||
|
for _, id := range tool.ids {
|
||||||
|
seen[id] = true
|
||||||
|
}
|
||||||
|
if !seen["pa"] || !seen["pb"] {
|
||||||
|
t.Fatalf("ids = %v", tool.ids)
|
||||||
|
}
|
||||||
|
_ = res
|
||||||
|
}
|
||||||
@@ -0,0 +1,49 @@
|
|||||||
|
package runtime
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"agent-native/runtime/llm"
|
||||||
|
"agent-native/runtime/tools"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestLoopToolEventsCarryCallID 验证 tool_execution_start/end 事件带工具调用
|
||||||
|
// ID:并行执行下消费方按 ID 配对(回归锚点,防止字段被删后顺序假设回归)。
|
||||||
|
func TestLoopToolEventsCarryCallID(t *testing.T) {
|
||||||
|
hits := 0
|
||||||
|
reg := tools.NewRegistry(t.TempDir())
|
||||||
|
reg.Register(hitTool{hits: &hits})
|
||||||
|
var starts, ends []Event
|
||||||
|
res, err := Run(context.Background(), Config{
|
||||||
|
Model: "fake",
|
||||||
|
Tools: reg,
|
||||||
|
Stream: script(
|
||||||
|
llm.Message{Role: llm.RoleAssistant, StopReason: llm.StopToolCalls, ToolCalls: []llm.ToolCall{{
|
||||||
|
ID: "call-7", Name: "echo", Arguments: `{}`,
|
||||||
|
}}},
|
||||||
|
llm.Message{Role: llm.RoleAssistant, Content: "done", StopReason: llm.StopEnd},
|
||||||
|
),
|
||||||
|
OnEvent: func(ev Event) {
|
||||||
|
switch ev.Type {
|
||||||
|
case "tool_execution_start":
|
||||||
|
starts = append(starts, ev)
|
||||||
|
case "tool_execution_end":
|
||||||
|
ends = append(ends, ev)
|
||||||
|
}
|
||||||
|
},
|
||||||
|
}, llm.Message{Role: llm.RoleUser, Content: "go"})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if hits != 1 || len(starts) != 1 || len(ends) != 1 {
|
||||||
|
t.Fatalf("hits=%d starts=%d ends=%d", hits, len(starts), len(ends))
|
||||||
|
}
|
||||||
|
if starts[0].CallID != "call-7" || starts[0].Tool != "echo" {
|
||||||
|
t.Fatalf("start = %+v", starts[0])
|
||||||
|
}
|
||||||
|
if ends[0].CallID != "call-7" || ends[0].Tool != "echo" {
|
||||||
|
t.Fatalf("end = %+v", ends[0])
|
||||||
|
}
|
||||||
|
_ = res
|
||||||
|
}
|
||||||
@@ -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
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user