package runtime import ( "encoding/json" "fmt" "strings" "time" "agent-native/runtime/gates" agentllm "agent-native/runtime/llm" "agent-system-backend/internal/workflow" core "agent-system-core" ) const workflowPlannerID = "agent.sub.workflow-plan" // WorkflowRunner 执行启动时拷贝下来的工作流定义。 type WorkflowRunner struct { // JudgeStream 注入验收模型。nil 时按轻量模型构造客户端。 JudgeStream agentllm.StreamFunc // ScriptTimeout 覆盖脚本墙钟超时。0 表示 600 秒,超过 1800 秒会被截断。 ScriptTimeout time.Duration } func (WorkflowRunner) Name() string { return "workflow" } type workflowSource interface { Workflow() (workflow.Launch, bool) } type workflowOutcome struct { Summary string Parts []core.Part Artifacts []*core.ArtifactRef } // planWorkflow 让规划子代理写出步骤,校验通过后必须经用户确认才执行。 func planWorkflow(rc RunCtx, goal string, runner WorkflowRunner) (workflowOutcome, error) { sp, ok := rc.(interface { SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) }) if !ok { return workflowOutcome{}, fmt.Errorf("workflow: 当前运行不能规划") } res, err := sp.SpawnAgent(workflowPlannerID, "规划", "把下面的目标规划成 WorkflowDef。只输出 JSON,不要完成目标本身。\n目标:"+goal, "") if err != nil { return workflowOutcome{}, err } text := res.Text if text == "" { text = res.Summary } def, err := parseWorkflowDef(text) if err != nil { return workflowOutcome{}, err } known := map[string]bool{} for _, tool := range rc.Tools() { known[tool.Name] = true } if err := workflow.Validate(def, workflow.DynamicLimits(), func(name string) bool { return known[name] }); err != nil { return workflowOutcome{}, err } resp, err := rc.AskHITL(core.HITLCard{ Question: planPreview(def), Kind: core.HITLKindConfirm, AllowFreeInput: false, Choices: []core.HITLChoice{ {ID: "yes", Label: "执行", Description: "按这些步骤做", Impact: "会调用工具并可能写入工件"}, {ID: "no", Label: "取消", Description: "不执行", Impact: "不会改动任何东西"}, }, }) if err != nil { return workflowOutcome{}, err } if resp.ChoiceID != "yes" { return workflowOutcome{Summary: "未确认,未执行"}, nil } return runner.run(rc, workflow.Launch{Def: def}) } func parseWorkflowDef(raw string) (core.WorkflowDef, error) { s := stripFenceBlock(raw) start := strings.Index(s, "{") end := strings.LastIndex(s, "}") if start < 0 || end < start { return core.WorkflowDef{}, fmt.Errorf("workflow: 规划结果无法解析") } var def core.WorkflowDef if err := json.Unmarshal([]byte(s[start:end+1]), &def); err != nil { return core.WorkflowDef{}, fmt.Errorf("workflow: 规划结果无法解析") } if strings.TrimSpace(def.ID) == "" { def.ID = "dynamic" } if strings.TrimSpace(def.Title) == "" { def.Title = "临时工作流" } return def, nil } func planPreview(def core.WorkflowDef) string { var b strings.Builder fmt.Fprintf(&b, "按这个计划执行「%s」?\n", def.Title) for i, step := range def.Steps { fmt.Fprintf(&b, "%d. %s(%s)\n", i+1, step.Title, step.Kind) } return strings.TrimRight(b.String(), "\n") } func (w WorkflowRunner) Execute(rc RunCtx) error { src, ok := rc.(workflowSource) if !ok { return fmt.Errorf("workflow: 当前运行没有工作流定义") } launch, ok := src.Workflow() if !ok { return fmt.Errorf("workflow: 当前运行没有工作流定义") } res, err := w.run(rc, launch) if len(res.Parts) > 0 { _ = rc.SetAssistantParts(res.Parts) } return err } func (w WorkflowRunner) run(rc RunCtx, launch workflow.Launch) (workflowOutcome, error) { def := launch.Def vars := workflow.Vars{ Input: currentUserText(rc), Slots: map[string]string{}, Steps: map[string]string{}, } for k, v := range launch.Slots { vars.Slots[k] = v } for _, missing := range launch.Missing { if rc.Context().Err() != nil { return workflowOutcome{}, rc.Context().Err() } resp, err := rc.AskHITL(core.HITLCard{ Question: missing.Question, Kind: core.HITLKindInput, AllowFreeInput: true, }) if err != nil { return workflowOutcome{}, err } answer := strings.TrimSpace(resp.FreeText) if answer == "" { answer = resp.ChoiceID } vars.Slots[missing.Key] = answer } plan := core.PlanSnapshot{Title: def.Title, Steps: make([]core.PlanStep, len(def.Steps))} for i, step := range def.Steps { plan.Steps[i] = core.PlanStep{ID: step.ID, Text: step.Title, Status: "pending"} } if err := rc.UpdatePlan(plan); err != nil { return workflowOutcome{}, err } wholeLeft := def.MaxRetries if wholeLeft <= 0 || wholeLeft > workflow.MaxWorkflowRetries { wholeLeft = workflow.MaxWorkflowRetries } var drops []string var arts []*core.ArtifactRef var prev string statuses := make([]string, len(def.Steps)) for i := range def.Steps { if rc.Context().Err() != nil { return outcome(def, statuses, arts, drops, ""), rc.Context().Err() } step := def.Steps[i] plan.Steps[i].Status = "in_progress" if err := rc.UpdatePlan(plan); err != nil { return workflowOutcome{}, err } rendered, err := workflow.RenderStep(step, vars) if err != nil { return outcome(def, statuses, arts, drops, err.Error()), err } output, stepArts, reason, err := w.runStepWithCritic(rc, def, rendered, vars, prev, launch.Scripts, &wholeLeft, &drops) arts = append(arts, stepArts...) if err != nil { statuses[i] = "failed" plan.Steps[i].Status = "skipped" _ = rc.UpdatePlan(plan) res := outcome(def, statuses, arts, drops, reason) return res, err } vars.Steps[step.ID] = output prev = output statuses[i] = "done" plan.Steps[i].Status = "done" if err := rc.UpdatePlan(plan); err != nil { return workflowOutcome{}, err } } res := outcome(def, statuses, arts, drops, "") return res, nil } func (w WorkflowRunner) runStepWithCritic(rc RunCtx, def core.WorkflowDef, step core.WorkflowStep, vars workflow.Vars, prev string, scripts map[string]string, wholeLeft *int, drops *[]string) (string, []*core.ArtifactRef, string, error) { stepRetries := 1 if step.Critic != nil && step.Critic.MaxRetries > 0 { stepRetries = step.Critic.MaxRetries } if stepRetries > workflow.MaxStepRetries { stepRetries = workflow.MaxStepRetries } used := 0 note := "" var arts []*core.ArtifactRef for { current := step if note != "" && current.Prompt != "" { current.Prompt = current.Prompt + "\n\n未通过验收:" + note } else if note != "" { current.Prompt = "未通过验收:" + note } output, got, err := w.runWorkflowStep(rc, current, vars, prev, scripts) arts = append(arts, got...) if err != nil { return output, arts, err.Error(), err } if step.Critic == nil || strings.TrimSpace(step.Critic.Criteria) == "" { return output, arts, "", nil } verdict := w.judgeStep(rc, def.Title, step.Critic.Criteria, output) if verdict.Passed { return output, arts, "", nil } why := verdict.Reason if verdict.NextAction != "" { why += "。下一步:" + verdict.NextAction } *drops = append(*drops, step.ID+":"+why) if used >= stepRetries || *wholeLeft <= 0 { return output, arts, "步骤 " + step.Title + " 未通过验收:" + why, fmt.Errorf("workflow: 步骤 %s 未通过验收", step.ID) } used++ *wholeLeft-- note = why } } func (w WorkflowRunner) runWorkflowStep(rc RunCtx, step core.WorkflowStep, vars workflow.Vars, prev string, scripts map[string]string) (string, []*core.ArtifactRef, error) { switch step.Kind { case "prompt": return runPromptStep(rc, step) case "tool": res := rc.ExecuteTool(core.ToolCallRequest{Name: step.Tool, Arguments: step.Args}) if res.Status == "error" { return res.Output, nil, fmt.Errorf("workflow: 工具 %s 失败:%s", step.Tool, res.Output) } var arts []*core.ArtifactRef if res.Artifact != nil { arts = append(arts, &core.ArtifactRef{ArtifactID: res.Artifact.ID, Name: res.Artifact.Name, Kind: res.Artifact.Kind}) } return res.Output, arts, nil case "hitl": card := core.HITLCard{Question: step.Title, Kind: core.HITLKindInput, AllowFreeInput: true} if step.Hitl != nil { card.Question = step.Hitl.Question card.Kind = step.Hitl.Kind card.Choices = step.Hitl.Choices card.AllowFreeInput = step.Hitl.AllowFreeInput card.FreeInputHint = step.Hitl.FreeInputHint } resp, err := rc.AskHITL(card) if err != nil { return "", nil, err } if strings.TrimSpace(resp.FreeText) != "" { return strings.TrimSpace(resp.FreeText), nil, nil } return resp.ChoiceID, nil, nil case "artifact": content := step.Prompt if strings.TrimSpace(content) == "" { content = prev } name := step.Artifact if name == "" { name = step.ID + ".md" } art, err := rc.WriteArtifact(name, "document", "text/markdown", []byte(content)) if err != nil { return "", nil, err } return string(art.ID), []*core.ArtifactRef{{ArtifactID: art.ID, Name: art.Name, Kind: art.Kind}}, nil case "card": card := core.ActivityCard{Title: step.Title, Kind: core.CardKindResult} if step.Card != nil { card.Kind = step.Card.Kind card.Title = step.Card.Title card.Tone = step.Card.Tone card.Body = step.Card.Body card.Actions = step.Card.Actions } created, err := rc.CreateCard(card) if err != nil { return "", nil, err } return string(created.ID), nil, nil case "script": return executeScriptStep(rc, w, step, scripts) default: return "", nil, fmt.Errorf("workflow: 未知步骤类型 %s", step.Kind) } } func runPromptStep(rc RunCtx, step core.WorkflowStep) (string, []*core.ArtifactRef, error) { agent := step.Agent if strings.TrimSpace(agent) == "" { agent = "general" } var res core.AgentResultPayload var err error if sp, ok := rc.(interface { SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) }); ok { res, err = sp.SpawnAgent(subagentProfileID(agent), step.Title, step.Prompt, "") } else { res, err = rc.SpawnSubagent(step.Title, step.Prompt, "") } if err != nil { return "", nil, err } text := res.Text if text == "" { text = res.Summary } return text, artifactsIn(rc, text), nil } func artifactsIn(rc RunCtx, text string) []*core.ArtifactRef { var out []*core.ArtifactRef seen := map[string]bool{} for _, id := range gates.ArtifactIDPattern.FindAllString(text, -1) { if seen[id] { continue } art, _, ok := rc.Artifact(id) if !ok { continue } seen[id] = true out = append(out, &core.ArtifactRef{ArtifactID: art.ID, Name: art.Name, Kind: art.Kind}) } return out } func (w WorkflowRunner) judgeStep(rc RunCtx, goal, criteria, output string) gates.Verdict { stream := w.JudgeStream model := "verify" if stream == nil { route := ResolveLightweightModel(rc.Profile(), rc.Settings()) if route.Model == "" || route.Base == "" { return gates.Verdict{Passed: false, Reason: "验证器不可用", NextAction: "检查轻量模型配置"} } model = route.Model client := openLLMClient(route.Base, route.Key) stream = client.Stream } prompt := fmt.Sprintf("你是验收员。目标:%s\n验收标准:%s\n这一步的产出:\n%s\n标准里要求出现的句子,产出里必须逐字包含,否则 passed 必须为 false,nextAction 写出要补上的原句。只输出一个 JSON 对象,字段为 passed、reason、nextAction。", goal, criteria, output) var b strings.Builder for ev := range stream(rc.Context(), model, []agentllm.Message{{Role: agentllm.RoleUser, Content: prompt}}, nil, agentllm.StreamOptions{}) { switch ev.Kind { case agentllm.KindTextDelta: b.WriteString(ev.Text) case agentllm.KindError: return gates.Verdict{Passed: false, Reason: "验证器不可用", NextAction: "稍后再试"} case agentllm.KindMessageEnd: if b.Len() == 0 { b.WriteString(ev.Message.Content) } } } verdict, err := gates.ParseVerdict(b.String()) if err != nil { return gates.Verdict{Passed: false, Reason: "无法解析", NextAction: "按验收标准重做"} } return verdict } func outcome(def core.WorkflowDef, statuses []string, arts []*core.ArtifactRef, drops []string, stopped string) workflowOutcome { var b strings.Builder title := def.Title if title == "" { title = def.ID } if stopped != "" { fmt.Fprintf(&b, "工作流「%s」停止:%s\n", title, stopped) } else { fmt.Fprintf(&b, "工作流「%s」完成。\n", title) } for i, step := range def.Steps { state := "pending" if i < len(statuses) && statuses[i] != "" { state = statuses[i] } fmt.Fprintf(&b, "- %s:%s\n", step.Title, state) } if len(arts) > 0 { b.WriteString("工件:") for i, art := range arts { if i > 0 { b.WriteString("、") } b.WriteString(string(art.ArtifactID)) } b.WriteString("\n") } if len(drops) > 0 { b.WriteString("已丢弃:") b.WriteString(strings.Join(drops, ";")) b.WriteString("\n") } parts := []core.Part{{Type: core.PartText, Text: strings.TrimRight(b.String(), "\n")}} for _, art := range arts { parts = append(parts, core.Part{Type: core.PartArtifactRef, Artifact: art}) } return workflowOutcome{Summary: parts[0].Text, Parts: parts, Artifacts: arts} } // stripFenceBlock 剥模型输出的 ``` 围栏(工作流定义解析用;与门禁域无关,本地持有)。 func stripFenceBlock(s string) string { s = strings.TrimSpace(s) if !strings.HasPrefix(s, "```") { return s } s = strings.TrimPrefix(s, "```") if i := strings.IndexByte(s, '\n'); i >= 0 { s = s[i+1:] } return strings.TrimSpace(strings.TrimSuffix(strings.TrimSpace(s), "```")) }