From 099c7c51fbdc0c609894cb4cbde6377bd8a30029 Mon Sep 17 00:00:00 2001 From: "shentong.martin" Date: Wed, 2 Sep 2026 19:33:08 +0800 Subject: [PATCH 1/2] feat(components): support multimodal function tool results Change-Id: I483e055f3a0a39a72ab8260c5b6740e4b3790a50 --- components/model/agenticark/README.md | 18 + components/model/agenticark/README.zh_CN.md | 18 + components/model/agenticark/convertor.go | 69 ++- components/model/agenticark/convertor_test.go | 28 +- .../model/agenticark/event_convertor.go | 6 +- components/model/agenticark/go.mod | 8 +- components/model/agenticark/go.sum | 18 +- components/model/agenticark/model.go | 30 +- components/model/agenticark/model_test.go | 19 +- components/model/agenticark/runtime_bridge.go | 180 ++++++++ .../model/agenticark/runtime_bridge_test.go | 412 ++++++++++++++++++ 11 files changed, 764 insertions(+), 42 deletions(-) create mode 100644 components/model/agenticark/runtime_bridge.go create mode 100644 components/model/agenticark/runtime_bridge_test.go diff --git a/components/model/agenticark/README.md b/components/model/agenticark/README.md index abf042967..874d31159 100644 --- a/components/model/agenticark/README.md +++ b/components/model/agenticark/README.md @@ -432,6 +432,24 @@ func main() { } ``` +#### Multimodal Function Tool Results + +Function tool results may contain text and images when the selected Ark endpoint supports vision input: + +```go +toolResultMsg := &schema.AgenticMessage{ + Role: schema.AgenticRoleTypeUser, + ContentBlocks: []*schema.ContentBlock{schema.NewContentBlock(&schema.FunctionToolResult{ + CallID: toolCall.CallID, + Name: toolCall.Name, + Content: []*schema.FunctionToolResultContentBlock{ + {Type: schema.FunctionToolResultContentBlockTypeText, Text: &schema.UserInputText{Text: "reference image"}}, + {Type: schema.FunctionToolResultContentBlockTypeImage, Image: &schema.UserInputImage{URL: "https://example.com/image.png", Detail: schema.ImageURLDetailHigh}}, + }, + })}, +} +``` + #### Server Tool Example diff --git a/components/model/agenticark/README.zh_CN.md b/components/model/agenticark/README.zh_CN.md index 232890e9e..a66ec25b9 100644 --- a/components/model/agenticark/README.zh_CN.md +++ b/components/model/agenticark/README.zh_CN.md @@ -431,6 +431,24 @@ func main() { } ``` +#### 多模态函数工具结果 + +当所选 Ark endpoint 支持视觉输入时，函数工具结果可以同时包含文本和图片： + +```go +toolResultMsg := &schema.AgenticMessage{ + Role: schema.AgenticRoleTypeUser, + ContentBlocks: []*schema.ContentBlock{schema.NewContentBlock(&schema.FunctionToolResult{ + CallID: toolCall.CallID, + Name: toolCall.Name, + Content: []*schema.FunctionToolResultContentBlock{ + {Type: schema.FunctionToolResultContentBlockTypeText, Text: &schema.UserInputText{Text: "参考图片"}}, + {Type: schema.FunctionToolResultContentBlockTypeImage, Image: &schema.UserInputImage{URL: "https://example.com/image.png", Detail: schema.ImageURLDetailHigh}}, + }, + })}, +} +``` + #### 服务器工具示例 diff --git a/components/model/agenticark/convertor.go b/components/model/agenticark/convertor.go index d47984c84..6b7c74c4b 100644 --- a/components/model/agenticark/convertor.go +++ b/components/model/agenticark/convertor.go @@ -22,11 +22,12 @@ import ( "sync" "github.com/bytedance/sonic" - "github.com/cloudwego/eino/schema" "github.com/eino-contrib/jsonschema" "github.com/volcengine/volcengine-go-sdk/service/arkruntime/model/responses" "golang.org/x/sync/errgroup" "google.golang.org/protobuf/types/known/structpb" + + "github.com/cloudwego/eino/schema" ) func toSystemRoleInputItems(msg *schema.AgenticMessage) (items []*responses.InputItem, err error) { @@ -447,6 +448,9 @@ func userInputTextToInputItem(role responses.MessageRole_Enum, block *schema.Use } func userInputImageToInputItem(role responses.MessageRole_Enum, block *schema.UserInputImage) (inputItem *responses.InputItem, err error) { + if isFileURL(block.URL) { + return nil, fmt.Errorf("file:// URLs are not supported for image input") + } imageURL, err := resolveURL(block.URL, block.Base64Data, block.MIMEType) if err != nil { return nil, err @@ -494,6 +498,9 @@ func toContentItemImageDetail(detail schema.ImageURLDetail) (*responses.ContentI } func userInputVideoToInputItem(role responses.MessageRole_Enum, block *schema.UserInputVideo) (inputItem *responses.InputItem, err error) { + if isFileURL(block.URL) { + return nil, fmt.Errorf("file:// URLs are not supported for video input") + } videoURL, err := resolveURL(block.URL, block.Base64Data, block.MIMEType) if err != nil { return nil, err @@ -556,6 +563,9 @@ func userInputAudioToInputItem(role responses.MessageRole_Enum, block *schema.Us } func userInputFileToInputItem(role responses.MessageRole_Enum, block *schema.UserInputFile) (inputItem *responses.InputItem, err error) { + if isFileURL(block.URL) { + return nil, fmt.Errorf("file:// URLs are not supported for file input") + } fileItem := &responses.ContentItemFile{ Type: responses.ContentItemType_input_file, Filename: &block.Name, @@ -608,18 +618,56 @@ func functionToolResultToInputItem(block *schema.FunctionToolResult) (item *resp } func functionToolResultContentToText(content []*schema.FunctionToolResultContentBlock) (string, error) { - if len(content) > 1 { - return "", fmt.Errorf("multiple function tool result content blocks are not supported, got %d", len(content)) + if len(content) == 0 { + return "", nil } + if len(content) == 1 && content[0] != nil && content[0].Type == schema.FunctionToolResultContentBlockTypeText { + if content[0].Text == nil { + return "", fmt.Errorf("function tool result text block is nil") + } + return escapeToolResultText(content[0].Text.Text), nil + } + + items := make([]map[string]any, 0, len(content)) for _, block := range content { + if block == nil { + return "", fmt.Errorf("function tool result content block is nil") + } switch block.Type { case schema.FunctionToolResultContentBlockTypeText: - return block.Text.Text, nil + if block.Text == nil { + return "", fmt.Errorf("function tool result text block is nil") + } + items = append(items, map[string]any{ + "type": "input_text", + "text": block.Text.Text, + }) + case schema.FunctionToolResultContentBlockTypeImage: + if block.Image == nil { + return "", fmt.Errorf("function tool result image block is nil") + } + imageURL, err := resolveURL(block.Image.URL, block.Image.Base64Data, block.Image.MIMEType) + if err != nil { + return "", fmt.Errorf("resolve function tool result image: %w", err) + } + item := map[string]any{ + "type": "input_image", + "image_url": imageURL, + } + if block.Image.Detail != "" { + item["detail"] = string(block.Image.Detail) + } + items = append(items, item) default: return "", fmt.Errorf("unsupported function tool result content block type: %s", block.Type) } } - return "", nil + + b, err := sonic.Marshal(items) + if err != nil { + return "", fmt.Errorf("marshal multimodal function tool result: %w", err) + } + return multimodalToolOutputPrefix + string(b), nil } func assistantGenTextToInputItem(block *schema.ContentBlock) (item *responses.InputItem, err error) { @@ -1878,6 +1926,17 @@ func resolveURL(url string, base64Data string, mimeType string) (real string, er return real, nil } +func isFileURL(raw string) bool { + return strings.HasPrefix(strings.ToLower(raw), "file://") +} + +func escapeToolResultText(text string) string { + if strings.HasPrefix(text, multimodalToolOutputPrefix) || strings.HasPrefix(text, escapedToolOutputPrefix) { + return escapedToolOutputPrefix + text + } + return text +} + func ensureDataURL(base64Data, mimeType string) (string, error) { if strings.HasPrefix(base64Data, "data:") { return "", fmt.Errorf("base64Data field must be a raw base64 string, but got a string with prefix 'data:'") diff --git a/components/model/agenticark/convertor_test.go b/components/model/agenticark/convertor_test.go index 3874d37d2..9e4cfbd03 100644 --- a/components/model/agenticark/convertor_test.go +++ b/components/model/agenticark/convertor_test.go @@ -21,11 +21,12 @@ import ( "testing" "github.com/bytedance/mockey" - "github.com/cloudwego/eino/schema" "github.com/eino-contrib/jsonschema" "github.com/stretchr/testify/assert" "github.com/volcengine/volcengine-go-sdk/service/arkruntime/model/responses" "google.golang.org/protobuf/types/known/structpb" + + "github.com/cloudwego/eino/schema" ) func TestToSystemRoleInputItems(t *testing.T) { @@ -69,10 +70,10 @@ func TestToAssistantRoleInputItems(t *testing.T) { setItemID(msg.ContentBlocks[1], "id-1") setItemStatus(msg.ContentBlocks[1], responses.ItemStatus_completed.String()) msg.ResponseMeta = &schema.AgenticResponseMeta{ - Extension: &ResponseMetaExtension{}, - } + Extension: &ResponseMetaExtension{}, + } - items, err := toAssistantRoleInputItems(msg) + items, err := toAssistantRoleInputItems(msg) assert.NoError(t, err) assert.Equal(t, 2, len(items)) assert.Equal(t, responses.MessageRole_assistant, items[0].GetInputMessage().Role) @@ -274,6 +275,25 @@ func TestFunctionToolResultToInputItem(t *testing.T) { assert.Equal(t, "r1", out.Output) } +func TestUserInputRejectsFileURL(t *testing.T) { + _, err := userInputImageToInputItem(responses.MessageRole_user, &schema.UserInputImage{ + URL: "file:///tmp/image.png", + Detail: schema.ImageURLDetailAuto, + }) + assert.ErrorContains(t, err, "file:// URLs are not supported") + + _, err = userInputVideoToInputItem(responses.MessageRole_user, &schema.UserInputVideo{ + URL: "file:///tmp/video.mp4", + }) + assert.ErrorContains(t, err, "file:// URLs are not supported") + + _, err = userInputFileToInputItem(responses.MessageRole_user, &schema.UserInputFile{ + URL: "file:///tmp/file.txt", + Name: "file.txt", + }) + assert.ErrorContains(t, err, "file:// URLs are not supported") +} + func TestAssistantGenTextToInputItem(t *testing.T) { block := schema.NewContentBlock(&schema.AssistantGenText{ Text: "answer"}, diff --git a/components/model/agenticark/event_convertor.go b/components/model/agenticark/event_convertor.go index e2d09d3ea..c9d761c31 100644 --- a/components/model/agenticark/event_convertor.go +++ b/components/model/agenticark/event_convertor.go @@ -21,13 +21,13 @@ import ( "fmt" "io" + "github.com/volcengine/volcengine-go-sdk/service/arkruntime/model/responses" + "github.com/cloudwego/eino/components/model" "github.com/cloudwego/eino/schema" - "github.com/volcengine/volcengine-go-sdk/service/arkruntime/model/responses" - "github.com/volcengine/volcengine-go-sdk/service/arkruntime/utils" ) -func receivedStreamResponse(streamReader *utils.ResponsesStreamReader, +func receivedStreamResponse(streamReader responseStreamReader, config *model.AgenticConfig, sw *schema.StreamWriter[*model.AgenticCallbackOutput]) { receiver := newStreamReceiver() diff --git a/components/model/agenticark/go.mod b/components/model/agenticark/go.mod index caef12545..59c89b381 100644 --- a/components/model/agenticark/go.mod +++ b/components/model/agenticark/go.mod @@ -8,7 +8,8 @@ require ( github.com/cloudwego/eino v0.9.1 github.com/eino-contrib/jsonschema v1.0.3 github.com/go-viper/mapstructure/v2 v2.5.0 - github.com/stretchr/testify v1.10.0 + github.com/stretchr/testify v1.11.1 + github.com/volcengine/ark-runtime-go v0.4.0 github.com/volcengine/volcengine-go-sdk v1.2.34 github.com/wk8/go-ordered-map/v2 v2.1.8 golang.org/x/sync v0.8.0 @@ -23,6 +24,8 @@ require ( github.com/cloudwego/base64x v0.1.6 // indirect github.com/davecgh/go-spew v1.1.1 // indirect github.com/dustin/go-humanize v1.0.1 // indirect + github.com/go-faster/errors v0.7.1 // indirect + github.com/go-faster/jx v1.2.0 // indirect github.com/google/uuid v1.6.0 // indirect github.com/goph/emperror v0.17.2 // indirect github.com/gopherjs/gopherjs v1.17.2 // indirect @@ -37,6 +40,7 @@ require ( github.com/pelletier/go-toml/v2 v2.0.9 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/segmentio/asm v1.2.1 // indirect github.com/sirupsen/logrus v1.9.3 // indirect github.com/slongfield/pyfmt v0.0.0-20220222012616-ea85ff4c361f // indirect github.com/smarty/assertions v1.15.0 // indirect @@ -47,6 +51,6 @@ require ( golang.org/x/arch v0.11.0 // indirect golang.org/x/exp v0.0.0-20230713183714-613f0c0eb8a1 // indirect golang.org/x/sys v0.29.0 // indirect - gopkg.in/yaml.v2 v2.2.8 // indirect + gopkg.in/yaml.v2 v2.4.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/components/model/agenticark/go.sum b/components/model/agenticark/go.sum index caa2e0cda..c9da6fc85 100644 --- a/components/model/agenticark/go.sum +++ b/components/model/agenticark/go.sum @@ -37,6 +37,10 @@ github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo= github.com/getsentry/raven-go v0.2.0/go.mod h1:KungGk8q33+aIAZUIVWZDr2OfAEBsO49PX4NzFV5kcQ= github.com/go-check/check v0.0.0-20180628173108-788fd7840127 h1:0gkP6mzaMqkmpcJYCFOLkIBwI7xFExG03bbkOkCvUPI= +github.com/go-faster/errors v0.7.1 h1:MkJTnDoEdi9pDabt1dpWf7AA8/BaSYZqibYyhZ20AYg= +github.com/go-faster/errors v0.7.1/go.mod h1:5ySTjWFiphBs07IKuiL69nxdfd5+fzh1u7FPGZP2quo= +github.com/go-faster/jx v1.2.0 h1:T2YHJPrFaYu21fJtUxC9GzmluKu8rVIFDwwGBKTDseI= +github.com/go-faster/jx v1.2.0/go.mod h1:UWLOVDmMG597a5tBFPLIWJdUxz5/2emOpfsj9Neg0PE= github.com/go-viper/mapstructure/v2 v2.5.0 h1:vM5IJoUAy3d7zRSVtIwQgBj7BiWtMPfmPEgAXnvj1Ro= github.com/go-viper/mapstructure/v2 v2.5.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= github.com/gofrs/uuid v3.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM= @@ -82,11 +86,11 @@ github.com/klauspost/cpuid/v2 v2.2.9 h1:66ze0taIn2H33fBvCkXuv9BmCwDfafmiIVpKV9kK github.com/klauspost/cpuid/v2 v2.2.9/go.mod h1:rqkxqrZ1EhYM9G+hXH7YdowN5R5RGN6NK4QwQ3WMXF8= github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= -github.com/kr/pretty v0.2.0 h1:s5hAObm+yFO5uHYt5dYjxi2rXrsnmRpJx4OYvIWUaQs= github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= +github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= -github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0= github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc= github.com/mattn/go-colorable v0.1.2 h1:/bC9yWikZXAL9uJdulbSfyVNIR3n3trXl+v8+1sx8mU= @@ -111,6 +115,8 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= github.com/rollbar/rollbar-go v1.0.2/go.mod h1:AcFs5f0I+c71bpHlXNNDbOWJiKwjFDtISeXco0L5PKQ= +github.com/segmentio/asm v1.2.1 h1:DTNbBqs57ioxAD4PrArqftgypG4/qNpXoJx8TVXxPR0= +github.com/segmentio/asm v1.2.1/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs= github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo= github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= @@ -132,10 +138,13 @@ github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= -github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI= github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08= +github.com/volcengine/ark-runtime-go v0.4.0 h1:yvGQiRr+18QKVfDAzzZlIVSUwlKvL9uP64/EWnWNkpg= +github.com/volcengine/ark-runtime-go v0.4.0/go.mod h1:qvi+Ax3sfmNph8DCY9kxQGXIra2Aui3iYqVoA9KP6fU= github.com/volcengine/volc-sdk-golang v1.0.23 h1:anOslb2Qp6ywnsbyq9jqR0ljuO63kg9PY+4OehIk5R8= github.com/volcengine/volc-sdk-golang v1.0.23/go.mod h1:AfG/PZRUkHJ9inETvbjNifTDgut25Wbkm2QoYBTbvyU= github.com/volcengine/volcengine-go-sdk v1.2.34 h1:oty2YY90UvH05RDbDn6348Y6oS7A5Tr6kIteAjVqFlY= @@ -209,8 +218,9 @@ gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMy gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw= gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= -gopkg.in/yaml.v2 v2.2.8 h1:obN1ZagJSUGI0Ek/LBmuj4SNLPfIny3KsKFopxRdj10= gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= +gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/components/model/agenticark/model.go b/components/model/agenticark/model.go index 79439c315..ed9da20c4 100644 --- a/components/model/agenticark/model.go +++ b/components/model/agenticark/model.go @@ -25,7 +25,7 @@ import ( "time" "github.com/bytedance/sonic" - "github.com/volcengine/volcengine-go-sdk/service/arkruntime" + runtimev2 "github.com/volcengine/ark-runtime-go/arkruntime" "github.com/volcengine/volcengine-go-sdk/service/arkruntime/model/contextmanagement" "github.com/volcengine/volcengine-go-sdk/service/arkruntime/model/responses" @@ -171,35 +171,35 @@ func New(_ context.Context, config *Config) (*Model, error) { } func buildClient(config *Config) (*Model, error) { - var opts []arkruntime.ConfigOption + var opts []runtimev2.ConfigOption if config.Region != "" { - opts = append(opts, arkruntime.WithRegion(config.Region)) + opts = append(opts, runtimev2.WithRegion(config.Region)) } if config.Timeout != nil { - opts = append(opts, arkruntime.WithTimeout(*config.Timeout)) + opts = append(opts, runtimev2.WithTimeout(*config.Timeout)) } if config.HTTPClient != nil { - opts = append(opts, arkruntime.WithHTTPClient(config.HTTPClient)) + opts = append(opts, runtimev2.WithHTTPClient(config.HTTPClient)) } if config.RetryTimes != nil { - opts = append(opts, arkruntime.WithRetryTimes(*config.RetryTimes)) + opts = append(opts, runtimev2.WithRetryTimes(*config.RetryTimes)) } if config.BaseURL != "" { - opts = append(opts, arkruntime.WithBaseUrl(config.BaseURL)) + opts = append(opts, runtimev2.WithBaseUrl(config.BaseURL)) } - var client *arkruntime.Client + var client *runtimev2.Client if len(config.APIKey) > 0 { - client = arkruntime.NewClientWithApiKey(config.APIKey, opts...) + client = runtimev2.NewClientWithApiKey(config.APIKey, opts...) } else if config.AccessKey != "" && config.SecretKey != "" { - client = arkruntime.NewClientWithAkSk(config.AccessKey, config.SecretKey, opts...) + client = runtimev2.NewClientWithAkSk(config.AccessKey, config.SecretKey, opts...) } else { return nil, fmt.Errorf("failed to create client: missing credentials (set 'APIKey' or both 'AccessKey' and 'SecretKey')") } cm := &Model{ - cli: client, + cli: &runtimeBridge{client: client}, model: config.Model, maxTokens: config.MaxTokens, temperature: config.Temperature, @@ -221,7 +221,7 @@ func buildClient(config *Config) (*Model, error) { } type Model struct { - cli *arkruntime.Client + cli *runtimeBridge rawFunctionTools []*schema.ToolInfo functionTools []*responses.ResponsesTool @@ -278,7 +278,7 @@ func (m *Model) Generate(ctx context.Context, input []*schema.AgenticMessage, op } }() - responseObject, err := m.cli.CreateResponses(ctx, responseReq, arkruntime.WithCustomHeaders(specOptions.customHeaders)) + responseObject, err := m.cli.CreateResponses(ctx, responseReq, specOptions.customHeaders) if err != nil { return nil, fmt.Errorf("failed to create responses: %w", err) } @@ -334,7 +334,7 @@ func (m *Model) Stream(ctx context.Context, input []*schema.AgenticMessage, opts } }() - responseStreamReader, err := m.cli.CreateResponsesStream(ctx, responseReq, arkruntime.WithCustomHeaders(specOptions.customHeaders)) + responseStreamReader, err := m.cli.CreateResponsesStream(ctx, responseReq, specOptions.customHeaders) if err != nil { return nil, fmt.Errorf("failed to create responses: %w", err) } @@ -475,7 +475,7 @@ func (m *Model) CreatePrefixCache(ctx context.Context, prefix []*schema.AgenticM return nil, fmt.Errorf("failed to populate tool choice: %w", err) } - responseObj, err := m.cli.CreateResponses(ctx, responseReq) + responseObj, err := m.cli.CreateResponses(ctx, responseReq, nil) if err != nil { return nil, err } diff --git a/components/model/agenticark/model_test.go b/components/model/agenticark/model_test.go index 9858690f3..90225afc6 100644 --- a/components/model/agenticark/model_test.go +++ b/components/model/agenticark/model_test.go @@ -25,11 +25,12 @@ import ( "github.com/bytedance/mockey" "github.com/bytedance/sonic" - "github.com/cloudwego/eino/components/model" - "github.com/cloudwego/eino/schema" "github.com/stretchr/testify/assert" "github.com/volcengine/volcengine-go-sdk/service/arkruntime" "github.com/volcengine/volcengine-go-sdk/service/arkruntime/model/responses" + + "github.com/cloudwego/eino/components/model" + "github.com/cloudwego/eino/schema" ) func TestNew(t *testing.T) { @@ -146,7 +147,7 @@ func TestModelGenerate(t *testing.T) { mockey.PatchConvey("TestModelGenerate", t, func() { ctx := context.Background() m := &Model{ - cli: &arkruntime.Client{}, + cli: &runtimeBridge{}, model: "m", } input := []*schema.AgenticMessage{ @@ -158,7 +159,7 @@ func TestModelGenerate(t *testing.T) { }, } - mockey.Mock((*arkruntime.Client).CreateResponses).Return(&responses.ResponseObject{ + mockey.Mock((*runtimeBridge).CreateResponses).Return(&responses.ResponseObject{ Id: "rid", Output: []*responses.OutputItem{ { @@ -190,7 +191,7 @@ func TestModelGenerate(t *testing.T) { }) mockey.PatchConvey("error", func() { - mockey.Mock((*arkruntime.Client).CreateResponses).Return(nil, errors.New("err")).Build() + mockey.Mock((*runtimeBridge).CreateResponses).Return(nil, errors.New("err")).Build() _, err := m.Generate(ctx, input) assert.Error(t, err) }) @@ -201,7 +202,7 @@ func TestModelStream(t *testing.T) { mockey.PatchConvey("TestModelStream", t, func() { ctx := context.Background() m := &Model{ - cli: &arkruntime.Client{}, + cli: &runtimeBridge{}, model: "m", } input := []*schema.AgenticMessage{ @@ -209,7 +210,7 @@ func TestModelStream(t *testing.T) { } mockey.PatchConvey("error creating stream", func() { - mockey.Mock((*arkruntime.Client).CreateResponsesStream).Return(nil, errors.New("err")).Build() + mockey.Mock((*runtimeBridge).CreateResponsesStream).Return(nil, errors.New("err")).Build() _, err := m.Stream(ctx, input) assert.Error(t, err) }) @@ -220,14 +221,14 @@ func TestModelCreatePrefixCache(t *testing.T) { mockey.PatchConvey("TestModelCreatePrefixCache", t, func() { ctx := context.Background() m := &Model{ - cli: &arkruntime.Client{}, + cli: &runtimeBridge{}, model: "m", } prefix := []*schema.AgenticMessage{ {Role: schema.AgenticRoleTypeUser}, } - mockey.Mock((*arkruntime.Client).CreateResponses).Return(&responses.ResponseObject{ + mockey.Mock((*runtimeBridge).CreateResponses).Return(&responses.ResponseObject{ Id: "rid", Usage: &responses.Usage{ InputTokens: 10, diff --git a/components/model/agenticark/runtime_bridge.go b/components/model/agenticark/runtime_bridge.go new file mode 100644 index 000000000..8da194854 --- /dev/null +++ b/components/model/agenticark/runtime_bridge.go @@ -0,0 +1,180 @@ +/* + * Copyright 2026 CloudWeGo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package agenticark + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + runtimev2 "github.com/volcengine/ark-runtime-go/arkruntime" + responsesv2 "github.com/volcengine/ark-runtime-go/arkruntime/model/responses" + legacyresponses "github.com/volcengine/volcengine-go-sdk/service/arkruntime/model/responses" +) + +// The legacy SDK only accepts string tool outputs. These private prefixes carry +// the new SDK's array form through the legacy request model without expanding +// agenticark's public API. +const ( + multimodalToolOutputPrefix = "\x00eino_ext_agenticark_multimodal_tool_output:" + escapedToolOutputPrefix = "\x00eino_ext_agenticark_escaped_tool_output:" +) + +type responseStreamReader interface { + Recv() (*legacyresponses.Event, error) + Close() error +} + +type runtimeBridge struct { + client *runtimev2.Client +} + +func (c *runtimeBridge) CreateResponses(ctx context.Context, req *legacyresponses.ResponsesRequest, + headers map[string]string) (*legacyresponses.ResponseObject, error) { + v2Req, err := toV2Request(req) + if err != nil { + return nil, err + } + + res, err := c.client.CreateResponses(ctx, v2Req, runtimev2.WithCustomHeaders(headers)) + if err != nil { + return nil, err + } + + var legacy legacyresponses.ResponseObject + if err := convertJSON(&res.Response, &legacy); err != nil { + return nil, fmt.Errorf("convert responses response to legacy model: %w", err) + } + return &legacy, nil +} + +func (c *runtimeBridge) CreateResponsesStream(ctx context.Context, req *legacyresponses.ResponsesRequest, + headers map[string]string) (responseStreamReader, error) { + v2Req, err := toV2Request(req) + if err != nil { + return nil, err + } + + stream, err := c.client.CreateResponsesStream(ctx, v2Req, runtimev2.WithCustomHeaders(headers)) + if err != nil { + return nil, err + } + return &streamBridge{stream: stream}, nil +} + +type streamBridge struct { + stream interface { + Recv() (*responsesv2.ResponseStreamEvent, error) + Close() error + } +} + +func (s *streamBridge) Recv() (*legacyresponses.Event, error) { + event, err := s.stream.Recv() + if err != nil { + return nil, err + } + + var legacy legacyresponses.Event + if err := convertJSON(event, &legacy); err != nil { + return nil, fmt.Errorf("convert responses stream event to legacy model: %w", err) + } + return &legacy, nil +} + +func (s *streamBridge) Close() error { + return s.stream.Close() +} + +func toV2Request(req *legacyresponses.ResponsesRequest) (*responsesv2.ResponsesRequest, error) { + if req == nil { + return nil, fmt.Errorf("legacy responses request is nil") + } + raw, err := json.Marshal(req) + if err != nil { + return nil, fmt.Errorf("marshal legacy responses request: %w", err) + } + raw, err = rewriteMultimodalToolOutput(raw) + if err != nil { + return nil, fmt.Errorf("rewrite multimodal function tool output: %w", err) + } + + var v2Req responsesv2.ResponsesRequest + if err := json.Unmarshal(raw, &v2Req); err != nil { + return nil, fmt.Errorf("unmarshal request into ark-runtime-go model: %w", err) + } + return &v2Req, nil +} + +func rewriteMultimodalToolOutput(raw json.RawMessage) (json.RawMessage, error) { + var object map[string]json.RawMessage + if err := json.Unmarshal(raw, &object); err == nil { + if output, ok := object["output"]; ok { + var value string + if err := json.Unmarshal(output, &value); err == nil { + switch { + case strings.HasPrefix(value, escapedToolOutputPrefix): + unescaped, err := json.Marshal(strings.TrimPrefix(value, escapedToolOutputPrefix)) + if err != nil { + return nil, err + } + object["output"] = unescaped + case strings.HasPrefix(value, multimodalToolOutputPrefix): + content := json.RawMessage(strings.TrimPrefix(value, multimodalToolOutputPrefix)) + var blocks []json.RawMessage + if err := json.Unmarshal(content, &blocks); err != nil { + return nil, fmt.Errorf("invalid multimodal tool output: %w", err) + } + object["output"] = content + } + } + } + for key, child := range object { + if key == "output" { + continue + } + rewritten, err := rewriteMultimodalToolOutput(child) + if err != nil { + return nil, err + } + object[key] = rewritten + } + return json.Marshal(object) + } + + var list []json.RawMessage + if err := json.Unmarshal(raw, &list); err == nil { + for i, child := range list { + rewritten, err := rewriteMultimodalToolOutput(child) + if err != nil { + return nil, err + } + list[i] = rewritten + } + return json.Marshal(list) + } + return raw, nil +} + +func convertJSON(src, dst any) error { + b, err := json.Marshal(src) + if err != nil { + return err + } + return json.Unmarshal(b, dst) +} diff --git a/components/model/agenticark/runtime_bridge_test.go b/components/model/agenticark/runtime_bridge_test.go new file mode 100644 index 000000000..c247d6ef4 --- /dev/null +++ b/components/model/agenticark/runtime_bridge_test.go @@ -0,0 +1,412 @@ +/* + * Copyright 2026 CloudWeGo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package agenticark + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + runtimev2 "github.com/volcengine/ark-runtime-go/arkruntime" + responsesv2 "github.com/volcengine/ark-runtime-go/arkruntime/model/responses" + legacyresponses "github.com/volcengine/volcengine-go-sdk/service/arkruntime/model/responses" + + "github.com/cloudwego/eino/schema" +) + +func TestFunctionToolResultContentToText_Multimodal(t *testing.T) { + output, err := functionToolResultContentToText([]*schema.FunctionToolResultContentBlock{ + {Type: schema.FunctionToolResultContentBlockTypeText, Text: &schema.UserInputText{Text: "caption"}}, + {Type: schema.FunctionToolResultContentBlockTypeImage, Image: &schema.UserInputImage{ + URL: "https://example.com/image.png", + Detail: schema.ImageURLDetailHigh, + MIMEType: "image/png", + }}, + }) + require.NoError(t, err) + require.Contains(t, output, multimodalToolOutputPrefix) + + var content []map[string]any + require.NoError(t, json.Unmarshal([]byte(output[len(multimodalToolOutputPrefix):]), &content)) + require.Len(t, content, 2) + assert.Equal(t, "input_text", content[0]["type"]) + assert.Equal(t, "caption", content[0]["text"]) + assert.Equal(t, "input_image", content[1]["type"]) + assert.Equal(t, "https://example.com/image.png", content[1]["image_url"]) + assert.Equal(t, string(schema.ImageURLDetailHigh), content[1]["detail"]) +} + +func TestFunctionToolResultContentToText_SingleText(t *testing.T) { + output, err := functionToolResultContentToText([]*schema.FunctionToolResultContentBlock{ + {Type: schema.FunctionToolResultContentBlockTypeText, Text: &schema.UserInputText{Text: "plain"}}, + }) + require.NoError(t, err) + assert.Equal(t, "plain", output) +} + +func TestFunctionToolResultContentToText_Validation(t *testing.T) { + output, err := functionToolResultContentToText(nil) + require.NoError(t, err) + assert.Empty(t, output) + + _, err = functionToolResultContentToText([]*schema.FunctionToolResultContentBlock{nil}) + assert.ErrorContains(t, err, "content block is nil") + + _, err = functionToolResultContentToText([]*schema.FunctionToolResultContentBlock{{ + Type: schema.FunctionToolResultContentBlockTypeText, + }}) + assert.ErrorContains(t, err, "text block is nil") + + _, err = functionToolResultContentToText([]*schema.FunctionToolResultContentBlock{{ + Type: schema.FunctionToolResultContentBlockTypeImage, + }}) + assert.ErrorContains(t, err, "image block is nil") + + _, err = functionToolResultContentToText([]*schema.FunctionToolResultContentBlock{{ + Type: schema.FunctionToolResultContentBlockTypeImage, + Image: &schema.UserInputImage{Base64Data: "aGVsbG8="}, + }}) + assert.ErrorContains(t, err, "mimeType is required") + + _, err = functionToolResultContentToText([]*schema.FunctionToolResultContentBlock{{ + Type: schema.FunctionToolResultContentBlockTypeFile, + }}) + assert.ErrorContains(t, err, "unsupported function tool result content block type") +} + +func TestToV2Request_ReplacesMultimodalToolOutput(t *testing.T) { + item, err := functionToolResultToInputItem(&schema.FunctionToolResult{ + CallID: "call-1", + Content: []*schema.FunctionToolResultContentBlock{ + {Type: schema.FunctionToolResultContentBlockTypeImage, Image: &schema.UserInputImage{ + Base64Data: "aGVsbG8=", + MIMEType: "image/png", + }}, + }, + }) + require.NoError(t, err) + + req := &legacyresponses.ResponsesRequest{ + Model: "model", + Input: &legacyresponses.ResponsesInput{ + Union: &legacyresponses.ResponsesInput_ListValue{ + ListValue: &legacyresponses.InputItemList{ListValue: []*legacyresponses.InputItem{item}}, + }, + }, + } + v2Req, err := toV2Request(req) + require.NoError(t, err) + + b, err := json.Marshal(v2Req) + require.NoError(t, err) + var payload map[string]any + require.NoError(t, json.Unmarshal(b, &payload)) + input := payload["input"].([]any) + output := input[0].(map[string]any)["output"].([]any) + image := output[0].(map[string]any) + assert.Equal(t, "input_image", image["type"]) + assert.Equal(t, "data:image/png;base64,aGVsbG8=", image["image_url"]) +} + +func TestRuntimeBridgeSendsMultimodalToolOutput(t *testing.T) { + requests := make(chan map[string]any, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]any + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + requests <- body + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"error":{"message":"expected test failure"}}`)) + })) + defer server.Close() + + item, err := functionToolResultToInputItem(&schema.FunctionToolResult{ + CallID: "call-1", + Content: []*schema.FunctionToolResultContentBlock{ + {Type: schema.FunctionToolResultContentBlockTypeText, Text: &schema.UserInputText{Text: "caption"}}, + {Type: schema.FunctionToolResultContentBlockTypeImage, Image: &schema.UserInputImage{ + URL: "https://example.com/image.png", + }}, + }, + }) + require.NoError(t, err) + req := &legacyresponses.ResponsesRequest{ + Model: "model", + Input: &legacyresponses.ResponsesInput{ + Union: &legacyresponses.ResponsesInput_ListValue{ + ListValue: &legacyresponses.InputItemList{ListValue: []*legacyresponses.InputItem{item}}, + }, + }, + } + bridge := &runtimeBridge{client: runtimev2.NewClientWithApiKey("test", runtimev2.WithBaseUrl(server.URL))} + + _, err = bridge.CreateResponses(context.Background(), req, nil) + require.Error(t, err) + + select { + case body := <-requests: + input := body["input"].([]any) + output := input[0].(map[string]any)["output"].([]any) + assert.Equal(t, "input_text", output[0].(map[string]any)["type"]) + assert.Equal(t, "input_image", output[1].(map[string]any)["type"]) + case <-time.After(time.Second): + t.Fatal("bridge did not send a request") + } +} + +func TestRuntimeBridgeConvertsResponse(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{ + "id":"resp-1", + "object":"response", + "created_at":0, + "model":"model", + "output":[{ + "id":"msg-1", + "type":"message", + "role":"assistant", + "status":"completed", + "content":[{"type":"output_text","text":"hello","annotations":[]}] + }], + "status":"completed", + "usage":{ + "input_tokens":1, + "output_tokens":1, + "total_tokens":2, + "input_tokens_details":{"cached_tokens":0}, + "output_tokens_details":{"reasoning_tokens":0} + } + }`)) + })) + defer server.Close() + + bridge := &runtimeBridge{client: runtimev2.NewClientWithApiKey("test", runtimev2.WithBaseUrl(server.URL))} + res, err := bridge.CreateResponses(context.Background(), &legacyresponses.ResponsesRequest{ + Model: "model", + Input: &legacyresponses.ResponsesInput{ + Union: &legacyresponses.ResponsesInput_StringValue{StringValue: "hello"}, + }, + }, nil) + require.NoError(t, err) + require.NotNil(t, res) + assert.Equal(t, "resp-1", res.Id) + require.Len(t, res.Output, 1) + assert.Equal(t, "hello", res.Output[0].GetOutputMessage().GetContent()[0].GetText().GetText()) +} + +func TestRuntimeBridgeCreateResponsesStreamReturnsTransportError(t *testing.T) { + requests := make(chan map[string]any, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]any + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + requests <- body + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"error":{"message":"expected test failure"}}`)) + })) + defer server.Close() + + bridge := &runtimeBridge{client: runtimev2.NewClientWithApiKey("test", runtimev2.WithBaseUrl(server.URL))} + stream, err := bridge.CreateResponsesStream(context.Background(), &legacyresponses.ResponsesRequest{ + Model: "model", + Input: &legacyresponses.ResponsesInput{ + Union: &legacyresponses.ResponsesInput_StringValue{StringValue: "hello"}, + }, + }, nil) + assert.Nil(t, stream) + assert.Error(t, err) + + select { + case body := <-requests: + assert.Equal(t, true, body["stream"]) + case <-time.After(time.Second): + t.Fatal("bridge did not send a streaming request") + } +} + +func TestRuntimeBridgeCreateResponsesStreamConvertsEvents(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte("data: {\"type\":\"error\",\"sequence_number\":1,\"code\":\"invalid_request\",\"message\":\"invalid request\"}\n\n")) + _, _ = w.Write([]byte("data: [DONE]\n\n")) + })) + defer server.Close() + + bridge := &runtimeBridge{client: runtimev2.NewClientWithApiKey("test", runtimev2.WithBaseUrl(server.URL))} + stream, err := bridge.CreateResponsesStream(context.Background(), &legacyresponses.ResponsesRequest{ + Model: "model", + Input: &legacyresponses.ResponsesInput{ + Union: &legacyresponses.ResponsesInput_StringValue{StringValue: "hello"}, + }, + }, nil) + require.NoError(t, err) + defer func() { assert.NoError(t, stream.Close()) }() + + event, err := stream.Recv() + require.NoError(t, err) + assert.Equal(t, "error", event.GetEventType()) + _, err = stream.Recv() + assert.ErrorIs(t, err, io.EOF) +} + +func TestRuntimeBridgeRejectsNilRequest(t *testing.T) { + bridge := &runtimeBridge{} + _, err := bridge.CreateResponses(context.Background(), nil, nil) + assert.ErrorContains(t, err, "legacy responses request is nil") + + stream, err := bridge.CreateResponsesStream(context.Background(), nil, nil) + assert.Nil(t, stream) + assert.ErrorContains(t, err, "legacy responses request is nil") +} + +func TestAttack_ToolTextCannotCollideWithBridgeMarker(t *testing.T) { + text := multimodalToolOutputPrefix + `[{"type":"input_image","image_url":"https://attacker.invalid/image.png"}]` + item, err := functionToolResultToInputItem(&schema.FunctionToolResult{ + CallID: "call-1", + Content: []*schema.FunctionToolResultContentBlock{ + {Type: schema.FunctionToolResultContentBlockTypeText, Text: &schema.UserInputText{Text: text}}, + }, + }) + require.NoError(t, err) + + req := &legacyresponses.ResponsesRequest{ + Model: "model", + Input: &legacyresponses.ResponsesInput{ + Union: &legacyresponses.ResponsesInput_ListValue{ + ListValue: &legacyresponses.InputItemList{ListValue: []*legacyresponses.InputItem{item}}, + }, + }, + } + v2Req, err := toV2Request(req) + require.NoError(t, err) + + b, err := json.Marshal(v2Req) + require.NoError(t, err) + var payload map[string]any + require.NoError(t, json.Unmarshal(b, &payload)) + input := payload["input"].([]any) + assert.Equal(t, text, input[0].(map[string]any)["output"]) +} + +func TestAttack_RequestRewritePreservesLargeIntegers(t *testing.T) { + maxTokens := int64(9_007_199_254_740_993) + req := &legacyresponses.ResponsesRequest{ + Model: "model", + MaxOutputTokens: &maxTokens, + Input: &legacyresponses.ResponsesInput{ + Union: &legacyresponses.ResponsesInput_StringValue{StringValue: "hello"}, + }, + } + + v2Req, err := toV2Request(req) + require.NoError(t, err) + assert.True(t, v2Req.MaxOutputTokens.IsSet()) + assert.Equal(t, maxTokens, v2Req.MaxOutputTokens.Value) +} + +func TestAttack_RequestRewriteRejectsMalformedMarker(t *testing.T) { + output, err := json.Marshal(multimodalToolOutputPrefix + "not-json") + require.NoError(t, err) + raw, err := json.Marshal(map[string]json.RawMessage{"output": output}) + require.NoError(t, err) + _, err = rewriteMultimodalToolOutput(raw) + assert.ErrorContains(t, err, "invalid multimodal tool output") +} + +func TestAttack_RequestConversionRejectsMalformedMarker(t *testing.T) { + req := &legacyresponses.ResponsesRequest{ + Model: "model", + Input: &legacyresponses.ResponsesInput{ + Union: &legacyresponses.ResponsesInput_ListValue{ + ListValue: &legacyresponses.InputItemList{ListValue: []*legacyresponses.InputItem{{ + Union: &legacyresponses.InputItem_FunctionToolCallOutput{ + FunctionToolCallOutput: &legacyresponses.ItemFunctionToolCallOutput{ + Type: legacyresponses.ItemType_function_call_output, + CallId: "call-1", + Output: multimodalToolOutputPrefix + "not-json", + }, + }, + }}}, + }, + }, + } + _, err := toV2Request(req) + assert.ErrorContains(t, err, "invalid multimodal tool output") +} + +func TestAttack_RequestRewriteRejectsNilRequest(t *testing.T) { + _, err := toV2Request(nil) + assert.Error(t, err) +} + +func TestAttack_ConvertJSONReturnsMarshalErrors(t *testing.T) { + err := convertJSON(make(chan int), &map[string]any{}) + assert.Error(t, err) +} + +func TestAttack_StreamBridgeConvertsAndCloses(t *testing.T) { + var event responsesv2.ResponseStreamEvent + require.NoError(t, json.Unmarshal([]byte(`{ + "type":"error", + "sequence_number":1, + "code":"invalid_request", + "message":"invalid request" + }`), &event)) + stream := &fakeResponseStream{event: &event} + bridge := &streamBridge{stream: stream} + + legacy, err := bridge.Recv() + require.NoError(t, err) + assert.Equal(t, "error", legacy.GetEventType()) + assert.NoError(t, bridge.Close()) + assert.True(t, stream.closed) + + _, err = bridge.Recv() + assert.ErrorIs(t, err, io.EOF) +} + +type fakeResponseStream struct { + event *responsesv2.ResponseStreamEvent + closed bool +} + +func (s *fakeResponseStream) Recv() (*responsesv2.ResponseStreamEvent, error) { + if s.event == nil { + return nil, io.EOF + } + event := s.event + s.event = nil + return event, nil +} + +func (s *fakeResponseStream) Close() error { + s.closed = true + return nil +} From 669c7f25a1f2d97d3d16396b31955cea7c4e6ad9 Mon Sep 17 00:00:00 2001 From: "shentong.martin" Date: Thu, 3 Sep 2026 10:29:37 +0800 Subject: [PATCH 2/2] fix(components): preserve skipped thinking summaries Change-Id: I7331e2f23005464df7ab5ee5dbc17297d763c70e --- .../model/agenticark/event_convertor.go | 43 +++++ .../model/agenticark/event_convertor_test.go | 64 +++++++ components/model/agenticark/runtime_bridge.go | 170 +++++++++++++++++- .../model/agenticark/runtime_bridge_test.go | 42 ++++- 4 files changed, 314 insertions(+), 5 deletions(-) diff --git a/components/model/agenticark/event_convertor.go b/components/model/agenticark/event_convertor.go index c9d761c31..35e5ea778 100644 --- a/components/model/agenticark/event_convertor.go +++ b/components/model/agenticark/event_convertor.go @@ -110,6 +110,10 @@ func receivedStreamResponse(streamReader responseStreamReader, block := receiver.reasoningSummaryTextDeltaEventToContentBlock(ev.ReasoningText) sender.sendBlock(block, nil) + case *responses.Event_ReasoningRawTextDelta: + block := receiver.reasoningTextDeltaEventToContentBlock(ev.ReasoningRawTextDelta) + sender.sendBlock(block, nil) + case *responses.Event_FunctionCallArguments: block := receiver.functionCallArgumentsDeltaEventToContentBlock(ev.FunctionCallArguments) sender.sendBlock(block, nil) @@ -300,6 +304,9 @@ type streamReceiver struct { MaxReasoningSummaryIndex map[string]int ReasoningSummaryIndexMapper map[string]int + MaxReasoningContentIndex map[string]int + ReasoningContentIndexMapper map[string]int + MaxTextAnnotationIndex map[string]int TextAnnotationIndexMapper map[string]int @@ -313,6 +320,8 @@ func newStreamReceiver() *streamReceiver { IndexMapper: map[string]int{}, MaxReasoningSummaryIndex: map[string]int{}, ReasoningSummaryIndexMapper: map[string]int{}, + MaxReasoningContentIndex: map[string]int{}, + ReasoningContentIndexMapper: map[string]int{}, TextAnnotationIndexMapper: map[string]int{}, MaxTextAnnotationIndex: map[string]int{}, ItemAddedEventCache: map[string]any{}, @@ -348,6 +357,24 @@ func (r *streamReceiver) isNewReasoningSummaryIndex(outputIdx, summaryIdx int64) return true } +func (r *streamReceiver) isNewReasoningContentIndex(outputIdx, contentIdx int64) bool { + maxContentIndex := -1 + if idx, ok := r.MaxReasoningContentIndex[int64ToStr(outputIdx)]; ok { + maxContentIndex = idx + } + + idxKey := fmt.Sprintf("%d:%d", outputIdx, contentIdx) + if _, ok := r.ReasoningContentIndexMapper[idxKey]; ok { + return false + } + + maxContentIndex++ + r.ReasoningContentIndexMapper[idxKey] = maxContentIndex + r.MaxReasoningContentIndex[int64ToStr(outputIdx)] = maxContentIndex + + return true +} + func (r *streamReceiver) getTextAnnotationIndex(outputIdx, contentIdx, annotationIdx int64) int { maxAnnotationIndex := -1 @@ -797,6 +824,22 @@ func (r *streamReceiver) reasoningSummaryTextDeltaEventToContentBlock(ev *respon return block } +func (r *streamReceiver) reasoningTextDeltaEventToContentBlock(ev *responses.ReasoningTextDeltaEvent) *schema.ContentBlock { + text := ev.GetDelta() + if r.isNewReasoningContentIndex(ev.OutputIndex, ev.ContentIndex) && ev.ContentIndex != 0 { + text = "\n" + text + } + + meta := &schema.StreamingMeta{ + Index: r.getBlockIndex(makeReasoningIndexKey(ev.OutputIndex)), + } + block := schema.NewContentBlockChunk(&schema.Reasoning{Text: text}, meta) + + setItemID(block, ev.ItemId) + + return block +} + func (r *streamReceiver) functionCallArgumentsDeltaEventToContentBlock(ev *responses.FunctionCallArgumentsEvent) *schema.ContentBlock { meta := &schema.StreamingMeta{ Index: r.getBlockIndex(makeFunctionToolCallIndexKey(ev.OutputIndex)), diff --git a/components/model/agenticark/event_convertor_test.go b/components/model/agenticark/event_convertor_test.go index 1ee8e63ba..827caaeea 100644 --- a/components/model/agenticark/event_convertor_test.go +++ b/components/model/agenticark/event_convertor_test.go @@ -18,6 +18,7 @@ package agenticark import ( "errors" + "io" "testing" "github.com/cloudwego/eino/components/model" @@ -33,6 +34,8 @@ func TestNewStreamReceiverInit(t *testing.T) { assert.NotNil(t, r.IndexMapper) assert.NotNil(t, r.MaxReasoningSummaryIndex) assert.NotNil(t, r.ReasoningSummaryIndexMapper) + assert.NotNil(t, r.MaxReasoningContentIndex) + assert.NotNil(t, r.ReasoningContentIndexMapper) assert.NotNil(t, r.TextAnnotationIndexMapper) assert.NotNil(t, r.MaxTextAnnotationIndex) } @@ -391,6 +394,50 @@ func TestReasoningSummaryTextDeltaEventToContentBlock(t *testing.T) { assert.Equal(t, "x", block.Reasoning.Text) } +func TestReasoningTextDeltaEventToContentBlock(t *testing.T) { + r := newStreamReceiver() + first := r.reasoningTextDeltaEventToContentBlock(&responses.ReasoningTextDeltaEvent{ + ItemId: "iid", + OutputIndex: 2, + ContentIndex: 0, + Delta: ptrOf("first"), + }) + assert.Equal(t, "first", first.Reasoning.Text) + + second := r.reasoningTextDeltaEventToContentBlock(&responses.ReasoningTextDeltaEvent{ + ItemId: "iid", + OutputIndex: 2, + ContentIndex: 1, + Delta: ptrOf("second"), + }) + assert.Equal(t, "\nsecond", second.Reasoning.Text) + assert.Equal(t, first.StreamingMeta.Index, second.StreamingMeta.Index) +} + +func TestReceivedStreamResponse_ReasoningTextDelta(t *testing.T) { + stream := &fakeLegacyResponseStream{ + events: []*responses.Event{{ + Event: &responses.Event_ReasoningRawTextDelta{ + ReasoningRawTextDelta: &responses.ReasoningTextDeltaEvent{ + ItemId: "iid", + OutputIndex: 2, + ContentIndex: 0, + Delta: ptrOf("reasoning text"), + }, + }, + }}, + } + sr, sw := schema.Pipe[*model.AgenticCallbackOutput](1) + go func() { + receivedStreamResponse(stream, &model.AgenticConfig{}, sw) + sw.Close() + }() + + output, err := sr.Recv() + assert.NoError(t, err) + assert.Equal(t, "reasoning text", output.Message.ContentBlocks[0].Reasoning.Text) +} + func TestFunctionCallArgumentsDeltaEventToContentBlock(t *testing.T) { r := newStreamReceiver() block := r.functionCallArgumentsDeltaEventToContentBlock(&responses.FunctionCallArgumentsEvent{ @@ -482,3 +529,20 @@ func TestNewCallbackSenderAndSend(t *testing.T) { _, err = r0.Recv() assert.Error(t, err) } + +type fakeLegacyResponseStream struct { + events []*responses.Event +} + +func (s *fakeLegacyResponseStream) Recv() (*responses.Event, error) { + if len(s.events) == 0 { + return nil, io.EOF + } + event := s.events[0] + s.events = s.events[1:] + return event, nil +} + +func (s *fakeLegacyResponseStream) Close() error { + return nil +} diff --git a/components/model/agenticark/runtime_bridge.go b/components/model/agenticark/runtime_bridge.go index 8da194854..2e2943a76 100644 --- a/components/model/agenticark/runtime_bridge.go +++ b/components/model/agenticark/runtime_bridge.go @@ -57,7 +57,7 @@ func (c *runtimeBridge) CreateResponses(ctx context.Context, req *legacyresponse } var legacy legacyresponses.ResponseObject - if err := convertJSON(&res.Response, &legacy); err != nil { + if err := responseToLegacy(&res.Response, &legacy); err != nil { return nil, fmt.Errorf("convert responses response to legacy model: %w", err) } return &legacy, nil @@ -89,14 +89,79 @@ func (s *streamBridge) Recv() (*legacyresponses.Event, error) { if err != nil { return nil, err } + if legacy, ok := reasoningStreamEventToLegacy(event); ok { + return legacy, nil + } + + raw, err := json.Marshal(event) + if err != nil { + return nil, fmt.Errorf("marshal responses stream event: %w", err) + } var legacy legacyresponses.Event - if err := convertJSON(event, &legacy); err != nil { + if err := json.Unmarshal(raw, &legacy); err != nil { return nil, fmt.Errorf("convert responses stream event to legacy model: %w", err) } return &legacy, nil } +func reasoningStreamEventToLegacy(event *responsesv2.ResponseStreamEvent) (*legacyresponses.Event, bool) { + switch event.OneOf.Type { + case responsesv2.ResponseReasoningTextDeltaEventResponseStreamEventSum: + ev := event.OneOf.ResponseReasoningTextDeltaEvent + return newLegacyReasoningTextDeltaEvent(ev.ItemID, ev.OutputIndex, ev.ContentIndex, ev.SequenceNumber, ev.Delta), true + case responsesv2.ResponseReasoningRawTextDeltaEventResponseStreamEventSum: + ev := event.OneOf.ResponseReasoningRawTextDeltaEvent + return newLegacyReasoningTextDeltaEvent(ev.ItemID, ev.OutputIndex, ev.ContentIndex, ev.SequenceNumber, ev.Delta), true + case responsesv2.ResponseReasoningTextDoneEventResponseStreamEventSum: + ev := event.OneOf.ResponseReasoningTextDoneEvent + return newLegacyReasoningTextDoneEvent(ev.ItemID, ev.OutputIndex, ev.ContentIndex, ev.SequenceNumber, ev.Text), true + case responsesv2.ResponseReasoningRawTextDoneEventResponseStreamEventSum: + ev := event.OneOf.ResponseReasoningRawTextDoneEvent + return newLegacyReasoningTextDoneEvent(ev.ItemID, ev.OutputIndex, ev.ContentIndex, ev.SequenceNumber, ev.Text), true + default: + return nil, false + } +} + +func newLegacyReasoningTextDeltaEvent(itemID string, outputIndex, contentIndex, sequenceNumber int64, + delta responsesv2.OptString) *legacyresponses.Event { + event := &legacyresponses.ReasoningTextDeltaEvent{ + Type: legacyresponses.EventType_response_reasoning_text_delta, + ItemId: itemID, + OutputIndex: outputIndex, + ContentIndex: contentIndex, + SequenceNumber: sequenceNumber, + } + if value, ok := delta.Get(); ok { + event.Delta = &value + } + return &legacyresponses.Event{ + Event: &legacyresponses.Event_ReasoningRawTextDelta{ + ReasoningRawTextDelta: event, + }, + } +} + +func newLegacyReasoningTextDoneEvent(itemID string, outputIndex, contentIndex, sequenceNumber int64, + text responsesv2.OptString) *legacyresponses.Event { + event := &legacyresponses.ReasoningTextDoneEvent{ + Type: legacyresponses.EventType_response_reasoning_text_done, + ItemId: itemID, + OutputIndex: outputIndex, + ContentIndex: contentIndex, + SequenceNumber: sequenceNumber, + } + if value, ok := text.Get(); ok { + event.Text = &value + } + return &legacyresponses.Event{ + Event: &legacyresponses.Event_ReasoningRawTextDone{ + ReasoningRawTextDone: event, + }, + } +} + func (s *streamBridge) Close() error { return s.stream.Close() } @@ -171,6 +236,107 @@ func rewriteMultimodalToolOutput(raw json.RawMessage) (json.RawMessage, error) { return raw, nil } +func responseToLegacy(src *responsesv2.Response, dst *legacyresponses.ResponseObject) error { + raw, err := json.Marshal(src) + if err != nil { + return err + } + raw, err = rewriteReasoningContent(raw) + if err != nil { + return err + } + return json.Unmarshal(raw, dst) +} + +func rewriteReasoningContent(raw json.RawMessage) (json.RawMessage, error) { + var object map[string]json.RawMessage + if err := json.Unmarshal(raw, &object); err == nil { + var itemType string + if typeValue, ok := object["type"]; ok { + _ = json.Unmarshal(typeValue, &itemType) + } + if itemType == "reasoning" { + if err := appendReasoningContentToSummary(object); err != nil { + return nil, err + } + } + + for key, child := range object { + rewritten, err := rewriteReasoningContent(child) + if err != nil { + return nil, err + } + object[key] = rewritten + } + return json.Marshal(object) + } + + var list []json.RawMessage + if err := json.Unmarshal(raw, &list); err == nil { + for i, child := range list { + rewritten, err := rewriteReasoningContent(child) + if err != nil { + return nil, err + } + list[i] = rewritten + } + return json.Marshal(list) + } + return raw, nil +} + +func appendReasoningContentToSummary(item map[string]json.RawMessage) error { + content, ok := item["content"] + if !ok { + return nil + } + + var contentItems []json.RawMessage + if err := json.Unmarshal(content, &contentItems); err != nil { + return fmt.Errorf("unmarshal reasoning content: %w", err) + } + + var reasoningTextParts []json.RawMessage + for _, contentItem := range contentItems { + var block map[string]json.RawMessage + if err := json.Unmarshal(contentItem, &block); err != nil { + return fmt.Errorf("unmarshal reasoning content block: %w", err) + } + + var blockType string + if err := json.Unmarshal(block["type"], &blockType); err != nil { + return fmt.Errorf("unmarshal reasoning content block type: %w", err) + } + if blockType != "reasoning_text" { + continue + } + + text, ok := block["text"] + if !ok { + continue + } + reasoningTextParts = append(reasoningTextParts, json.RawMessage(fmt.Sprintf(`{"type":"summary_text","text":%s}`, text))) + } + if len(reasoningTextParts) == 0 { + return nil + } + + var summary []json.RawMessage + if summaryJSON, ok := item["summary"]; ok { + if err := json.Unmarshal(summaryJSON, &summary); err != nil { + return fmt.Errorf("unmarshal reasoning summary: %w", err) + } + } + summary = append(summary, reasoningTextParts...) + + summaryJSON, err := json.Marshal(summary) + if err != nil { + return err + } + item["summary"] = summaryJSON + return nil +} + func convertJSON(src, dst any) error { b, err := json.Marshal(src) if err != nil { diff --git a/components/model/agenticark/runtime_bridge_test.go b/components/model/agenticark/runtime_bridge_test.go index c247d6ef4..380b62e7d 100644 --- a/components/model/agenticark/runtime_bridge_test.go +++ b/components/model/agenticark/runtime_bridge_test.go @@ -179,6 +179,10 @@ func TestRuntimeBridgeSendsMultimodalToolOutput(t *testing.T) { func TestRuntimeBridgeConvertsResponse(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("ark-thinking-summary") != "skip-thinking-summary" { + http.Error(w, "missing thinking summary header", http.StatusBadRequest) + return + } w.Header().Set("Content-Type", "application/json") _, _ = w.Write([]byte(`{ "id":"resp-1", @@ -186,6 +190,12 @@ func TestRuntimeBridgeConvertsResponse(t *testing.T) { "created_at":0, "model":"model", "output":[{ + "id":"reasoning-1", + "type":"reasoning", + "status":"completed", + "summary":[], + "content":[{"type":"reasoning_text","text":"hidden reasoning","annotations":[]}] + },{ "id":"msg-1", "type":"message", "role":"assistant", @@ -210,12 +220,18 @@ func TestRuntimeBridgeConvertsResponse(t *testing.T) { Input: &legacyresponses.ResponsesInput{ Union: &legacyresponses.ResponsesInput_StringValue{StringValue: "hello"}, }, - }, nil) + }, map[string]string{"ark-thinking-summary": "skip-thinking-summary"}) require.NoError(t, err) require.NotNil(t, res) assert.Equal(t, "resp-1", res.Id) - require.Len(t, res.Output, 1) - assert.Equal(t, "hello", res.Output[0].GetOutputMessage().GetContent()[0].GetText().GetText()) + require.Len(t, res.Output, 2) + assert.Equal(t, "hidden reasoning", res.Output[0].GetReasoning().GetSummary()[0].GetText()) + assert.Equal(t, "hello", res.Output[1].GetOutputMessage().GetContent()[0].GetText().GetText()) + + message, err := toOutputMessage(res) + require.NoError(t, err) + require.Len(t, message.ContentBlocks, 2) + assert.Equal(t, "hidden reasoning", message.ContentBlocks[0].Reasoning.Text) } func TestRuntimeBridgeCreateResponsesStreamReturnsTransportError(t *testing.T) { @@ -254,6 +270,9 @@ func TestRuntimeBridgeCreateResponsesStreamReturnsTransportError(t *testing.T) { func TestRuntimeBridgeCreateResponsesStreamConvertsEvents(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte("data: {\"type\":\"response.reasoning_text.delta\",\"content_index\":0,\"delta\":\"reasoning text\",\"item_id\":\"reasoning-1\",\"output_index\":0,\"sequence_number\":1}\n\n")) + _, _ = w.Write([]byte("data: {\"type\":\"response.reasoning_raw_text.delta\",\"content_index\":1,\"delta\":\"raw reasoning\",\"item_id\":\"reasoning-1\",\"output_index\":0,\"sequence_number\":2}\n\n")) + _, _ = w.Write([]byte("data: {\"type\":\"response.reasoning_raw_text.done\",\"content_index\":1,\"text\":\"raw reasoning\",\"item_id\":\"reasoning-1\",\"output_index\":0,\"sequence_number\":3}\n\n")) _, _ = w.Write([]byte("data: {\"type\":\"error\",\"sequence_number\":1,\"code\":\"invalid_request\",\"message\":\"invalid request\"}\n\n")) _, _ = w.Write([]byte("data: [DONE]\n\n")) })) @@ -271,6 +290,23 @@ func TestRuntimeBridgeCreateResponsesStreamConvertsEvents(t *testing.T) { event, err := stream.Recv() require.NoError(t, err) + rawReasoning, ok := event.Event.(*legacyresponses.Event_ReasoningRawTextDelta) + require.True(t, ok) + assert.Equal(t, "reasoning text", rawReasoning.ReasoningRawTextDelta.GetDelta()) + + event, err = stream.Recv() + require.NoError(t, err) + rawReasoning, ok = event.Event.(*legacyresponses.Event_ReasoningRawTextDelta) + require.True(t, ok) + assert.Equal(t, "raw reasoning", rawReasoning.ReasoningRawTextDelta.GetDelta()) + + event, err = stream.Recv() + require.NoError(t, err) + _, ok = event.Event.(*legacyresponses.Event_ReasoningRawTextDone) + assert.True(t, ok) + + event, err = stream.Recv() + require.NoError(t, err) assert.Equal(t, "error", event.GetEventType()) _, err = stream.Recv() assert.ErrorIs(t, err, io.EOF)