From 2a482f3ae7181598fe41a9d045831b8cf6f5548c Mon Sep 17 00:00:00 2001 From: Hz_ Date: Mon, 13 Jul 2026 15:48:07 +0800 Subject: [PATCH] fix(go-agent): align GitHub and Invoke Canvas components (#16849) ## Summary - Add the GitHub Canvas component with tool registration and reference propagation. - Align the Invoke component with the Python contract for node config, input form, response output, and timing fields. - GitHub search and HTTP Invoke now work correctly in the Go Canvas runtime. ## Tests - `bash build.sh --test ./internal/agent/tool/...` - `bash build.sh --test ./internal/agent/component/...` Note: the untracked go_ragflow_cli file is not part of the PR changes. image image --- internal/agent/component/fixture_stubs.go | 1 + internal/agent/component/github.go | 230 ++++++++++++++++++++++ internal/agent/component/github_test.go | 165 ++++++++++++++++ internal/agent/component/invoke.go | 144 +++++++++----- internal/agent/component/invoke_test.go | 89 ++++++--- internal/agent/runtime/state.go | 38 ++++ internal/agent/runtime/state_test.go | 46 +++++ internal/agent/tool/github.go | 110 ++++++----- internal/agent/tool/github_test.go | 84 ++++++-- internal/agent/tool/registry.go | 29 ++- 10 files changed, 796 insertions(+), 140 deletions(-) create mode 100644 internal/agent/component/github.go create mode 100644 internal/agent/component/github_test.go diff --git a/internal/agent/component/fixture_stubs.go b/internal/agent/component/fixture_stubs.go index 1b8be85f77..1aa94f12f5 100644 --- a/internal/agent/component/fixture_stubs.go +++ b/internal/agent/component/fixture_stubs.go @@ -518,6 +518,7 @@ func init() { Register(componentNameIteration, NewIterationStub) Register(componentNameIterationItem, NewIterationItemStub) Register("BGPT", newBGPTComponent) + Register("GitHub", newGitHubComponent) Register("Wikipedia", newWikipediaComponent) Register("DuckDuckGo", newDuckDuckGoComponent) Register("Google", newGoogleComponent) diff --git a/internal/agent/component/github.go b/internal/agent/component/github.go new file mode 100644 index 0000000000..b338370426 --- /dev/null +++ b/internal/agent/component/github.go @@ -0,0 +1,230 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// 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 component + +import ( + "context" + "crypto/sha1" + "encoding/json" + "fmt" + "math/big" + "strconv" + "strings" + "unicode/utf8" + + einotool "github.com/cloudwego/eino/components/tool" + + "ragflow/internal/agent/runtime" + agenttool "ragflow/internal/agent/tool" + "ragflow/internal/tokenizer" +) + +const githubPromptMaxTokens = 200000 + +type githubInvoker interface { + InvokableRun(ctx context.Context, argsJSON string, opts ...einotool.Option) (string, error) +} + +// githubComponent is the Canvas-facing GitHub repository search component. +// It mirrors agent/tools/github.py: query is a runtime input, while top_n is +// validated once from node parameters and defaults to ten. +type githubComponent struct { + inner githubInvoker +} + +func newGitHubComponent(params map[string]any) (Component, error) { + toolParams := make(map[string]any, 1) + for _, key := range []string{"top_n"} { + if value, ok := params[key]; ok { + toolParams[key] = value + } + } + inner, err := agenttool.BuildByName("github", toolParams) + if err != nil { + return nil, err + } + invoker, ok := inner.(githubInvoker) + if !ok { + return nil, fmt.Errorf("GitHub: tool does not implement InvokableRun") + } + return newGitHubComponentWithInvoker(invoker), nil +} + +func newGitHubComponentWithInvoker(inner githubInvoker) Component { + return &githubComponent{inner: inner} +} + +func (c *githubComponent) Name() string { return "GitHub" } + +func (c *githubComponent) Inputs() map[string]string { + return map[string]string{ + "query": "The search keywords to execute with GitHub. Use the most important terms and synonyms from the original request.", + } +} + +func (c *githubComponent) GetInputForm() map[string]any { + return map[string]any{ + "query": map[string]any{ + "type": "line", + "name": "Query", + }, + } +} + +func (c *githubComponent) Outputs() map[string]string { + return map[string]string{ + "formalized_content": "GitHub repositories formatted for downstream prompts.", + "json": "Raw GitHub repository items.", + } +} + +func (c *githubComponent) Invoke(ctx context.Context, inputs map[string]any) (map[string]any, error) { + query := stringParam(inputs["query"]) + if query == "" { + return map[string]any{"formalized_content": "", "json": []any{}}, nil + } + argsJSON, err := json.Marshal(map[string]any{"query": query}) + if err != nil { + return nil, fmt.Errorf("canvas: GitHub: encode query: %w", err) + } + out, err := c.inner.InvokableRun(ctx, string(argsJSON)) + decoded := parseToolEnvelope(out) + items := anySlice(decoded["results"]) + if existing, _ := decoded["_ERROR"].(string); strings.TrimSpace(existing) != "" { + return map[string]any{"formalized_content": "", "json": items, "_ERROR": existing}, nil + } + if err != nil { + if len(decoded) > 0 { + return map[string]any{"formalized_content": "", "json": items, "_ERROR": decoded["_ERROR"]}, nil + } + return nil, fmt.Errorf("canvas: GitHub: %w", err) + } + chunks, docAggs := buildGitHubReferences(items) + if state, _, stateErr := runtime.GetStateFromContext[*runtime.CanvasState](ctx); stateErr == nil && state != nil { + state.SetRetrievalReferences(chunks, docAggs) + } + return map[string]any{ + "formalized_content": renderGitHubReferences(chunks), + "json": items, + }, nil +} + +func (c *githubComponent) Stream(_ context.Context, _ map[string]any) (<-chan map[string]any, error) { + return nil, nil +} + +func buildGitHubReferences(items []any) ([]map[string]any, []map[string]any) { + chunks := make([]map[string]any, 0, len(items)) + docAggs := make([]map[string]any, 0, len(items)) + for _, item := range items { + repository, ok := item.(map[string]any) + if !ok { + continue + } + content := truncateRunes(githubValueString(repository["description"])+"\n stars:"+githubValueString(repository["watchers"]), 10000) + if content == "" { + continue + } + documentID := strconv.FormatInt(githubHashInt(content, 100000000), 10) + title := githubValueString(repository["name"]) + url := githubValueString(repository["html_url"]) + displayID := strconv.FormatInt(githubHashInt(documentID, 500), 10) + chunks = append(chunks, map[string]any{ + "id": displayID, + "chunk_id": documentID, + "content": content, + "doc_id": documentID, + "document_id": documentID, + "docnm_kwd": title, + "document_name": title, + "similarity": 1, + "score": 1, + "url": url, + }) + docAggs = append(docAggs, map[string]any{ + "doc_name": title, + "doc_id": documentID, + "count": 1, + "url": url, + }) + } + return chunks, docAggs +} + +// renderGitHubReferences mirrors the Python +// "\n".join(kb_prompt({"chunks": chunks, "doc_aggs": aggs}, 200000, True)) +// layout for the GitHub chunks built above. +func renderGitHubReferences(chunks []map[string]any) string { + chunks = limitGitHubReferences(chunks, githubPromptMaxTokens) + blocks := make([]string, 0, len(chunks)) + for _, chunk := range chunks { + blocks = append(blocks, strings.Join([]string{ + "\nID: " + githubValueString(chunk["id"]), + "├── Title: " + githubValueString(chunk["docnm_kwd"]), + "├── URL: " + githubValueString(chunk["url"]), + "└── Content:\n" + githubValueString(chunk["content"]), + }, "\n")) + } + return strings.Join(blocks, "\n") +} + +// limitGitHubReferences mirrors kb_prompt's 200000-token guard. References +// are recorded before this step, as Python calls add_reference before +// formatting the prompt. +func limitGitHubReferences(chunks []map[string]any, maxTokens int) []map[string]any { + if maxTokens <= 0 { + return nil + } + usedTokens := 0 + for index, chunk := range chunks { + content := githubValueString(chunk["content"]) + if content == "" { + continue + } + usedTokens += tokenizer.NumTokensFromString(content) + if float64(maxTokens)*0.97 < float64(usedTokens) { + return chunks[:index+1] + } + } + return chunks +} + +func githubValueString(value any) string { + if value == nil { + return "None" + } + if boolean, ok := value.(bool); ok { + if boolean { + return "True" + } + return "False" + } + return fmt.Sprint(value) +} + +func githubHashInt(value string, modulus int64) int64 { + sum := sha1.Sum([]byte(value)) + number := new(big.Int).SetBytes(sum[:]) + return new(big.Int).Mod(number, big.NewInt(modulus)).Int64() +} + +func truncateRunes(value string, limit int) string { + if utf8.RuneCountInString(value) <= limit { + return value + } + return string([]rune(value)[:limit]) +} diff --git a/internal/agent/component/github_test.go b/internal/agent/component/github_test.go new file mode 100644 index 0000000000..00863cd62c --- /dev/null +++ b/internal/agent/component/github_test.go @@ -0,0 +1,165 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// 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 component + +import ( + "context" + "encoding/json" + "strings" + "testing" + + einotool "github.com/cloudwego/eino/components/tool" + + "ragflow/internal/agent/canvas" + "ragflow/internal/agent/runtime" +) + +type fakeGitHubInvoker struct { + args map[string]any + out string + err error +} + +func (f *fakeGitHubInvoker) InvokableRun(_ context.Context, argsJSON string, _ ...einotool.Option) (string, error) { + if err := json.Unmarshal([]byte(argsJSON), &f.args); err != nil { + return "", err + } + return f.out, f.err +} + +func TestGitHub_RegisteredFactory(t *testing.T) { + c, err := New("GitHub", map[string]any{"top_n": float64(10)}) + if err != nil { + t.Fatalf("New(GitHub) errored: %v", err) + } + if got := c.Name(); got != "GitHub" { + t.Fatalf("Name() = %q, want GitHub", got) + } + formGetter, ok := c.(interface{ GetInputForm() map[string]any }) + if !ok { + t.Fatal("GitHub component does not expose GetInputForm") + } + query, ok := formGetter.GetInputForm()["query"].(map[string]any) + if !ok || query["type"] != "line" { + t.Fatalf("GetInputForm query = %#v, want line input", query) + } + if _, ok := c.Outputs()["formalized_content"]; !ok { + t.Fatal("Outputs() missing formalized_content") + } + if _, ok := c.Outputs()["json"]; !ok { + t.Fatal("Outputs() missing json") + } +} + +func TestGitHub_CanvasBuildWorkflow(t *testing.T) { + c := &canvas.Canvas{ + Components: map[string]canvas.CanvasComponent{ + "begin_0": { + Obj: canvas.CanvasComponentObj{ComponentName: "Begin", Params: map[string]any{}}, + Downstream: []string{"github_0"}, + }, + "github_0": { + Obj: canvas.CanvasComponentObj{ComponentName: "GitHub", Params: map[string]any{"top_n": float64(10)}}, + Upstream: []string{"begin_0"}, + Downstream: []string{}, + }, + }, + Path: []string{"begin_0", "github_0"}, + } + if _, err := canvas.BuildWorkflow(context.Background(), c); err != nil { + t.Fatalf("BuildWorkflow with GitHub component: %v", err) + } +} + +func TestGitHub_InvokeMatchesPythonOutputsAndReferences(t *testing.T) { + fake := &fakeGitHubInvoker{out: `{"results":[{"name":"ragflow","html_url":"https://github.com/infiniflow/ragflow","description":"RAG engine","watchers":12000}]}`} + c := newGitHubComponentWithInvoker(fake) + state := runtime.NewCanvasState("run-github", "task-github") + ctx := runtime.WithState(context.Background(), state) + + out, err := c.Invoke(ctx, map[string]any{"query": "ragflow"}) + if err != nil { + t.Fatalf("Invoke errored: %v", err) + } + if got := fake.args["query"]; got != "ragflow" { + t.Errorf("query = %v, want ragflow", got) + } + content, _ := out["formalized_content"].(string) + for _, want := range []string{"Title: ragflow", "URL: https://github.com/infiniflow/ragflow", "RAG engine\n stars:12000"} { + if !strings.Contains(content, want) { + t.Errorf("formalized_content missing %q: %q", want, content) + } + } + items := anySlice(out["json"]) + if len(items) != 1 { + t.Fatalf("json length = %d, want 1", len(items)) + } + chunks := state.GetRetrievalChunks() + if len(chunks) != 1 { + t.Fatalf("recorded reference chunks = %d, want 1", len(chunks)) + } + if chunks[0]["document_name"] != "ragflow" || chunks[0]["url"] != "https://github.com/infiniflow/ragflow" { + t.Errorf("reference chunk metadata = %#v", chunks[0]) + } + if chunks[0]["similarity"] != 1 { + t.Errorf("reference similarity = %v, want 1", chunks[0]["similarity"]) + } + encodedState, err := json.Marshal(state) + if err != nil { + t.Fatalf("marshal state: %v", err) + } + var statePayload struct { + Retrieval struct { + DocAggs map[string]any `json:"doc_aggs"` + } `json:"retrieval"` + } + if err := json.Unmarshal(encodedState, &statePayload); err != nil { + t.Fatalf("unmarshal state: %v", err) + } + if _, ok := statePayload.Retrieval.DocAggs["ragflow"]; !ok { + t.Fatalf("doc_aggs missing ragflow: %#v", statePayload.Retrieval.DocAggs) + } +} + +func TestGitHub_InvokeEmptyQueryMatchesPython(t *testing.T) { + fake := &fakeGitHubInvoker{} + c := newGitHubComponentWithInvoker(fake) + out, err := c.Invoke(context.Background(), map[string]any{"query": ""}) + if err != nil { + t.Fatalf("Invoke errored: %v", err) + } + if got := out["formalized_content"]; got != "" { + t.Errorf("formalized_content = %v, want empty", got) + } + if items := anySlice(out["json"]); len(items) != 0 { + t.Errorf("json = %#v, want empty", items) + } + if fake.args != nil { + t.Fatal("GitHub tool was called for an empty query") + } +} + +func TestGitHub_LimitReferencesKeepsBoundaryChunk(t *testing.T) { + chunks := []map[string]any{ + {"content": "first repository description"}, + {"content": "second repository description"}, + } + limited := limitGitHubReferences(chunks, 1) + if len(limited) != 1 { + t.Fatalf("limited chunks = %d, want 1", len(limited)) + } +} diff --git a/internal/agent/component/invoke.go b/internal/agent/component/invoke.go index 5c35f4cf39..655c114d16 100644 --- a/internal/agent/component/invoke.go +++ b/internal/agent/component/invoke.go @@ -37,6 +37,7 @@ package component import ( "bytes" "context" + "encoding/json" "errors" "fmt" "io" @@ -61,21 +62,29 @@ const ( maxInvokeResponseBody = 16 << 20 // 16 MiB; hard cap to avoid OOM ) -// InvokeComponent is the HTTP client node. Stateless across invocations. +// invokeClockOrigin gives Invoke's _created_time the same monotonic-clock +// semantics as Python's time.perf_counter(). Its absolute value is process +// local; only elapsed durations are meaningful. +var invokeClockOrigin = time.Now() + +// InvokeComponent is the HTTP client node. Its node configuration is immutable +// across invocations. type InvokeComponent struct { - name string + name string + params map[string]any } // NewInvokeComponent constructs an Invoke component. -func NewInvokeComponent(_ map[string]any) (Component, error) { - return &InvokeComponent{name: componentNameInvoke}, nil +func NewInvokeComponent(params map[string]any) (Component, error) { + return &InvokeComponent{name: componentNameInvoke, params: cloneInvokeParams(params)}, nil } // Name returns the registered component name. func (i *InvokeComponent) Name() string { return i.name } -// Invoke executes a single HTTP request and returns the status code, -// body, and response headers. See Inputs() for the param contract. +// Invoke executes a single HTTP request and returns its response text as +// `result`, matching the Python Invoke component. See Inputs() for the +// param contract. // // SSRF flow (PR #15426): // 1. Validate the target URL via utility.AssertURLSafe (loopback / @@ -92,7 +101,17 @@ func (i *InvokeComponent) Name() string { return i.name } // On any of those checks failing the function returns an `_ERROR` // output (no Go error) so the canvas can route around the failure // the same way the Python fix does, instead of crashing the node. -func (i *InvokeComponent) Invoke(ctx context.Context, inputs map[string]any) (map[string]any, error) { +func (i *InvokeComponent) Invoke(ctx context.Context, inputs map[string]any) (output map[string]any, invokeErr error) { + startedAt := time.Now() + defer func() { + if output == nil { + return + } + output["_created_time"] = startedAt.Sub(invokeClockOrigin).Seconds() + output["_elapsed_time"] = time.Since(startedAt).Seconds() + }() + + inputs = i.mergeInputs(inputs) method, _ := inputs["method"].(string) method = strings.ToUpper(strings.TrimSpace(method)) switch method { @@ -187,11 +206,13 @@ func (i *InvokeComponent) Invoke(ctx context.Context, inputs map[string]any) (ma req.Header.Set("Content-Type", contentType) } req.Header.Set("User-Agent", defaultInvokeUserAgent) - if h, ok := inputs["headers"].(map[string]any); ok { - for k, v := range h { - if s, ok := v.(string); ok { - req.Header.Set(k, s) - } + headers, err := invokeHeaders(inputs["headers"]) + if err != nil { + return nil, err + } + for k, v := range headers { + if s, ok := v.(string); ok { + req.Header.Set(k, s) } } @@ -270,16 +291,6 @@ func (i *InvokeComponent) Invoke(ctx context.Context, inputs map[string]any) (ma return nil, fmt.Errorf("Invoke: read body: %w", err) } - hdr := make(map[string]string, len(resp.Header)) - for k, vs := range resp.Header { - // First value only — multi-value headers are uncommon in - // canvas-DSL HTTP responses, and the param contract specifies - // a string map. - if len(vs) > 0 { - hdr[k] = vs[0] - } - } - bodyStr := string(bodyBytes) // Clean HTML from response body when clean_html input is set. @@ -287,24 +298,72 @@ func (i *InvokeComponent) Invoke(ctx context.Context, inputs map[string]any) (ma bodyStr = stripHTMLTags(bodyStr) } - // Parse body according to the requested datatype. - datatype, _ := inputs["datatype"].(string) - if datatype == "" { - // Infer from Content-Type header. - ct := resp.Header.Get("Content-Type") - if strings.Contains(ct, "application/json") { - datatype = "json" - } else { - datatype = "text" + return map[string]any{ + "result": bodyStr, + }, nil +} + +// GetInputForm returns the variables an Invoke node accepts from the +// surrounding canvas. The HTTP method, URL, headers, and timeout are node +// configuration, while variables are supplied at run time. +func (i *InvokeComponent) GetInputForm() map[string]any { + form := make(map[string]any) + variables, _ := i.params["variables"].([]any) + for _, raw := range variables { + variable, ok := raw.(map[string]any) + if !ok { + continue + } + ref := strings.TrimSpace(stringParam(variable["ref"])) + if ref == "" { + continue + } + name := strings.TrimSpace(stringParam(variable["key"])) + if name == "" { + name = ref + } + form[ref] = map[string]any{"type": "line", "name": name} + } + return form +} + +func (i *InvokeComponent) mergeInputs(inputs map[string]any) map[string]any { + merged := cloneInvokeParams(i.params) + for key, value := range inputs { + if _, configured := merged[key]; !configured { + merged[key] = value } } + return merged +} - return map[string]any{ - "status": resp.StatusCode, - "body": bodyStr, - "headers": hdr, - "datatype": datatype, - }, nil +func cloneInvokeParams(params map[string]any) map[string]any { + cloned := make(map[string]any, len(params)) + for key, value := range params { + cloned[key] = value + } + return cloned +} + +func invokeHeaders(raw any) (map[string]any, error) { + if raw == nil { + return nil, nil + } + if headers, ok := raw.(map[string]any); ok { + return headers, nil + } + text, ok := raw.(string) + if !ok || strings.TrimSpace(text) == "" { + return nil, nil + } + var headers map[string]any + if err := json.Unmarshal([]byte(text), &headers); err != nil { + return nil, fmt.Errorf("Invoke: headers must be a JSON object: %w", err) + } + if headers == nil { + return nil, errors.New("Invoke: headers must be a JSON object") + } + return headers, nil } // invokeSSRFError builds the _ERROR output the canvas uses to route @@ -318,10 +377,8 @@ func invokeSSRFError(kind, raw string, err error) map[string]any { zap.Error(err), ) return map[string]any{ - "_ERROR": "URL not valid", - "status": 0, - "body": "", - "headers": map[string]string{}, + "_ERROR": "URL not valid", + "result": nil, } } @@ -377,10 +434,7 @@ func (i *InvokeComponent) Inputs() map[string]string { // Outputs returns the response surface. func (i *InvokeComponent) Outputs() map[string]string { return map[string]string{ - "status": "HTTP status code (int).", - "body": "Response body (string, truncated at 16 MiB).", - "headers": "Response headers (first value per key).", - "datatype": "Inferred response datatype: 'json' | 'text' | 'html'.", + "result": "Response body as text.", } } diff --git a/internal/agent/component/invoke_test.go b/internal/agent/component/invoke_test.go index 649592171f..f50c08390b 100644 --- a/internal/agent/component/invoke_test.go +++ b/internal/agent/component/invoke_test.go @@ -49,8 +49,7 @@ func setupAllowAnyHost(t *testing.T, enabled bool) { } // TestInvoke_GET exercises the happy path: a GET request to a stub -// server returns the canned body, and the response map carries the -// expected status / body / headers. +// server returns the canned body as the Python-compatible result output. func TestInvoke_GET(t *testing.T) { setupAllowAnyHost(t, true) @@ -58,7 +57,6 @@ func TestInvoke_GET(t *testing.T) { if r.Method != http.MethodGet { t.Errorf("server: got method %q, want GET", r.Method) } - w.Header().Set("X-Test", "ok") w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte("hello")) })) @@ -72,15 +70,17 @@ func TestInvoke_GET(t *testing.T) { if err != nil { t.Fatalf("Invoke: %v", err) } - if status, _ := out["status"].(int); status != http.StatusOK { - t.Errorf("status: got %d, want 200", status) + if got, _ := out["result"].(string); got != "hello" { + t.Errorf("result: got %q, want hello", got) } - if body, _ := out["body"].(string); body != "hello" { - t.Errorf("body: got %q, want %q", body, "hello") + if len(out) != 3 { + t.Errorf("output = %#v, want result and timing fields", out) } - hdr, _ := out["headers"].(map[string]string) - if hdr["X-Test"] != "ok" { - t.Errorf("headers[X-Test]: got %q, want %q", hdr["X-Test"], "ok") + if _, ok := out["_created_time"].(float64); !ok { + t.Errorf("_created_time = %#v, want float64", out["_created_time"]) + } + if elapsed, ok := out["_elapsed_time"].(float64); !ok || elapsed < 0 { + t.Errorf("_elapsed_time = %#v, want non-negative float64", out["_elapsed_time"]) } } @@ -109,17 +109,64 @@ func TestInvoke_POST(t *testing.T) { if err != nil { t.Fatalf("Invoke: %v", err) } - if status, _ := out["status"].(int); status != http.StatusCreated { - t.Errorf("status: got %d, want 201", status) - } if seenCT != "application/json" { t.Errorf("server saw Content-Type %q, want application/json (default)", seenCT) } if seenBody != `{"k":"v"}` { t.Errorf("server saw body %q, want %q", seenBody, `{"k":"v"}`) } - if body, _ := out["body"].(string); body != `echo:{"k":"v"}` { - t.Errorf("body: got %q, want %q", body, `echo:{"k":"v"}`) + if got, _ := out["result"].(string); got != `echo:{"k":"v"}` { + t.Errorf("result: got %q, want %q", got, `echo:{"k":"v"}`) + } +} + +func TestInvoke_UsesNodeParams(t *testing.T) { + setupAllowAnyHost(t, true) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("X-Configured"); got != "yes" { + t.Errorf("X-Configured = %q, want yes", got) + } + _, _ = w.Write([]byte("configured")) + })) + defer srv.Close() + + c, err := NewInvokeComponent(map[string]any{ + "method": "GET", + "url": srv.URL, + "headers": `{"X-Configured":"yes"}`, + }) + if err != nil { + t.Fatalf("NewInvokeComponent: %v", err) + } + out, err := c.Invoke(context.Background(), map[string]any{}) + if err != nil { + t.Fatalf("Invoke: %v", err) + } + if got, _ := out["result"].(string); got != "configured" { + t.Errorf("result = %q, want configured", got) + } +} + +func TestInvoke_GetInputForm(t *testing.T) { + c, err := NewInvokeComponent(map[string]any{ + "variables": []any{ + map[string]any{"key": "Search query", "ref": "sys.query"}, + }, + }) + if err != nil { + t.Fatalf("NewInvokeComponent: %v", err) + } + getter, ok := c.(interface{ GetInputForm() map[string]any }) + if !ok { + t.Fatal("Invoke does not expose GetInputForm") + } + field, ok := getter.GetInputForm()["sys.query"].(map[string]any) + if !ok { + t.Fatalf("GetInputForm()[sys.query] = %#v, want field", getter.GetInputForm()["sys.query"]) + } + if field["type"] != "line" || field["name"] != "Search query" { + t.Errorf("field = %#v, want line Search query", field) } } @@ -188,8 +235,8 @@ func TestInvoke_SSRFGuard_BlocksLoopback(t *testing.T) { if got, _ := out["_ERROR"].(string); got != "URL not valid" { t.Errorf("_ERROR = %q, want %q", got, "URL not valid") } - if status, _ := out["status"].(int); status != 0 { - t.Errorf("status = %d, want 0 on SSRF block", status) + if result, exists := out["result"]; !exists || result != nil { + t.Errorf("result = %#v, want nil result on SSRF block", result) } } @@ -280,12 +327,8 @@ func TestInvoke_NoRedirects_NotFollowed(t *testing.T) { if err != nil { t.Fatalf("Invoke: %v", err) } - if status, _ := out["status"].(int); status != http.StatusFound { - t.Errorf("status = %d, want 302 (no follow)", status) - } - hdr, _ := out["headers"].(map[string]string) - if hdr["Location"] != "http://127.0.0.1:1/secret" { - t.Errorf("Location header not preserved: %v", hdr) + if got, _ := out["result"].(string); got != "" { + t.Errorf("result = %q, want empty 302 response body", got) } } diff --git a/internal/agent/runtime/state.go b/internal/agent/runtime/state.go index 73580552bd..4f7ad0794a 100644 --- a/internal/agent/runtime/state.go +++ b/internal/agent/runtime/state.go @@ -389,6 +389,44 @@ func (s *CanvasState) SetRetrievalChunks(chunks []map[string]any) { s.Retrieval["chunks"] = asAny } +// SetRetrievalReferences records the chunks and document aggregates emitted by +// a canvas search component. It is the lock-safe counterpart of Python +// Graph.add_reference for components that produce externally sourced results. +func (s *CanvasState) SetRetrievalReferences(chunks, docAggs []map[string]any) { + if s == nil { + return + } + s.mu.Lock() + defer s.mu.Unlock() + if s.Retrieval == nil { + s.Retrieval = make(map[string]any) + } + chunkValues, _ := s.Retrieval["chunks"].([]any) + if chunkValues == nil { + chunkValues = make([]any, 0, len(chunks)) + } + for _, chunk := range chunks { + chunkValues = append(chunkValues, chunk) + } + docAggValues, _ := s.Retrieval["doc_aggs"].(map[string]any) + if docAggValues == nil { + docAggValues = make(map[string]any, len(docAggs)) + } + for _, docAgg := range docAggs { + docName, _ := docAgg["doc_name"].(string) + if docName == "" { + continue + } + // Match Python Graph.add_reference: retain the first aggregate for + // a document name across the run-level reference set. + if _, exists := docAggValues[docName]; !exists { + docAggValues[docName] = docAgg + } + } + s.Retrieval["chunks"] = chunkValues + s.Retrieval["doc_aggs"] = docAggValues +} + // getVarLocked is the lock-free inner GetVar. Caller must hold s.mu (read or // write) for the entire call. func getVarLocked(s *CanvasState, ref string) (any, error) { diff --git a/internal/agent/runtime/state_test.go b/internal/agent/runtime/state_test.go index 39b138c432..ce142869be 100644 --- a/internal/agent/runtime/state_test.go +++ b/internal/agent/runtime/state_test.go @@ -85,6 +85,52 @@ func TestCanvasState_MarshalJSON_DoesNotLeakMutex(t *testing.T) { } } +func TestCanvasState_SetRetrievalReferencesMergesCalls(t *testing.T) { + t.Parallel() + + state := NewCanvasState("run-1", "task-1") + state.SetRetrievalReferences( + []map[string]any{{"id": "chunk-1"}}, + []map[string]any{{"doc_name": "doc-1", "count": 1}}, + ) + state.SetRetrievalReferences( + []map[string]any{{"id": "chunk-2"}}, + []map[string]any{ + {"doc_name": "doc-1", "count": 99}, + {"doc_name": "doc-2", "count": 2}, + }, + ) + + chunks := state.GetRetrievalChunks() + if len(chunks) != 2 { + t.Fatalf("chunks length = %d, want 2", len(chunks)) + } + if chunks[0]["id"] != "chunk-1" || chunks[1]["id"] != "chunk-2" { + t.Errorf("chunk IDs = [%v %v], want [chunk-1 chunk-2]", chunks[0]["id"], chunks[1]["id"]) + } + + state.mu.RLock() + rawDocAggs := state.Retrieval["doc_aggs"] + docAggs, ok := rawDocAggs.(map[string]any) + state.mu.RUnlock() + if !ok { + t.Fatalf("doc_aggs type = %T, want map[string]any", rawDocAggs) + } + if len(docAggs) != 2 { + t.Fatalf("doc_aggs length = %d, want 2", len(docAggs)) + } + firstDoc, ok := docAggs["doc-1"].(map[string]any) + if !ok { + t.Fatalf("doc_aggs[doc-1] type = %T, want map[string]any", docAggs["doc-1"]) + } + if firstDoc["count"] != 1 { + t.Errorf("doc_aggs[doc-1].count = %v, want first value 1", firstDoc["count"]) + } + if _, ok := docAggs["doc-2"]; !ok { + t.Error("doc_aggs missing doc-2 from the second call") + } +} + func contains(s, substr string) bool { for i := 0; i+len(substr) <= len(s); i++ { if s[i:i+len(substr)] == substr { diff --git a/internal/agent/tool/github.go b/internal/agent/tool/github.go index 22b20f8610..7589f67270 100644 --- a/internal/agent/tool/github.go +++ b/internal/agent/tool/github.go @@ -22,42 +22,39 @@ import ( "fmt" "net/http" "net/url" - "strings" + "time" "github.com/cloudwego/eino/components/tool" "github.com/cloudwego/eino/schema" ) -const githubToolName = "github" +const githubToolName = "github_search" -const githubToolDescription = "Search GitHub repositories. Returns items[].{full_name, html_url, description, stargazers_count}." +const githubToolDescription = "GitHub repository search finds repositories, projects, and codebases hosted on GitHub." -// githubParams is the JSON shape the model sends into InvokableRun. token -// is optional — anonymous requests succeed but are heavily rate-limited -// (60/hr vs 5000/hr with a PAT). +const ( + defaultGitHubTopN = 10 + maxGitHubTopN = 100 +) + +// githubParams mirrors Python GitHubParam. Info() exposes only Query to the +// model, while TopN is a canvas-side configuration value merged with defaults. type githubParams struct { - Query string `json:"query"` - MaxResults int `json:"max_results"` - Token string `json:"token"` + Query string `json:"query"` + TopN int `json:"top_n"` } -// githubResult mirrors one element of the upstream `items` array. -type githubResult struct { - FullName string `json:"full_name"` - HTMLURL string `json:"html_url"` - Description string `json:"description"` - StargazersCount int `json:"stargazers_count"` -} - -// githubResponse is the upstream GitHub Search envelope. +// githubResponse keeps GitHub's raw repository objects intact. The Python +// component stores response["items"] in its json output, so narrowing this to +// selected fields would change downstream DSL behaviour. type githubResponse struct { - Items []githubResult `json:"items"` + Items []map[string]any `json:"items"` } -// githubEnvelope is what the model sees. +// githubEnvelope is the shared tool-to-component transport shape. type githubEnvelope struct { - Results []githubResult `json:"results"` - Error string `json:"_ERROR,omitempty"` + Results []map[string]any `json:"results"` + Error string `json:"_ERROR,omitempty"` } // GitHubTool is the GitHub @@ -65,21 +62,35 @@ type githubEnvelope struct { // GETs the GitHub Search API via the shared HTTPHelper and returns the // top N repository matches. type GitHubTool struct { - helper *HTTPHelper + helper *HTTPHelper + defaults githubParams } // NewGitHubTool returns a GitHubTool using the default HTTPHelper. func NewGitHubTool() *GitHubTool { - return NewGitHubToolWith(NewHTTPHelper()) + return NewGitHubToolWithDefaults(nil, githubParams{}) } // NewGitHubToolWith returns a GitHubTool that uses the provided // HTTPHelper. Useful for tests. func NewGitHubToolWith(h *HTTPHelper) *GitHubTool { + return NewGitHubToolWithDefaults(h, githubParams{}) +} + +// NewGitHubToolWithDefaults returns a GitHubTool with component-level +// defaults. It follows NewPubMedToolWithDefaults: the constructor owns +// defaults while Info() exposes only the model-call input schema. +func NewGitHubToolWithDefaults(h *HTTPHelper, defaults githubParams) *GitHubTool { if h == nil { - h = NewHTTPHelper() + // Python uses DEFAULT_TIMEOUT (15 seconds) and ToolParamBase starts + // with max_retries=0, so a default GitHub component makes one request. + h = NewHTTPHelperWithRetry(RetryConfig{MaxAttempts: 1}) + h.client.Timeout = 15 * time.Second } - return &GitHubTool{helper: h} + if defaults.TopN == 0 { + defaults.TopN = defaultGitHubTopN + } + return &GitHubTool{helper: h, defaults: defaults} } // Info returns the tool's metadata for the chat model. @@ -90,35 +101,24 @@ func (g *GitHubTool) Info(_ context.Context) (*schema.ToolInfo, error) { ParamsOneOf: schema.NewParamsOneOfByParams(map[string]*schema.ParameterInfo{ "query": { Type: schema.String, - Desc: "Search query (GitHub search syntax).", + Desc: "The search keywords to execute with GitHub. Use the most important terms and synonyms from the original request.", Required: true, }, - "max_results": { - Type: schema.Integer, - Desc: "Maximum number of results to return. Defaults to 5 (max 30 per page).", - Required: false, - }, - "token": { - Type: schema.String, - Desc: "Optional GitHub personal access token. Increases rate limit from 60 to 5000 req/hr.", - Required: false, - }, }), }, nil } -// buildGitHubURL constructs the GitHub repository search URL. Centralized -// so the test suite can verify URL encoding without spinning up a server. -func buildGitHubURL(query string, maxResults int) string { - if maxResults <= 0 { - maxResults = 5 - } - if maxResults > 30 { - maxResults = 30 +// buildGitHubURL constructs the repository search URL used by the Python +// GitHub component: most-starred repositories first, with per_page=top_n. +func buildGitHubURL(query string, topN int) string { + if topN <= 0 { + topN = defaultGitHubTopN } q := url.Values{} q.Set("q", query) - q.Set("per_page", fmt.Sprintf("%d", maxResults)) + q.Set("sort", "stars") + q.Set("order", "desc") + q.Set("per_page", fmt.Sprintf("%d", topN)) return "https://api.github.com/search/repositories?" + q.Encode() } @@ -129,17 +129,16 @@ func (g *GitHubTool) InvokableRun(ctx context.Context, argsJSON string, _ ...too return githubErrJSON(fmt.Errorf("github: parse arguments: %w", err)), fmt.Errorf("github: parse arguments: %w", err) } - if strings.TrimSpace(p.Query) == "" { + if p.Query == "" { return githubErrJSON(fmt.Errorf("query is required")), fmt.Errorf("github: query is required") } + p = mergeGitHubDefaults(g.defaults, p) - endpoint := buildGitHubURL(p.Query, p.MaxResults) + endpoint := buildGitHubURL(p.Query, p.TopN) headers := map[string]string{ - "Accept": "application/vnd.github+json", - } - if p.Token != "" { - headers["Authorization"] = "Bearer " + p.Token + "Content-Type": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", } resp, err := g.helper.Do(ctx, http.MethodGet, endpoint, "", "", headers) @@ -161,6 +160,13 @@ func (g *GitHubTool) InvokableRun(ctx context.Context, argsJSON string, _ ...too return githubJSON(githubEnvelope{Results: raw.Items}), nil } +func mergeGitHubDefaults(defaults, params githubParams) githubParams { + if params.TopN == 0 { + params.TopN = defaults.TopN + } + return params +} + func githubJSON(env githubEnvelope) string { b, err := json.Marshal(env) if err != nil { diff --git a/internal/agent/tool/github_test.go b/internal/agent/tool/github_test.go index a13f17fe8f..e18cd81009 100644 --- a/internal/agent/tool/github_test.go +++ b/internal/agent/tool/github_test.go @@ -40,14 +40,14 @@ func TestGitHub_BuildURL(t *testing.T) { name: "default", query: "ragflow", max: 0, - wantPerPage: "5", + wantPerPage: "10", wantHost: "api.github.com", }, { - name: "clamped high", + name: "preserves configured top n", query: "x", max: 99, - wantPerPage: "30", + wantPerPage: "99", wantHost: "api.github.com", }, { @@ -79,6 +79,12 @@ func TestGitHub_BuildURL(t *testing.T) { if q.Get("per_page") != tc.wantPerPage { t.Errorf("per_page = %q, want %q", q.Get("per_page"), tc.wantPerPage) } + if q.Get("sort") != "stars" { + t.Errorf("sort = %q, want stars", q.Get("sort")) + } + if q.Get("order") != "desc" { + t.Errorf("order = %q, want desc", q.Get("order")) + } }) } } @@ -86,14 +92,16 @@ func TestGitHub_BuildURL(t *testing.T) { func TestGitHub_ParseResponse(t *testing.T) { t.Parallel() - var gotAuth string + var gotContentType, gotAPIVersion, gotPerPage string srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - gotAuth = r.Header.Get("Authorization") + gotContentType = r.Header.Get("Content-Type") + gotAPIVersion = r.Header.Get("X-GitHub-Api-Version") + gotPerPage = r.URL.Query().Get("per_page") w.Header().Set("Content-Type", "application/json") _, _ = w.Write([]byte(`{ "items": [ - {"full_name":"infiniflow/ragflow","html_url":"https://github.com/infiniflow/ragflow","description":"RAG engine","stargazers_count":12000}, - {"full_name":"foo/bar","html_url":"https://github.com/foo/bar","description":"","stargazers_count":5} + {"name":"ragflow","html_url":"https://github.com/infiniflow/ragflow","description":"RAG engine","watchers":12000,"private":false}, + {"name":"bar","html_url":"https://github.com/foo/bar","description":"","watchers":5} ] }`)) })) @@ -102,9 +110,9 @@ func TestGitHub_ParseResponse(t *testing.T) { helper := NewHTTPHelper().WithClient(&http.Client{ Transport: rewriteHostTransport(srv.URL), }) - tool := NewGitHubToolWith(helper) + tool := NewGitHubToolWithDefaults(helper, githubParams{TopN: 17}) out, err := tool.InvokableRun(context.Background(), - `{"query":"ragflow","max_results":5,"token":"ghp_xyz"}`) + `{"query":"ragflow"}`) if err != nil { t.Fatalf("InvokableRun: %v", err) } @@ -119,17 +127,23 @@ func TestGitHub_ParseResponse(t *testing.T) { if len(env.Results) != 2 { t.Fatalf("Results len = %d, want 2", len(env.Results)) } - if env.Results[0].FullName != "infiniflow/ragflow" { - t.Errorf("Results[0].FullName = %q, want infiniflow/ragflow", env.Results[0].FullName) + if env.Results[0]["name"] != "ragflow" { + t.Errorf("Results[0].name = %v, want ragflow", env.Results[0]["name"]) } - if env.Results[0].StargazersCount != 12000 { - t.Errorf("Results[0].StargazersCount = %d, want 12000", env.Results[0].StargazersCount) + if env.Results[0]["watchers"] != float64(12000) { + t.Errorf("Results[0].watchers = %v, want 12000", env.Results[0]["watchers"]) } - if env.Results[1].Description != "" { - t.Errorf("Results[1].Description = %q, want empty", env.Results[1].Description) + if env.Results[1]["description"] != "" { + t.Errorf("Results[1].description = %q, want empty", env.Results[1]["description"]) } - if gotAuth != "Bearer ghp_xyz" { - t.Errorf("Authorization = %q, want Bearer ghp_xyz", gotAuth) + if gotContentType != "application/vnd.github+json" { + t.Errorf("Content-Type = %q, want application/vnd.github+json", gotContentType) + } + if gotAPIVersion != "2022-11-28" { + t.Errorf("X-GitHub-Api-Version = %q, want 2022-11-28", gotAPIVersion) + } + if gotPerPage != "17" { + t.Errorf("per_page = %q, want 17 from the component top_n default", gotPerPage) } } @@ -146,6 +160,38 @@ func TestGitHub_RequiresQuery(t *testing.T) { } } +func TestGitHub_BuildByNameUsesPythonNodeParams(t *testing.T) { + built, err := BuildByName("github", map[string]any{"top_n": float64(17)}) + if err != nil { + t.Fatalf("BuildByName(github): %v", err) + } + github, ok := built.(*GitHubTool) + if !ok { + t.Fatalf("BuildByName(github) returned %T, want *GitHubTool", built) + } + if github.defaults.TopN != 17 { + t.Errorf("defaults.TopN = %d, want 17", github.defaults.TopN) + } + if _, err := BuildByName("github", map[string]any{"top_n": 100}); err != nil { + t.Errorf("BuildByName(github) rejected GitHub's maximum top_n: %v", err) + } + if _, err := BuildByName("github", map[string]any{"top_n": 0}); err == nil { + t.Fatal("BuildByName(github) accepted non-positive top_n") + } + if _, err := BuildByName("github", map[string]any{"top_n": 1.5}); err == nil { + t.Fatal("BuildByName(github) accepted fractional top_n") + } + if _, err := BuildByName("github", map[string]any{"top_n": "10"}); err == nil { + t.Fatal("BuildByName(github) accepted string top_n") + } + if _, err := BuildByName("github", map[string]any{"top_n": 101}); err == nil { + t.Fatal("BuildByName(github) accepted top_n above GitHub's per_page limit") + } + if _, err := BuildByName("github", map[string]any{"max_results": 5}); err == nil { + t.Fatal("BuildByName(github) accepted removed max_results parameter") + } +} + func TestGitHub_Info(t *testing.T) { t.Parallel() @@ -154,8 +200,8 @@ func TestGitHub_Info(t *testing.T) { if err != nil { t.Fatalf("Info: %v", err) } - if info.Name != "github" { - t.Errorf("Name = %q, want github", info.Name) + if info.Name != "github_search" { + t.Errorf("Name = %q, want github_search", info.Name) } if !strings.Contains(info.Desc, "GitHub") { t.Errorf("Desc = %q, want to mention GitHub", info.Desc) diff --git a/internal/agent/tool/registry.go b/internal/agent/tool/registry.go index 1fcac9500b..30d334056e 100644 --- a/internal/agent/tool/registry.go +++ b/internal/agent/tool/registry.go @@ -18,6 +18,7 @@ package tool import ( "fmt" + "math" "strings" einotool "github.com/cloudwego/eino/components/tool" @@ -39,7 +40,7 @@ var registry = map[string]Factory{ "email": noConfig("email", func() einotool.BaseTool { return NewEmailTool() }), "execute_sql": buildExeSQLTool, "exesql": buildExeSQLTool, - "github": noConfig("github", func() einotool.BaseTool { return NewGitHubTool() }), + "github": buildGitHubTool, "google": buildGoogleTool, "google_scholar": buildGoogleScholarTool, "google_scholar_search": buildGoogleScholarTool, @@ -194,6 +195,32 @@ func buildGoogleTool(params map[string]any) (einotool.BaseTool, error) { return NewGoogleToolWithDefaults(nil, defaults), nil } +func buildGitHubTool(params map[string]any) (einotool.BaseTool, error) { + topN := defaultGitHubTopN + for key := range params { + if key != "top_n" { + return nil, fmt.Errorf("agent tool: tool %q only accepts node-level param top_n", "github") + } + } + if raw, exists := params["top_n"]; exists { + value, ok := intParam(params, "top_n") + if !ok { + return nil, fmt.Errorf("agent tool: tool %q requires positive integer node-level param top_n", "github") + } + if decimal, ok := raw.(float64); ok && math.Trunc(decimal) != decimal { + return nil, fmt.Errorf("agent tool: tool %q requires positive integer node-level param top_n", "github") + } + topN = value + } + if topN <= 0 { + return nil, fmt.Errorf("agent tool: tool %q requires positive integer node-level param top_n", "github") + } + if topN > maxGitHubTopN { + return nil, fmt.Errorf("agent tool: tool %q requires node-level param top_n to be at most %d", "github", maxGitHubTopN) + } + return NewGitHubToolWithDefaults(nil, githubParams{TopN: topN}), nil +} + func buildGoogleScholarTool(params map[string]any) (einotool.BaseTool, error) { if len(params) == 0 { return NewGoogleScholarTool(), nil