From 2a3b080e291f0fd061c62892860deae305139d65 Mon Sep 17 00:00:00 2001 From: Test Date: Sat, 5 Sep 2026 00:53:56 -0700 Subject: [PATCH] feat: add CanvasReasonerActivity for auto-inferring workflow connections --- action/canvas_reasoner.go | 184 ++++++++++++++++++ cmd/server/main.go | 1 + internal/routing/activity_knowledge_base.json | 75 ++++++- internal/routing/canvas_converter.go | 1 + internal/routing/canvas_validator.go | 1 + k8s/git-commit.yaml | 2 +- k8s/worker-deployment.yaml | 2 +- 7 files changed, 261 insertions(+), 5 deletions(-) create mode 100644 action/canvas_reasoner.go diff --git a/action/canvas_reasoner.go b/action/canvas_reasoner.go new file mode 100644 index 0000000..8e5bb17 --- /dev/null +++ b/action/canvas_reasoner.go @@ -0,0 +1,184 @@ +package action + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/rockliang/poimen/workflows/action/llm" + "github.com/rockliang/poimen/workflows/pkg/db" + "github.com/rockliang/poimen/workflows/statemachine" +) + +// CanvasReasonerInput infers connections between nodes using LLM reasoning +type CanvasReasonerInput struct { + Nodes []db.WorkflowNode `json:"nodes"` // Canvas nodes + Edges []db.WorkflowEdge `json:"edges"` // Existing edges + // If true, only suggest new edges; if false, redesign entire canvas + PreserveExisting bool `json:"preserve_existing,omitempty"` + AuthToken string `json:"auth_token,omitempty"` // JWT for LLM calls +} + +// CanvasReasonerOutput returns suggested edges and reasoning +type CanvasReasonerOutput struct { + SuggestedEdges []db.WorkflowEdge `json:"suggested_edges"` // New edges to add + RemovedEdges []db.WorkflowEdge `json:"removed_edges,omitempty"` // Edges to remove (if redesign) + Reasoning string `json:"reasoning"` // LLM explanation + Confidence float64 `json:"confidence"` // 0.0-1.0 +} + +// CanvasReasonerActivity uses LLM to infer connections between workflow activities +func CanvasReasonerActivity(ctx context.Context, in CanvasReasonerInput) (CanvasReasonerOutput, error) { + logger := newActivityLogger(ctx) + + output := CanvasReasonerOutput{ + SuggestedEdges: []db.WorkflowEdge{}, + } + + if len(in.Nodes) == 0 { + return output, fmt.Errorf("no nodes provided") + } + + logger.logf("info", "Analyzing canvas with %d nodes, %d edges", len(in.Nodes), len(in.Edges)) + + // Build activity descriptions for LLM context + nodeDesc := buildNodeDescriptions(in.Nodes) + edgeDesc := buildEdgeDescriptions(in.Edges) + + // Create prompt for LLM reasoning + systemPrompt := `You are a workflow automation expert. Analyze the following activities and suggest logical connections (edges) between them based on: +1. Activity input/output compatibility +2. Logical execution order +3. Data flow requirements +4. Common workflow patterns + +Respond with JSON containing: +{ + "edges": [{"source": "node-1", "target": "node-2"}, ...], + "reasoning": "explanation of why these connections make sense", + "confidence": 0.85 +}` + + userPrompt := fmt.Sprintf(`Canvas Analysis: + +Nodes: +%s + +Current Edges: +%s + +Task: %s + +Preserve existing edges and suggest only NEW edges to add. +If any existing edges don't make sense, note them but keep them unless explicitly wrong. + +Return ONLY valid JSON, no markdown code blocks.`, nodeDesc, edgeDesc, getReasoningTask(in.PreserveExisting)) + + logger.logf("info", "Calling LLM reasoning (preserve_existing=%v)", in.PreserveExisting) + + // Call LLM + client, err := llm.NewClient() + if err != nil { + return output, fmt.Errorf("failed to create LLM client: %w", err) + } + + response, err := client.CreateMessage(ctx, llm.MessageInput{ + Model: statemachine.ModelSpec{ + ModelID: "reasoning", // Use reasoning model for complex analysis + }, + SystemPrompt: systemPrompt, + Messages: []llm.MessageParam{ + { + Role: "user", + Content: userPrompt, + }, + }, + AuthToken: in.AuthToken, + }) + + if err != nil { + return output, fmt.Errorf("LLM reasoning failed: %w", err) + } + + // Parse LLM response + var reasonerResp struct { + Edges []db.WorkflowEdge `json:"edges"` + Reasoning string `json:"reasoning"` + Confidence float64 `json:"confidence"` + } + + if err := json.Unmarshal([]byte(response), &reasonerResp); err != nil { + logger.logf("warn", "Failed to parse LLM response as JSON: %v", err) + // Try to extract from response text + output.Reasoning = response + output.Confidence = 0.5 + return output, fmt.Errorf("failed to parse LLM response: %w", err) + } + + // Validate suggested edges + nodeMap := make(map[string]bool) + for _, n := range in.Nodes { + nodeMap[n.ID] = true + } + + validEdges := []db.WorkflowEdge{} + for _, edge := range reasonerResp.Edges { + if !nodeMap[edge.Source] { + logger.logf("warn", "Suggested edge references unknown source: %s", edge.Source) + continue + } + if !nodeMap[edge.Target] { + logger.logf("warn", "Suggested edge references unknown target: %s", edge.Target) + continue + } + // Don't suggest self-loops + if edge.Source == edge.Target { + logger.logf("warn", "Skipping self-loop: %s", edge.Source) + continue + } + validEdges = append(validEdges, edge) + } + + output.SuggestedEdges = validEdges + output.Reasoning = reasonerResp.Reasoning + output.Confidence = reasonerResp.Confidence + + logger.logf("info", "LLM suggested %d edges with confidence %.2f", len(validEdges), output.Confidence) + + return output, nil +} + +// buildNodeDescriptions creates readable node descriptions for LLM +func buildNodeDescriptions(nodes []db.WorkflowNode) string { + var desc string + for i, node := range nodes { + desc += fmt.Sprintf("%d. %s (type: %s)\n", i+1, node.ID, node.Type) + desc += fmt.Sprintf(" Label: %s\n", node.Label) + if node.Data != nil { + if b, err := json.MarshalIndent(node.Data, " ", " "); err == nil { + desc += fmt.Sprintf(" Config: %s\n", string(b)) + } + } + } + return desc +} + +// buildEdgeDescriptions creates readable edge descriptions for LLM +func buildEdgeDescriptions(edges []db.WorkflowEdge) string { + if len(edges) == 0 { + return "None" + } + var desc string + for i, edge := range edges { + desc += fmt.Sprintf("%d. %s → %s\n", i+1, edge.Source, edge.Target) + } + return desc +} + +// getReasoningTask returns task description based on preservation mode +func getReasoningTask(preserveExisting bool) string { + if preserveExisting { + return "Keep all existing edges and suggest ONLY NEW edges to improve workflow" + } + return "Design optimal workflow by suggesting all connections and noting any redundant edges" +} diff --git a/cmd/server/main.go b/cmd/server/main.go index 7a6cdfc..e410360 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -75,6 +75,7 @@ func main() { w.RegisterActivity(action.AssumeRoleActivity) w.RegisterActivity(action.LLMInferenceActivity) w.RegisterActivity(action.LLMBatchInferenceActivity) + w.RegisterActivity(action.CanvasReasonerActivity) var wg sync.WaitGroup errChan := make(chan error, 2) diff --git a/internal/routing/activity_knowledge_base.json b/internal/routing/activity_knowledge_base.json index 0d25a09..39a8d42 100644 --- a/internal/routing/activity_knowledge_base.json +++ b/internal/routing/activity_knowledge_base.json @@ -593,11 +593,79 @@ "dependencies": [], "notes": "Sequential processing of multiple prompts. Use for batch analysis, summarization, etc." } + }, + { + "name": "CanvasReasonerActivity", + "description": "Use LLM reasoning to infer and suggest connections between workflow activities", + "category": "workflow", + "inputs": { + "nodes": { + "type": "array", + "description": "Canvas workflow nodes to analyze", + "required": true, + "items": { + "type": "object", + "properties": { + "id": {"type": "string"}, + "type": {"type": "string"}, + "label": {"type": "string"} + } + } + }, + "edges": { + "type": "array", + "description": "Existing edges in the workflow", + "required": false, + "items": { + "type": "object" + } + }, + "preserve_existing": { + "type": "boolean", + "description": "If true, only suggest new edges; if false, redesign entire workflow", + "required": false, + "default": true + }, + "auth_token": { + "type": "string", + "description": "JWT token for authenticated LLM calls", + "required": false + } + }, + "outputs": { + "suggested_edges": { + "type": "array", + "description": "Edges suggested by LLM reasoning", + "items": { + "type": "object", + "properties": { + "source": {"type": "string"}, + "target": {"type": "string"} + } + } + }, + "reasoning": { + "type": "string", + "description": "LLM explanation of suggested connections" + }, + "confidence": { + "type": "number", + "description": "Confidence score (0.0-1.0) of the suggestions" + } + }, + "constraints": { + "defaultTimeout": "120s", + "isFlaky": true, + "recommendedRetries": 2, + "retryBackoff": 2.0, + "dependencies": [], + "notes": "Uses reasoning model to analyze workflow logic. Good for understanding data flow and connections between activities." + } } ], "metadata": { - "totalActivities": 12, - "lastUpdated": "2025-08-31T00:00:00Z", + "totalActivities": 13, + "lastUpdated": "2025-09-05T00:00:00Z", "categories": { "repository": 1, "analysis": 1, @@ -609,7 +677,8 @@ "storage": 1, "memory": 1, "authentication": 1, - "llm": 2 + "llm": 2, + "workflow": 1 } } } diff --git a/internal/routing/canvas_converter.go b/internal/routing/canvas_converter.go index 1ad68ad..c9d7562 100644 --- a/internal/routing/canvas_converter.go +++ b/internal/routing/canvas_converter.go @@ -113,6 +113,7 @@ func (cc *CanvasConverter) mapActivityType(canvasType string) string { "assume-role": "AssumeRoleActivity", "llm-inference": "LLMInferenceActivity", "llm-batch-inference": "LLMBatchInferenceActivity", + "canvas-reasoner": "CanvasReasonerActivity", } if mapped, ok := typeMap[canvasType]; ok { diff --git a/internal/routing/canvas_validator.go b/internal/routing/canvas_validator.go index 2a21d96..8a971ef 100644 --- a/internal/routing/canvas_validator.go +++ b/internal/routing/canvas_validator.go @@ -28,6 +28,7 @@ func NewCanvasValidator() *CanvasValidator { "assume-role": true, "llm-inference": true, "llm-batch-inference": true, + "canvas-reasoner": true, }, } } diff --git a/k8s/git-commit.yaml b/k8s/git-commit.yaml index 112aab1..0aac3d9 100644 --- a/k8s/git-commit.yaml +++ b/k8s/git-commit.yaml @@ -9,6 +9,6 @@ metadata: app.kubernetes.io/name: poimen app.kubernetes.io/component: orchestrator data: - GIT_COMMIT: "991c1f97" # Updated automatically by CI/CD + GIT_COMMIT: "5342bb51" # Updated automatically by CI/CD GIT_BRANCH: "main" DEPLOYMENT_DATE: "2026-09-05" diff --git a/k8s/worker-deployment.yaml b/k8s/worker-deployment.yaml index e2fe1e7..60600fe 100644 --- a/k8s/worker-deployment.yaml +++ b/k8s/worker-deployment.yaml @@ -13,7 +13,7 @@ spec: labels: app: poimen-worker annotations: - git-commit: "991c1f97" # ✅ Updated on each push, triggers rolling restart + git-commit: "5342bb51" # ✅ Updated on each push, triggers rolling restart deployment-date: "2026-09-05" spec: containers: