Add workflow runtime execution and condition validation
This commit is contained in:
mlogclub
@@ -0,0 +1,408 @@
|
||||
package workflow
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"agent-desk/internal/ai"
|
||||
"agent-desk/internal/ai/runtime/internal/impl/retrievers"
|
||||
"agent-desk/internal/ai/workflow/dsl"
|
||||
workflowregistry "agent-desk/internal/ai/workflow/registry"
|
||||
"agent-desk/internal/models"
|
||||
"agent-desk/internal/pkg/utils"
|
||||
)
|
||||
|
||||
const maxWorkflowSteps = 128
|
||||
|
||||
type Input struct {
|
||||
Definition dsl.Definition
|
||||
Conversation models.Conversation
|
||||
UserMessage models.Message
|
||||
AIAgent models.AIAgent
|
||||
AIConfig models.AIConfig
|
||||
}
|
||||
|
||||
type Result struct {
|
||||
Status string
|
||||
ReplyText string
|
||||
NodePath []string
|
||||
PromptTokens int
|
||||
CompletionTokens int
|
||||
RetrieverCount int
|
||||
TraceData string
|
||||
}
|
||||
|
||||
type Executor struct{}
|
||||
|
||||
func NewExecutor() *Executor {
|
||||
return &Executor{}
|
||||
}
|
||||
|
||||
type runState struct {
|
||||
input Input
|
||||
nodesByID map[string]dsl.Node
|
||||
outgoing map[string][]dsl.Edge
|
||||
vars map[string]map[string]any
|
||||
result Result
|
||||
}
|
||||
|
||||
func (e *Executor) Execute(ctx context.Context, input Input) (*Result, error) {
|
||||
state := newRunState(input)
|
||||
currentID := strings.TrimSpace(input.Definition.EntryNodeID)
|
||||
if currentID == "" {
|
||||
return nil, fmt.Errorf("workflow entry node is required")
|
||||
}
|
||||
for step := 0; step < maxWorkflowSteps; step++ {
|
||||
node, ok := state.nodesByID[currentID]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("workflow node does not exist: %s", currentID)
|
||||
}
|
||||
state.result.NodePath = append(state.result.NodePath, node.ID)
|
||||
if err := e.executeNode(ctx, state, node); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if node.Type == workflowregistry.NodeTypeEnd {
|
||||
state.result.Status = "completed"
|
||||
return &state.result, nil
|
||||
}
|
||||
nextID, ok, err := state.nextNodeID(node.ID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !ok {
|
||||
state.result.Status = "completed"
|
||||
return &state.result, nil
|
||||
}
|
||||
currentID = nextID
|
||||
}
|
||||
return nil, fmt.Errorf("workflow exceeded max steps")
|
||||
}
|
||||
|
||||
func newRunState(input Input) *runState {
|
||||
state := &runState{
|
||||
input: input,
|
||||
nodesByID: make(map[string]dsl.Node, len(input.Definition.Nodes)),
|
||||
outgoing: make(map[string][]dsl.Edge),
|
||||
vars: make(map[string]map[string]any),
|
||||
result: Result{
|
||||
Status: "started",
|
||||
NodePath: make([]string, 0),
|
||||
},
|
||||
}
|
||||
for _, node := range input.Definition.Nodes {
|
||||
node.ID = strings.TrimSpace(node.ID)
|
||||
node.Type = strings.TrimSpace(node.Type)
|
||||
if node.ID != "" {
|
||||
state.nodesByID[node.ID] = node
|
||||
}
|
||||
}
|
||||
for _, edge := range input.Definition.Edges {
|
||||
state.outgoing[edge.Source] = append(state.outgoing[edge.Source], edge)
|
||||
}
|
||||
return state
|
||||
}
|
||||
|
||||
func (e *Executor) executeNode(ctx context.Context, state *runState, node dsl.Node) error {
|
||||
switch node.Type {
|
||||
case workflowregistry.NodeTypeStart:
|
||||
state.setNodeVars(node.ID, map[string]any{
|
||||
"conversationId": state.input.Conversation.ID,
|
||||
"messageId": state.input.UserMessage.ID,
|
||||
"aiAgentId": state.input.AIAgent.ID,
|
||||
"userMessage": strings.TrimSpace(state.input.UserMessage.Content),
|
||||
"knowledgeBaseIds": utils.SplitInt64s(state.input.AIAgent.KnowledgeIDs),
|
||||
"conversationState": state.input.Conversation.Status,
|
||||
})
|
||||
case workflowregistry.NodeTypeKnowledgeRetrieve:
|
||||
return e.executeKnowledgeRetrieve(ctx, state, node)
|
||||
case workflowregistry.NodeTypeAnswerabilityGate:
|
||||
return e.executeAnswerabilityGate(state, node)
|
||||
case workflowregistry.NodeTypeCondition:
|
||||
state.setNodeVars(node.ID, map[string]any{"matched": true})
|
||||
case workflowregistry.NodeTypeLLMReply:
|
||||
return e.executeLLMReply(ctx, state, node)
|
||||
case workflowregistry.NodeTypeSendReply:
|
||||
replyText := strings.TrimSpace(toString(state.resolveInput(node, "replyText")))
|
||||
state.result.ReplyText = replyText
|
||||
state.setNodeVars(node.ID, map[string]any{
|
||||
"sent": replyText != "",
|
||||
"replyMessageId": int64(0),
|
||||
})
|
||||
case workflowregistry.NodeTypeHandoffToHuman:
|
||||
reason := strings.TrimSpace(toString(state.resolveInput(node, "reason")))
|
||||
replyText := strings.TrimSpace(readStringConfig(node.Config, "replyText"))
|
||||
if replyText == "" {
|
||||
replyText = "已为你转接人工客服，请稍候。"
|
||||
}
|
||||
state.result.ReplyText = replyText
|
||||
state.setNodeVars(node.ID, map[string]any{
|
||||
"handoffId": int64(0),
|
||||
"reason": reason,
|
||||
})
|
||||
case workflowregistry.NodeTypeEnd:
|
||||
state.setNodeVars(node.ID, map[string]any{"status": "completed"})
|
||||
default:
|
||||
return fmt.Errorf("unsupported workflow node type: %s", node.Type)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *Executor) executeKnowledgeRetrieve(ctx context.Context, state *runState, node dsl.Node) error {
|
||||
query := strings.TrimSpace(toString(state.resolveInput(node, "query")))
|
||||
retriever := retrievers.NewKnowledgeRetriever(state.input.AIAgent)
|
||||
result, err := retriever.RetrieveContext(ctx, query)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
items := make([]map[string]any, 0, len(result.ContextResults))
|
||||
for _, item := range result.ContextResults {
|
||||
items = append(items, map[string]any{
|
||||
"knowledgeBaseId": item.KnowledgeBaseID,
|
||||
"documentId": item.DocumentID,
|
||||
"chunkId": item.ChunkID,
|
||||
"content": item.Content,
|
||||
"score": item.Score,
|
||||
})
|
||||
}
|
||||
state.result.RetrieverCount = len(result.Hits)
|
||||
state.setNodeVars(node.ID, map[string]any{
|
||||
"items": items,
|
||||
"summary": result.ContextText,
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *Executor) executeAnswerabilityGate(state *runState, node dsl.Node) error {
|
||||
items := state.resolveInput(node, "knowledgeItems")
|
||||
answerability := "unanswerable"
|
||||
reason := "no retrieved knowledge items"
|
||||
if hasItems(items) {
|
||||
answerability = "answerable"
|
||||
reason = "retrieved knowledge items are available"
|
||||
}
|
||||
state.setNodeVars(node.ID, map[string]any{
|
||||
"answerability": answerability,
|
||||
"reason": reason,
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *Executor) executeLLMReply(ctx context.Context, state *runState, node dsl.Node) error {
|
||||
if staticReply := strings.TrimSpace(readStringConfig(node.Config, "staticReply")); staticReply != "" {
|
||||
state.setNodeVars(node.ID, map[string]any{"replyText": staticReply})
|
||||
return nil
|
||||
}
|
||||
userPrompt := strings.TrimSpace(toString(state.resolveInput(node, "userMessage")))
|
||||
if userPrompt == "" {
|
||||
userPrompt = strings.TrimSpace(state.input.UserMessage.Content)
|
||||
}
|
||||
knowledgeItems := toString(state.resolveInput(node, "knowledgeItems"))
|
||||
systemPrompt := strings.TrimSpace(state.input.AIAgent.SystemPrompt)
|
||||
if prompt := strings.TrimSpace(readStringConfig(node.Config, "prompt")); prompt != "" {
|
||||
systemPrompt = strings.TrimSpace(systemPrompt + "\n\n" + prompt)
|
||||
}
|
||||
if knowledgeItems != "" {
|
||||
userPrompt = userPrompt + "\n\nKnowledge context:\n" + knowledgeItems
|
||||
}
|
||||
result, err := ai.LLM.ChatWithConfig(ctx, state.input.AIConfig, systemPrompt, userPrompt)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
state.result.PromptTokens += result.PromptTokens
|
||||
state.result.CompletionTokens += result.CompletionTokens
|
||||
state.setNodeVars(node.ID, map[string]any{"replyText": result.Content})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *runState) nextNodeID(sourceNodeID string) (string, bool, error) {
|
||||
edges := s.outgoing[sourceNodeID]
|
||||
if len(edges) == 0 {
|
||||
return "", false, nil
|
||||
}
|
||||
for _, edge := range edges {
|
||||
if edge.Condition == nil {
|
||||
continue
|
||||
}
|
||||
matched, err := s.evaluateCondition(edge.Condition)
|
||||
if err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
if matched {
|
||||
return strings.TrimSpace(edge.Target), true, nil
|
||||
}
|
||||
}
|
||||
for _, edge := range edges {
|
||||
if edge.Condition == nil {
|
||||
return strings.TrimSpace(edge.Target), true, nil
|
||||
}
|
||||
}
|
||||
return "", false, nil
|
||||
}
|
||||
|
||||
func (s *runState) evaluateCondition(condition *dsl.Condition) (bool, error) {
|
||||
if condition == nil {
|
||||
return true, nil
|
||||
}
|
||||
left := s.resolveSelector(condition.Left)
|
||||
operator := strings.TrimSpace(condition.Operator)
|
||||
if operator == "" && strings.TrimSpace(condition.Expression) != "" {
|
||||
return false, fmt.Errorf("free-form workflow condition expressions are not supported")
|
||||
}
|
||||
switch operator {
|
||||
case "eq", "equals":
|
||||
return compareString(left, condition.Right) == 0, nil
|
||||
case "neq", "not_equals":
|
||||
return compareString(left, condition.Right) != 0, nil
|
||||
case "contains":
|
||||
return strings.Contains(toString(left), toString(condition.Right)), nil
|
||||
case "exists":
|
||||
return exists(left), nil
|
||||
case "not_exists":
|
||||
return !exists(left), nil
|
||||
case "truthy", "is_true":
|
||||
return truthy(left), nil
|
||||
case "falsy", "is_false":
|
||||
return !truthy(left), nil
|
||||
case "gt":
|
||||
return compareNumber(left, condition.Right) > 0, nil
|
||||
case "gte":
|
||||
return compareNumber(left, condition.Right) >= 0, nil
|
||||
case "lt":
|
||||
return compareNumber(left, condition.Right) < 0, nil
|
||||
case "lte":
|
||||
return compareNumber(left, condition.Right) <= 0, nil
|
||||
default:
|
||||
return false, fmt.Errorf("unsupported workflow condition operator: %s", operator)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *runState) setNodeVars(nodeID string, values map[string]any) {
|
||||
s.vars[nodeID] = values
|
||||
}
|
||||
|
||||
func (s *runState) resolveInput(node dsl.Node, inputName string) any {
|
||||
selector, ok := node.Inputs[inputName]
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
return s.resolveSelector(&selector)
|
||||
}
|
||||
|
||||
func (s *runState) resolveSelector(selector *dsl.VariableSelector) any {
|
||||
if selector == nil {
|
||||
return nil
|
||||
}
|
||||
fields := s.vars[strings.TrimSpace(selector.NodeID)]
|
||||
if fields == nil {
|
||||
return nil
|
||||
}
|
||||
return fields[strings.TrimSpace(selector.Field)]
|
||||
}
|
||||
|
||||
func readStringConfig(raw json.RawMessage, key string) string {
|
||||
if len(raw) == 0 {
|
||||
return ""
|
||||
}
|
||||
var cfg map[string]any
|
||||
if err := json.Unmarshal(raw, &cfg); err != nil {
|
||||
return ""
|
||||
}
|
||||
return toString(cfg[key])
|
||||
}
|
||||
|
||||
func compareString(left any, right any) int {
|
||||
return strings.Compare(toString(left), toString(right))
|
||||
}
|
||||
|
||||
func compareNumber(left any, right any) int {
|
||||
leftNum := toFloat(left)
|
||||
rightNum := toFloat(right)
|
||||
switch {
|
||||
case leftNum > rightNum:
|
||||
return 1
|
||||
case leftNum < rightNum:
|
||||
return -1
|
||||
default:
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
func toString(value any) string {
|
||||
switch v := value.(type) {
|
||||
case nil:
|
||||
return ""
|
||||
case string:
|
||||
return v
|
||||
case fmt.Stringer:
|
||||
return v.String()
|
||||
case []map[string]any:
|
||||
buf, _ := json.Marshal(v)
|
||||
return string(buf)
|
||||
default:
|
||||
return strings.TrimSpace(fmt.Sprint(v))
|
||||
}
|
||||
}
|
||||
|
||||
func toFloat(value any) float64 {
|
||||
switch v := value.(type) {
|
||||
case int:
|
||||
return float64(v)
|
||||
case int64:
|
||||
return float64(v)
|
||||
case float64:
|
||||
return v
|
||||
case float32:
|
||||
return float64(v)
|
||||
case json.Number:
|
||||
f, _ := v.Float64()
|
||||
return f
|
||||
case string:
|
||||
f, _ := strconv.ParseFloat(strings.TrimSpace(v), 64)
|
||||
return f
|
||||
default:
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
func truthy(value any) bool {
|
||||
switch v := value.(type) {
|
||||
case nil:
|
||||
return false
|
||||
case bool:
|
||||
return v
|
||||
case string:
|
||||
normalized := strings.ToLower(strings.TrimSpace(v))
|
||||
return normalized != "" && normalized != "false" && normalized != "0"
|
||||
default:
|
||||
return !reflect.ValueOf(value).IsZero()
|
||||
}
|
||||
}
|
||||
|
||||
func exists(value any) bool {
|
||||
if value == nil {
|
||||
return false
|
||||
}
|
||||
switch v := value.(type) {
|
||||
case string:
|
||||
return strings.TrimSpace(v) != ""
|
||||
default:
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
func hasItems(value any) bool {
|
||||
if value == nil {
|
||||
return false
|
||||
}
|
||||
rv := reflect.ValueOf(value)
|
||||
switch rv.Kind() {
|
||||
case reflect.Array, reflect.Slice, reflect.Map:
|
||||
return rv.Len() > 0
|
||||
default:
|
||||
return exists(value)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
package workflow
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"agent-desk/internal/ai/workflow/dsl"
|
||||
workflowregistry "agent-desk/internal/ai/workflow/registry"
|
||||
"agent-desk/internal/models"
|
||||
)
|
||||
|
||||
func TestExecutorRoutesByConditionEdge(t *testing.T) {
|
||||
executor := NewExecutor()
|
||||
result, err := executor.Execute(context.Background(), Input{
|
||||
Definition: conditionalReplyDefinition(),
|
||||
UserMessage: models.Message{
|
||||
Content: "vip",
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("execute workflow: %v", err)
|
||||
}
|
||||
if result.ReplyText != "VIP reply" {
|
||||
t.Fatalf("unexpected reply: %q", result.ReplyText)
|
||||
}
|
||||
assertPath(t, result.NodePath, []string{"start_1", "condition_1", "vip_reply", "send_vip", "end_1"})
|
||||
}
|
||||
|
||||
func TestExecutorUsesDefaultEdgeWhenConditionDoesNotMatch(t *testing.T) {
|
||||
executor := NewExecutor()
|
||||
result, err := executor.Execute(context.Background(), Input{
|
||||
Definition: conditionalReplyDefinition(),
|
||||
UserMessage: models.Message{
|
||||
Content: "normal",
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("execute workflow: %v", err)
|
||||
}
|
||||
if result.ReplyText != "Normal reply" {
|
||||
t.Fatalf("unexpected reply: %q", result.ReplyText)
|
||||
}
|
||||
assertPath(t, result.NodePath, []string{"start_1", "condition_1", "normal_reply", "send_normal", "end_1"})
|
||||
}
|
||||
|
||||
func conditionalReplyDefinition() dsl.Definition {
|
||||
return dsl.Definition{
|
||||
SchemaVersion: 1,
|
||||
EntryNodeID: "start_1",
|
||||
Nodes: []dsl.Node{
|
||||
{ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"},
|
||||
{ID: "condition_1", Type: workflowregistry.NodeTypeCondition, Name: "Route"},
|
||||
{ID: "vip_reply", Type: workflowregistry.NodeTypeLLMReply, Name: "VIP", Config: []byte(`{"staticReply":"VIP reply"}`)},
|
||||
{ID: "normal_reply", Type: workflowregistry.NodeTypeLLMReply, Name: "Normal", Config: []byte(`{"staticReply":"Normal reply"}`)},
|
||||
{ID: "send_vip", Type: workflowregistry.NodeTypeSendReply, Name: "Send VIP", Inputs: map[string]dsl.VariableSelector{
|
||||
"replyText": {NodeID: "vip_reply", Field: "replyText"},
|
||||
}},
|
||||
{ID: "send_normal", Type: workflowregistry.NodeTypeSendReply, Name: "Send Normal", Inputs: map[string]dsl.VariableSelector{
|
||||
"replyText": {NodeID: "normal_reply", Field: "replyText"},
|
||||
}},
|
||||
{ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"},
|
||||
},
|
||||
Edges: []dsl.Edge{
|
||||
{ID: "edge_start_condition", Source: "start_1", Target: "condition_1"},
|
||||
{
|
||||
ID: "edge_condition_vip",
|
||||
Source: "condition_1",
|
||||
Target: "vip_reply",
|
||||
Condition: &dsl.Condition{
|
||||
Left: &dsl.VariableSelector{NodeID: "start_1", Field: "userMessage"},
|
||||
Operator: "eq",
|
||||
Right: "vip",
|
||||
},
|
||||
},
|
||||
{ID: "edge_condition_default", Source: "condition_1", Target: "normal_reply"},
|
||||
{ID: "edge_vip_send", Source: "vip_reply", Target: "send_vip"},
|
||||
{ID: "edge_normal_send", Source: "normal_reply", Target: "send_normal"},
|
||||
{ID: "edge_send_vip_end", Source: "send_vip", Target: "end_1"},
|
||||
{ID: "edge_send_normal_end", Source: "send_normal", Target: "end_1"},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func assertPath(t *testing.T, got []string, want []string) {
|
||||
t.Helper()
|
||||
if len(got) != len(want) {
|
||||
t.Fatalf("unexpected path length: got %#v want %#v", got, want)
|
||||
}
|
||||
for i := range want {
|
||||
if got[i] != want[i] {
|
||||
t.Fatalf("unexpected path: got %#v want %#v", got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user