Feat/wdz/react interrupt (#252)

* feat: react agent support tool node rerun

* feat: export IsInterruptRerun

Author: meguminnnnnnnnn

compose/graph_run.go

@@ -395,7 +395,7 @@ func (r *runner) resolveInterruptCompletedTasks(tempInfo *interruptTempInfo, com
 				tempInfo.subGraphInterrupts[completedTasks[i].nodeKey] = info
 				continue
 			}
-			extra, ok := isInterruptRerunError(completedTasks[i].err)
+			extra, ok := IsInterruptRerunError(completedTasks[i].err)
 			if ok {
 				tempInfo.interruptRerunNodes = append(tempInfo.interruptRerunNodes, completedTasks[i].nodeKey)
 				if extra != nil {

compose/interrupt.go

@@ -47,7 +47,7 @@ func (i *interruptAndRerun) Error() string {
 	return fmt.Sprintf("interrupt and rerun: %v", i.Extra)
 }
 
-func isInterruptRerunError(err error) (any, bool) {
+func IsInterruptRerunError(err error) (any, bool) {
 	if errors.Is(err, InterruptAndRerun) {
 		return nil, true
 	}
@@ -117,7 +117,7 @@ func isInterruptError(err error) bool {
 	if info := isSubGraphInterrupt(err); info != nil {
 		return true
 	}
-	if _, ok := isInterruptRerunError(err); ok {
+	if _, ok := IsInterruptRerunError(err); ok {
 		return true
 	}
 

compose/tool_node.go

@@ -343,7 +343,7 @@ func (tn *ToolsNode) Invoke(ctx context.Context, input *schema.Message,
 	rerun := false
 	for i := 0; i < n; i++ {
 		if tasks[i].err != nil {
-			extra, ok := isInterruptRerunError(tasks[i].err)
+			extra, ok := IsInterruptRerunError(tasks[i].err)
 			if !ok {
 				return nil, fmt.Errorf("failed to invoke tool[name:%s id:%s]: %w", tasks[i].name, tasks[i].callID, tasks[i].err)
 			}
@@ -404,7 +404,7 @@ func (tn *ToolsNode) Stream(ctx context.Context, input *schema.Message,
 	// check rerun
 	for i := 0; i < n; i++ {
 		if tasks[i].err != nil {
-			extra, ok := isInterruptRerunError(tasks[i].err)
+			extra, ok := IsInterruptRerunError(tasks[i].err)
 			if !ok {
 				return nil, fmt.Errorf("failed to stream tool call %s: %w", tasks[i].callID, tasks[i].err)
 			}

flow/agent/react/react.go

@@ -19,6 +19,7 @@ package react
 import (
 	"context"
 	"io"
+	"sync"
 
 	"github.com/cloudwego/eino/components/model"
 	"github.com/cloudwego/eino/compose"
@@ -153,6 +154,8 @@ type Agent struct {
 	graphAddNodeOpts []compose.GraphAddNodeOpt
 }
 
+var registerStateOnce sync.Once
+
 // NewAgent creates a ReAct agent that feeds tool response into next round of Chat Model generation.
 //
 // IMPORTANT!! For models that don't output tool calls in the first streaming chunk (e.g. Claude)
@@ -167,6 +170,13 @@ func NewAgent(ctx context.Context, config *AgentConfig) (_ *Agent, err error) {
 		messageModifier = config.MessageModifier
 	)
 
+	registerStateOnce.Do(func() {
+		err = compose.RegisterSerializableType[state]("_eino_react_state")
+	})
+	if err != nil {
+		return
+	}
+
 	if toolCallChecker == nil {
 		toolCallChecker = firstChunkStreamToolCallChecker
 	}
@@ -208,6 +218,9 @@ func NewAgent(ctx context.Context, config *AgentConfig) (_ *Agent, err error) {
 	}
 
 	toolsNodePreHandle := func(ctx context.Context, input *schema.Message, state *state) (*schema.Message, error) {
+		if input == nil {
+			return state.Messages[len(state.Messages)-1], nil // used for rerun interrupt resume
+		}
 		state.Messages = append(state.Messages, input)
 		state.ReturnDirectlyToolCallID = getReturnDirectlyToolCallID(input, config.ToolReturnDirectly)
 		return input, nil