diff --git a/deps/agent-native b/deps/agent-native index fb2b3b7..e4020d2 160000 --- a/deps/agent-native +++ b/deps/agent-native @@ -1 +1 @@ -Subproject commit fb2b3b796c57246306cd7384b60fb671ce371143 +Subproject commit e4020d297c168347626f9fd75456a71374a0a6c8 diff --git a/deps/agent-squad b/deps/agent-squad index a3b49e7..dcd523e 160000 --- a/deps/agent-squad +++ b/deps/agent-squad @@ -1 +1 @@ -Subproject commit a3b49e7b1f563b34661cec026f58eec3f135fdbd +Subproject commit dcd523ea05fe44901a6ab06e9cc8a2f0d074e0e6 diff --git a/internal/app/app.go b/internal/app/app.go index 436a32c..cc6db91 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -107,6 +107,9 @@ func Run(opt Options) error { if err := tools.RegisterInterpreterTools(reg); err != nil { return fmt.Errorf("interpreter tools: %w", err) } + if err := tools.RegisterCodeTool(reg); err != nil { + return fmt.Errorf("code orchestration tool: %w", err) + } setSvc := settings.NewService(st) setSvc.SetSystemOverride(map[string]any{"storage": map[string]any{"dataDir": opt.DataDir}}) // 网络搜索（Tavily）：配置密钥池才注册（对齐 agent-native 语义：未配置不静默降级）。 @@ -188,9 +191,26 @@ func Run(opt Options) error { if pluginHost != nil { rawMCP = mergeMCPServers(rawMCP, pluginHost.MCPServers()) } - if mgr := mcpManagerFor(rawMCP); mgr != nil { + if mgr := mcpManagerFor(rawMCP, strings.TrimSpace(effNow.GetString("tools.mcp.overrides", ""))); mgr != nil { cfg.MCP = mgr } + cfg.CodeOrchestration = effNow.GetBool("runtime.codeOrchestration", false) + if ask, ok := rc.(interface { + AskUser(card core.HITLCard) (core.HITLResponse, error) + }); ok { + cfg.Ask = func(_ context.Context, question string) (string, error) { + resp, err := ask.AskUser(core.HITLCard{ + Question: question, Kind: core.HITLKindInput, AllowFreeInput: true, + }) + if err != nil { + return "", err + } + if strings.TrimSpace(resp.FreeText) != "" { + return strings.TrimSpace(resp.FreeText), nil + } + return resp.ChoiceID, nil + } + } cfg.Trace = traceFn if host, ok := rc.(interface { OpenCheckpoints() (*checkpoint.Log, error) @@ -524,9 +544,10 @@ var ( mcpManager *agentmcp.Manager ) -func mcpManagerFor(raw string) *agentmcp.Manager { +func mcpManagerFor(raw, overrides string) *agentmcp.Manager { mcpMu.Lock() defer mcpMu.Unlock() + key := raw + "\n" + overrides if raw == "" { if mcpManager != nil { mcpManager.Close() @@ -535,7 +556,7 @@ func mcpManagerFor(raw string) *agentmcp.Manager { mcpKey = "" return nil } - if raw == mcpKey && mcpManager != nil { + if key == mcpKey && mcpManager != nil { return mcpManager } var servers []agentmcp.ServerConfig @@ -548,10 +569,14 @@ func mcpManagerFor(raw string) *agentmcp.Manager { log.Printf("squad agent-native: MCP config invalid, MCP disabled: %v", err) return nil } + if err := mgr.ApplyOverrides(overrides); err != nil { + log.Printf("squad agent-native: MCP overrides invalid, MCP disabled: %v", err) + return nil + } if mcpManager != nil { mcpManager.Close() } - mcpKey, mcpManager = raw, mgr + mcpKey, mcpManager = key, mgr return mgr } diff --git a/internal/app/mcp_cache_test.go b/internal/app/mcp_cache_test.go index 56ef5d3..24a5d6d 100644 --- a/internal/app/mcp_cache_test.go +++ b/internal/app/mcp_cache_test.go @@ -21,24 +21,24 @@ func TestMcpManagerForCacheAndRebuild(t *testing.T) { cfgA := `[{"name":"a","command":"/bin/true"}]` cfgB := `[{"name":"b","command":"/bin/true"}]` - m1 := mcpManagerFor(cfgA) + m1 := mcpManagerFor(cfgA, "") if m1 == nil { t.Fatal("valid config must build a manager") } - if m2 := mcpManagerFor(cfgA); m2 != m1 { + if m2 := mcpManagerFor(cfgA, ""); m2 != m1 { t.Fatal("same config must reuse the cached manager") } - m3 := mcpManagerFor(cfgB) + m3 := mcpManagerFor(cfgB, "") if m3 == nil || m3 == m1 { t.Fatal("changed config must rebuild a fresh manager") } - if mcpManagerFor("") != nil { + if mcpManagerFor("", "") != nil { t.Fatal("empty config disables MCP and clears the cache") } - if mcpManagerFor(`{bad json`) != nil { + if mcpManagerFor(`{bad json`, "") != nil { t.Fatal("invalid JSON must disable MCP (nil), not panic") } - if mcpManagerFor(`[{"name":"bad name","command":"x"}]`) != nil { + if mcpManagerFor(`[{"name":"bad name","command":"x"}]`, "") != nil { t.Fatal("invalid server config must disable MCP (nil)") } // 并发调用不死锁、返回同一实例或各自重建（缓存一致性由锁保证）。 @@ -47,7 +47,7 @@ func TestMcpManagerForCacheAndRebuild(t *testing.T) { wg.Add(1) go func() { defer wg.Done() - _ = mcpManagerFor(cfgB) + _ = mcpManagerFor(cfgB, "") }() } wg.Wait() @@ -66,7 +66,7 @@ func TestMcpCacheConcurrencyWithSettingsResolution(t *testing.T) { wg.Add(1) go func(i int) { defer wg.Done() - _ = mcpManagerFor(cfgs[i%len(cfgs)]) + _ = mcpManagerFor(cfgs[i%len(cfgs)], "") }(i) } wg.Wait() diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index eb3f513..d04a944 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -290,10 +290,14 @@ func (c *runCtx) Tools() []core.ToolDesc { list := c.mgr.registry.List() out := make([]core.ToolDesc, 0, len(list)) fs := c.eff.GetBool("runtime.fsTools", false) + code := c.eff.GetBool("runtime.codeOrchestration", false) for _, t := range list { if t.Hidden && !fs { continue } + if t.Name == "run_tool_script" && !code { + continue + } readOnly := t.ReadOnly || agentpolicy.TraitsFor(t.Name).ReadOnly if !agentpolicy.Visible(c.permissionMode, t.Name, readOnly) { continue diff --git a/internal/runtime/agentwf.py b/internal/runtime/agentwf.py deleted file mode 100644 index d31d477..0000000 --- a/internal/runtime/agentwf.py +++ /dev/null @@ -1,51 +0,0 @@ -"""工作流脚本回调。用法：import agentwf as wf。 - -请求写到 fd 3，响应从 fd 4 按行读回。宿主一次只处理一条。 -""" - -import json -import os - -_seq = 0 -_out = os.fdopen(3, "w", encoding="utf-8", buffering=1) -_inp = os.fdopen(4, "r", encoding="utf-8", buffering=1) - - -def _rpc(op, **fields): - global _seq - _seq += 1 - msg = {"id": str(_seq), "op": op} - msg.update(fields) - _out.write(json.dumps(msg, ensure_ascii=False) + "\n") - _out.flush() - line = _inp.readline() - if not line: - raise RuntimeError("workflow host closed") - resp = json.loads(line) - if not resp.get("ok"): - raise RuntimeError(resp.get("error") or "workflow call failed") - return resp.get("output", "") - - -def tool(name, args=None): - return _rpc("tool.call", name=name, args=args or {}) - - -def ask(question, choices=None): - return _rpc("hitl.ask", question=question, choices=choices or []) - - -def agent(name, task): - return _rpc("agent.spawn", agent=name, task=task) - - -def artifact(name, content): - return _rpc("artifact.register", name=name, content=content) - - -def card(title, kind="result"): - return _rpc("card.create", title=title, kind=kind) - - -def step(step_id, title, status): - return _rpc("step", stepId=step_id, title=title, status=status) diff --git a/internal/runtime/workflow_script.go b/internal/runtime/workflow_script.go index 0e126fa..c6f9880 100644 --- a/internal/runtime/workflow_script.go +++ b/internal/runtime/workflow_script.go @@ -1,27 +1,21 @@ package runtime import ( - "bufio" - "bytes" "context" "encoding/json" "fmt" - "io" "os" + "os/exec" "path/filepath" "strings" - "sync" "time" + "agent-native/runtime/codeorch" "agent-system-backend/internal/tools" "agent-system-backend/internal/workflow" core "agent-system-core" - _ "embed" ) -//go:embed agentwf.py -var agentwfSource string - const ( scriptDefaultTimeout = 600 * time.Second scriptMaxTimeout = 1800 * time.Second @@ -29,28 +23,6 @@ const ( scriptMaxLine = 1 << 20 ) -var ( - agentwfOnce sync.Once - agentwfDir string - agentwfErr error -) - -func agentwfHelperDir() (string, error) { - agentwfOnce.Do(func() { - dir, err := os.MkdirTemp("", "agentwf-*") - if err != nil { - agentwfErr = err - return - } - if err := os.WriteFile(filepath.Join(dir, "agentwf.py"), []byte(agentwfSource), 0o644); err != nil { - agentwfErr = err - return - } - agentwfDir = dir - }) - return agentwfDir, agentwfErr -} - func scriptsEnabled(rc RunCtx) bool { return rc.Settings().GetBool("runtime.workflowScripts", false) } @@ -176,176 +148,33 @@ func runScriptFile(rc RunCtx, runner WorkflowRunner, path string) (string, []*co if timeout <= 0 { timeout = scriptDefaultTimeout } - if timeout > scriptMaxTimeout { - timeout = scriptMaxTimeout - } - helper, err := agentwfHelperDir() - if err != nil { - return "", nil, err - } parent := rc.Context() if parent == nil { parent = context.Background() } - ctx, cancel := context.WithTimeout(parent, timeout) - defer cancel() - cmd, err := tools.ScriptCommand(ctx, sandbox, path) - if err != nil { - return "", nil, err - } - cmd.Env = prependPythonPath(cmd.Env, helper) - reqR, reqW, err := os.Pipe() - if err != nil { - return "", nil, err - } - respR, respW, err := os.Pipe() - if err != nil { - reqR.Close() - reqW.Close() - return "", nil, err - } - cmd.ExtraFiles = []*os.File{reqW, respR} - var stdout, stderr bytes.Buffer - cmd.Stdout = &stdout - cmd.Stderr = &stderr - if err := cmd.Start(); err != nil { - reqR.Close() - reqW.Close() - respR.Close() - respW.Close() - return "", nil, err - } - reqW.Close() - respR.Close() - defer reqR.Close() - defer respW.Close() - arts, callErr := serveScript(ctx, rc, reqR, respW) - if callErr != nil { - cancel() - } - waitErr := cmd.Wait() - text := clipScriptOutput(&stdout, &stderr) - if callErr != nil { - return text, arts, callErr - } - if ctx.Err() == context.DeadlineExceeded { - return text, arts, fmt.Errorf("workflow: 脚本超时") - } - if ctx.Err() == context.Canceled { - return text, arts, fmt.Errorf("workflow: 脚本已取消") - } - if waitErr != nil { - return text, arts, fmt.Errorf("workflow: 脚本退出：%v\n%s", waitErr, strings.TrimSpace(stderr.String())) - } - return text, arts, nil -} - -func clipScriptOutput(stdout, stderr *bytes.Buffer) string { - text := strings.TrimSpace(stdout.String()) - errText := strings.TrimSpace(stderr.String()) - if errText != "" { - if text != "" { - text += "\n" - } - text += errText - } - if len(text) > 16000 { - text = text[:16000] + "…（超长截断）" - } - return text -} - -func prependPythonPath(env []string, dir string) []string { - for i, kv := range env { - if strings.HasPrefix(kv, "PYTHONPATH=") { - env[i] = "PYTHONPATH=" + dir + string(os.PathListSeparator) + strings.TrimPrefix(kv, "PYTHONPATH=") - return env - } - } - return append(env, "PYTHONPATH="+dir) -} - -type scriptRequest struct { - ID string `json:"id"` - Op string `json:"op"` - Name string `json:"name"` - Args json.RawMessage `json:"args"` - Question string `json:"question"` - Choices []core.HITLChoice `json:"choices"` - Agent string `json:"agent"` - Task string `json:"task"` - Content string `json:"content"` - Title string `json:"title"` - Kind string `json:"kind"` - Status string `json:"status"` - StepID string `json:"stepId"` -} - -type scriptResponse struct { - ID string `json:"id"` - OK bool `json:"ok"` - Output string `json:"output,omitempty"` - Error string `json:"error,omitempty"` -} - -func serveScript(ctx context.Context, rc RunCtx, reqR, respW *os.File) ([]*core.ArtifactRef, error) { - sc := bufio.NewScanner(reqR) - sc.Buffer(make([]byte, 64*1024), scriptMaxLine) var arts []*core.ArtifactRef - calls := 0 - for sc.Scan() { - line := strings.TrimSpace(sc.Text()) - if line == "" { - continue - } - calls++ - var req scriptRequest - if err := json.Unmarshal([]byte(line), &req); err != nil { - _ = writeScriptResponse(respW, scriptResponse{OK: false, Error: "请求不是合法 JSON"}) - continue - } - if calls > scriptMaxCalls { - _ = writeScriptResponse(respW, scriptResponse{ID: req.ID, OK: false, Error: "脚本回调超过 50 次"}) - return arts, fmt.Errorf("workflow: 脚本回调超过 50 次") - } - if ctx.Err() != nil { - return arts, ctx.Err() - } - output, got, err := dispatchScript(rc, req) + out := codeorch.RunFile(parent, sandbox, path, codeorch.Limits{ + Timeout: timeout, + MaxTimeout: scriptMaxTimeout, + MaxCalls: scriptMaxCalls, + MaxResultBytes: scriptMaxLine, + MaxModelChars: 16000, + FullDir: sandbox, + }, func(ctx context.Context, dir, scriptPath string) (*exec.Cmd, error) { + return tools.ScriptCommand(ctx, dir, scriptPath) + }, func(_ context.Context, call codeorch.Call) (string, error) { + output, got, err := dispatchScript(rc, call) arts = append(arts, got...) - if err != nil { - _ = writeScriptResponse(respW, scriptResponse{ID: req.ID, OK: false, Error: err.Error()}) - if ctx.Err() != nil { - return arts, ctx.Err() - } - continue - } - if err := writeScriptResponse(respW, scriptResponse{ID: req.ID, OK: true, Output: output}); err != nil { - return arts, err - } - } - if err := sc.Err(); err != nil { - if err == bufio.ErrTooLong { - return arts, fmt.Errorf("workflow: 单行请求超过 1MB") - } - if ctx.Err() != nil { - return arts, ctx.Err() - } - return arts, err - } - return arts, nil + return output, err + }) + return out.ModelText(), arts, out.Err } -func writeScriptResponse(w io.Writer, resp scriptResponse) error { - raw, err := json.Marshal(resp) - if err != nil { - return err +func dispatchScript(rc RunCtx, req codeorch.Call) (string, []*core.ArtifactRef, error) { + var choices []core.HITLChoice + if len(req.Choices) > 0 { + _ = json.Unmarshal(req.Choices, &choices) } - _, err = w.Write(append(raw, '\n')) - return err -} - -func dispatchScript(rc RunCtx, req scriptRequest) (string, []*core.ArtifactRef, error) { switch req.Op { case "tool.call": if _, denied := workflow.DeniedTools[req.Name]; denied { @@ -363,10 +192,10 @@ func dispatchScript(rc RunCtx, req scriptRequest) (string, []*core.ArtifactRef, return res.Output, arts, nil case "hitl.ask": card := core.HITLCard{Question: req.Question, Kind: core.HITLKindInput, AllowFreeInput: true} - if len(req.Choices) > 0 { + if len(choices) > 0 { card.Kind = core.HITLKindChoice card.AllowFreeInput = false - card.Choices = req.Choices + card.Choices = choices for i := range card.Choices { if card.Choices[i].Impact == "" { card.Choices[i].Impact = "会按这个选择继续" diff --git a/internal/settings/settings.go b/internal/settings/settings.go index 79d89f1..6dfcb91 100644 --- a/internal/settings/settings.go +++ b/internal/settings/settings.go @@ -31,13 +31,14 @@ func SystemDefaults() map[string]any { "tavily": map[string]any{"apiKeys": "", "endpoint": ""}, // 网络搜索（逗号分隔密钥池；空 = 不注册搜索工具） }, "runtime": map[string]any{ - "mock": false, // true = 全部会话走 mock runner（演示/测试） - "squad": true, // 小队编排（Agent-Squad 引擎）可用 - "entry": "direct", // direct = 现有对话循环；squad = 小队分诊入口（灰度，达标后再改默认） - "rateLimitPerMin": 0, // 租户限流（次/分；0 = 不限） - "redisAddr": "", // 限流共享后端（原生 Redis addr；空 = 进程内） - "verify": true, // 终轮确定性检查与独立验证 - "llmRetries": 2, // 直连模型在尚未输出时的额外重试次数 + "mock": false, // true = 全部会话走 mock runner（演示/测试） + "squad": true, // 小队编排（Agent-Squad 引擎）可用 + "entry": "direct", // direct = 现有对话循环；squad = 小队分诊入口（灰度，达标后再改默认） + "rateLimitPerMin": 0, // 租户限流（次/分；0 = 不限） + "redisAddr": "", // 限流共享后端（原生 Redis addr；空 = 进程内） + "verify": true, // 终轮确定性检查与独立验证 + "llmRetries": 2, // 直连模型在尚未输出时的额外重试次数 + "codeOrchestration": false, // 模型用脚本批量调用工具；默认关 }, "storage": map[string]any{ "dataDir": "./data", // 启动注入实际值（--data） @@ -85,6 +86,7 @@ func Schema() []core.SettingDesc { {Key: "runtime.redisAddr", Group: "runtime", Label: "Redis 地址（限流共享）", Type: core.SettingString, Default: "", Effect: core.EffectRestart, Description: "原生 Redis（如 127.0.0.1:6379）；空 = 进程内限流（单实例）"}, {Key: "runtime.verify", Group: "runtime", Label: "终轮结果检查", Type: core.SettingBool, Default: true, Effect: core.EffectNewRun, Description: "终轮确定性检查与独立验证"}, {Key: "runtime.llmRetries", Group: "runtime", Label: "模型重试次数", Type: core.SettingNumber, Default: 2, Effect: core.EffectNewRun, Description: "直连模型在还没有任何输出时，遇到超时、限流或服务端错误的额外重试次数。0 = 不重试。已经开始输出后不再重试"}, + {Key: "runtime.codeOrchestration", Group: "runtime", Label: "代码编排", Type: core.SettingBool, Default: false, Effect: core.EffectNewRun, Description: "允许模型用 run_tool_script 写脚本批量调用工具。默认关闭"}, {Key: "runtime.composeFinal", Group: "runtime", Label: "终轮整理回复", Type: core.SettingBool, Default: true, Effect: core.EffectNewRun, Description: "用过工具的回合结束后，再用主模型写一段直接回答用户的最终回复"}, {Key: "context.maxHistoryMessages", Group: "context", Label: "历史窗口条数", Type: core.SettingNumber, Default: 40, Effect: core.EffectNewRun}, {Key: "context.maxToolResultChars", Group: "context", Label: "工具结果上限（字符）", Type: core.SettingNumber, Default: 20000, Effect: core.EffectNewRun}, @@ -94,6 +96,7 @@ func Schema() []core.SettingDesc { {Key: "ui.cardflow", Group: "ui", Label: "长结果进右侧卡片", Type: core.SettingBool, Default: false, Effect: core.EffectNewRun, Description: "开启后报告/纪要/方案等产物型长内容经 submit_card 进收件箱，对话只留结论摘要与锚点"}, {Key: "tools.tavily.apiKeys", Group: "tools", Label: "Tavily API Keys（池）", Type: core.SettingSecret, Default: "", Effect: core.EffectRestart, Description: "逗号分隔；配置后注册 tavily_search/web_search 工具（429/401/403 自动轮换下一把）"}, {Key: "tools.tavily.endpoint", Group: "tools", Label: "Tavily 端点", Type: core.SettingString, Default: "", Effect: core.EffectRestart, Description: "空 = 官方 https://api.tavily.com/search；可指向自建代理"}, + {Key: "tools.mcp.overrides", Group: "tools", Label: "MCP 项目覆盖", Type: core.SettingString, Default: "", Effect: core.EffectNewRun, Description: "JSON 对象。键是服务器名，值可含 enabled 与 exposure（direct 或 deferred）"}, {Key: "deliverables.style", Group: "deliverables", Label: "交付物默认样式", Type: core.SettingEnum, Default: "modern", Options: []string{"modern", "formal", "minimal"}, Effect: core.EffectNewRun, Description: "modern 现代专业 / formal 公文 / minimal 极简。本轮指令、上传模板或参考文档优先于这项"}, {Key: "deliverables.accentColor", Group: "deliverables", Label: "交付物强调色", Type: core.SettingString, Default: "", Effect: core.EffectNewRun, Description: "6 位十六进制，如 1B3A4B。空 = 使用样式自带主色"}, {Key: "deliverables.templateArtifactId", Group: "deliverables", Label: "默认模板工件 ID", Type: core.SettingString, Default: "", Effect: core.EffectNewRun, Description: "填写已上传模板的工件 ID。每次运行会复制到会话 inputs/_default-template"}, diff --git a/internal/tools/codeorch_tool.go b/internal/tools/codeorch_tool.go new file mode 100644 index 0000000..c578900 --- /dev/null +++ b/internal/tools/codeorch_tool.go @@ -0,0 +1,78 @@ +package tools + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "time" + + "agent-native/runtime/codeorch" + core "agent-system-core" +) + +// RegisterCodeTool 注册 run_tool_script。开关关闭时执行会被拒绝，清单侧另行隐藏。 +func RegisterCodeTool(r *Registry) error { + return r.Register(core.ToolDesc{ + Name: "run_tool_script", + Description: "用一段 Python 依次调用已有工具。禁止再调用解释器、本工具、派发和工作流。失败时会列出已经执行的调用", + Source: core.SourceBuiltin, + Risk: "medium", + Parameters: json.RawMessage(`{"type":"object","properties":{"script":{"type":"string"},"timeout_sec":{"type":"integer"}},"required":["script"]}`), + }, execCodeScript) +} + +type toolCaller interface { + ExecuteTool(call core.ToolCallRequest) core.ToolCallResult +} + +func execCodeScript(rc ToolExecCtx, args json.RawMessage) (string, error) { + if !rc.SettingBool("runtime.codeOrchestration", false) { + return "", fmt.Errorf("代码编排未开启") + } + caller, ok := rc.(toolCaller) + if !ok { + return "", fmt.Errorf("代码编排未接入工具执行") + } + var req struct { + Script string `json:"script"` + Timeout int `json:"timeout_sec"` + } + if err := json.Unmarshal(args, &req); err != nil { + return "", err + } + if strings.TrimSpace(req.Script) == "" { + return "", fmt.Errorf("script 不能为空") + } + root := "" + if s, ok := rc.(interface{ ConversationSandbox() string }); ok { + root = s.ConversationSandbox() + } + if root == "" { + root = rc.WorkspaceRoot() + } + if root == "" { + return "", fmt.Errorf("代码编排没有会话目录") + } + lim := codeorch.Limits{FullDir: root} + if req.Timeout > 0 { + lim.Timeout = time.Duration(req.Timeout) * time.Second + } + out := codeorch.Run(context.Background(), root, req.Script, lim, nil, func(_ context.Context, call codeorch.Call) (string, error) { + if call.Op != "tool.call" { + return "", fmt.Errorf("不能调用 %s", call.Op) + } + if codeorch.Forbidden(call.Name) { + return "", fmt.Errorf("不能调用 %s", call.Name) + } + res := caller.ExecuteTool(core.ToolCallRequest{Name: call.Name, Arguments: call.Args}) + if res.Status == "error" { + return "", fmt.Errorf("%s", res.Output) + } + return res.Output, nil + }) + if out.Err != nil { + return out.ModelText(), fmt.Errorf("%s", out.ModelText()) + } + return out.ModelText(), nil +}