204 lines
8.0 KiB
Go
204 lines
8.0 KiB
Go
package runtime
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
agentllm "agent-native/runtime/llm"
|
||
"agent-system-backend/internal/squadadapter"
|
||
"agent-system-backend/internal/wfcompile"
|
||
"agent-system-backend/internal/workflow"
|
||
core "agent-system-core"
|
||
)
|
||
|
||
// workflow_compiled.go — 工作流执行器（Q3 WP2）：WorkflowDef 一律经
|
||
// internal/wfcompile 编译为 squad 图执行（legacy 逐步循环 + SpawnSubagent
|
||
// 压缩路径已退役）。P4 语义由编译入口与宿主处理器持有：步数上限、动态
|
||
// 工作流禁 script 且必须确认、模型脚本必须确认、runtime.workflowScripts
|
||
// 默认关、critic 每步 ≤3 整条 ≤6、脚本回调限额、DeniedTools。
|
||
|
||
const workflowPlannerID = "agent.sub.workflow-plan"
|
||
|
||
// workflowSource 由 runmgr 的 runCtx 实现：run 启动时拷贝的工作流定义。
|
||
type workflowSource interface {
|
||
Workflow() (workflow.Launch, bool)
|
||
}
|
||
|
||
// CompiledWorkflowRunner 以编译图执行启动时绑定的工作流（意图/演示入口）。
|
||
type CompiledWorkflowRunner struct {
|
||
// JudgeStream 注入验收模型；nil 时按轻量模型配置构造，仍缺则验证器不可用。
|
||
JudgeStream agentllm.StreamFunc
|
||
// ScriptTimeout 覆盖脚本墙钟超时（0 = 600s；1800s 上限由宿主持有）。
|
||
ScriptTimeout time.Duration
|
||
}
|
||
|
||
func (CompiledWorkflowRunner) Name() string { return "workflow" }
|
||
|
||
func (w CompiledWorkflowRunner) Execute(rc RunCtx) error {
|
||
src, ok := rc.(workflowSource)
|
||
if !ok {
|
||
return fmt.Errorf("workflow: 当前运行没有工作流定义")
|
||
}
|
||
launch, ok := src.Workflow()
|
||
if !ok {
|
||
return fmt.Errorf("workflow: 当前运行没有工作流定义")
|
||
}
|
||
out, err := w.runLaunch(rc, launch, wfcompile.RunOptions{})
|
||
if len(out.Parts) > 0 {
|
||
_ = rc.SetAssistantParts(out.Parts)
|
||
}
|
||
return err
|
||
}
|
||
|
||
// runLaunch 装配编译图执行依赖并跑一次工作流（三个 run_workflow 分支与
|
||
// 意图入口共用）。
|
||
func (w CompiledWorkflowRunner) runLaunch(rc RunCtx, launch workflow.Launch, opts wfcompile.RunOptions) (wfcompile.Outcome, error) {
|
||
deps := wfcompile.Deps{
|
||
DataDir: rc.Settings().GetString("storage.dataDir", "./data"),
|
||
Backends: squadadapter.BuildBackends(rc),
|
||
DefaultBackend: "agent-native",
|
||
JudgeStream: w.JudgeStream,
|
||
ScriptTimeout: w.ScriptTimeout,
|
||
}
|
||
if deps.JudgeStream == nil {
|
||
deps.JudgeStream = judgeStreamFor(rc)
|
||
} else {
|
||
deps.JudgeStream = GuardProjectStream(rc, "", deps.JudgeStream)
|
||
}
|
||
return wfcompile.RunCompiled(workflowHost{rc: rc}, deps, launch, workflow.Vars{
|
||
Input: currentUserText(rc),
|
||
Slots: map[string]string{},
|
||
Steps: map[string]string{},
|
||
}, opts)
|
||
}
|
||
|
||
// judgeStreamFor 按轻量模型配置构造验收流（legacy judgeStep 回退路径同款：
|
||
// 模型名取路由结果；未配置 = nil，wfcompile 按「验证器不可用」fail-closed）。
|
||
func judgeStreamFor(rc RunCtx) agentllm.StreamFunc {
|
||
route := ResolveLightweightModel(rc.Profile(), rc.Settings())
|
||
if route.Model == "" || route.Base == "" {
|
||
return nil
|
||
}
|
||
client := openRouteClient(route, rc.Settings())
|
||
model := route.Model
|
||
base := func(ctx context.Context, _ string, msgs []agentllm.Message, tools []agentllm.ToolDecl, opts agentllm.StreamOptions) <-chan agentllm.Event {
|
||
return client.Stream(ctx, model, msgs, tools, opts)
|
||
}
|
||
return GuardProjectStream(rc, "", base)
|
||
}
|
||
|
||
// planCompiledWorkflow 规划分支（run_workflow goal=）：规划子代理产出
|
||
// WorkflowDef → 动态上限编译 → 执行前强制确认（legacy planWorkflow 同款流）。
|
||
func planCompiledWorkflow(rc RunCtx, goal string, runner CompiledWorkflowRunner) (wfcompile.Outcome, error) {
|
||
sp, ok := rc.(interface {
|
||
SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error)
|
||
})
|
||
if !ok {
|
||
return wfcompile.Outcome{}, fmt.Errorf("workflow: 当前运行不能规划")
|
||
}
|
||
res, err := sp.SpawnAgent(workflowPlannerID, "规划", "把下面的目标规划成 WorkflowDef。只输出 JSON，不要完成目标本身。\n目标："+goal, "")
|
||
if err != nil {
|
||
return wfcompile.Outcome{}, err
|
||
}
|
||
text := res.Text
|
||
if text == "" {
|
||
text = res.Summary
|
||
}
|
||
def, err := wfcompile.ParseWorkflowDef(text)
|
||
if err != nil {
|
||
return wfcompile.Outcome{}, err
|
||
}
|
||
return runner.runLaunch(rc, workflow.Launch{Def: def}, wfcompile.RunOptions{
|
||
DynamicLimits: true, RequireConfirm: true,
|
||
})
|
||
}
|
||
|
||
// registeredWorkflowLaunch 组装注册工作流的启动副本（脚本正文取装载快照，
|
||
// legacy 同款：插件后续变更不影响已启动运行）。
|
||
func registeredWorkflowLaunch(rc RunCtx, def core.WorkflowDef, slots map[string]string) workflow.Launch {
|
||
launch := workflow.Launch{Def: def, Slots: slots, Scripts: map[string]string{}}
|
||
lookup, ok := rc.(interface {
|
||
WorkflowScript(id, path string) (string, bool)
|
||
})
|
||
if !ok {
|
||
return launch
|
||
}
|
||
for _, step := range def.Steps {
|
||
if step.Kind != "script" || strings.TrimSpace(step.Script) == "" {
|
||
continue
|
||
}
|
||
if body, found := lookup.WorkflowScript(def.ID, step.Script); found {
|
||
launch.Scripts[step.Script] = body
|
||
}
|
||
}
|
||
return launch
|
||
}
|
||
|
||
// modelScriptLaunch 把模型提供的脚本正文包成单步工作流（script 分支；
|
||
// 确认与开关语义由 wfcompile 宿主处理器持有）。
|
||
func modelScriptLaunch(body string) workflow.Launch {
|
||
def := core.WorkflowDef{ID: "model.script", Title: "模型脚本", Steps: []core.WorkflowStep{
|
||
{ID: "run", Title: "跑脚本", Kind: "script", Script: "model.py"},
|
||
}}
|
||
return workflow.Launch{Def: def, Scripts: map[string]string{"model.py": body}}
|
||
}
|
||
|
||
// workflowHost 把 RunCtx 适配到 wfcompile.Host（ConversationSandbox 是
|
||
// 运行侧可选能力，缺失时回退空串由宿主处理器 fail-closed）。
|
||
type workflowHost struct{ rc RunCtx }
|
||
|
||
func (h workflowHost) Context() context.Context { return h.rc.Context() }
|
||
func (h workflowHost) Run() core.Run { return h.rc.Run() }
|
||
func (h workflowHost) Settings() core.EffectiveSettings { return h.rc.Settings() }
|
||
func (h workflowHost) Emit(t core.StreamEventType, p any) error { return h.rc.Emit(t, p) }
|
||
func (h workflowHost) AskHITL(c core.HITLCard) (core.HITLResponse, error) { return h.rc.AskHITL(c) }
|
||
func (h workflowHost) UpdatePlan(p core.PlanSnapshot) error { return h.rc.UpdatePlan(p) }
|
||
func (h workflowHost) WriteArtifact(name, kind, mime string, content []byte) (core.Artifact, error) {
|
||
return h.rc.WriteArtifact(name, kind, mime, content)
|
||
}
|
||
func (h workflowHost) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult {
|
||
return h.rc.ExecuteTool(call)
|
||
}
|
||
func (h workflowHost) CreateCard(card core.ActivityCard) (core.ActivityCard, error) {
|
||
return h.rc.CreateCard(card)
|
||
}
|
||
func (h workflowHost) Tools() []core.ToolDesc { return h.rc.Tools() }
|
||
func (h workflowHost) ConversationSandbox() string {
|
||
if sb, ok := h.rc.(interface{ ConversationSandbox() string }); ok {
|
||
return sb.ConversationSandbox()
|
||
}
|
||
return ""
|
||
}
|
||
|
||
func (h workflowHost) InvalidateProjectStepOutput(ctx context.Context, stepID string) error {
|
||
invalidate, ok := h.rc.(interface {
|
||
InvalidateProjectStepOutput(context.Context, string) error
|
||
})
|
||
if !ok {
|
||
return fmt.Errorf("project workflow step invalidation unavailable")
|
||
}
|
||
return invalidate.InvalidateProjectStepOutput(ctx, stepID)
|
||
}
|
||
|
||
func (h workflowHost) BindProjectStepOutput(ctx context.Context, stepID, stepAttemptID, kind string, startedAt time.Time, output []byte) error {
|
||
binder, ok := h.rc.(interface {
|
||
BindProjectStepOutput(context.Context, string, string, string, time.Time, []byte) error
|
||
})
|
||
if !ok {
|
||
return fmt.Errorf("project workflow step binding unavailable")
|
||
}
|
||
return binder.BindProjectStepOutput(ctx, stepID, stepAttemptID, kind, startedAt, output)
|
||
}
|
||
|
||
func (h workflowHost) VerifyProjectStepOutput(ctx context.Context, stepID, stepAttemptID, kind string, output []byte) error {
|
||
verifier, ok := h.rc.(interface {
|
||
VerifyProjectStepOutput(context.Context, string, string, string, []byte) error
|
||
})
|
||
if !ok {
|
||
return fmt.Errorf("project workflow step verification unavailable")
|
||
}
|
||
return verifier.VerifyProjectStepOutput(ctx, stepID, stepAttemptID, kind, output)
|
||
}
|