zlei6/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-07-18 13:47:21 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: make await response wait for user input (#16995)

Browse Source 
### What problem does this PR solve?

Await Response incorrectly consumed the initial `sys.query` as its
input, so the first user interaction was skipped.

This change makes Await Response wait for actual user input while
preserving the existing initial-query behavior for the Begin node.

### Type of change

- [x] Bug Fix (non-breaking change which fixes an issue)
This commit is contained in:
buua436
2026-07-17 13:12:42 +08:00
committed by GitHub
parent d0ac79dc96
commit 27d091b5b6
5 changed files with 185 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
agent/component/begin.py
View File

@@ -36,6 +36,27 @@ class BeginParam(UserFillUpParam):
class Begin(UserFillUp):
component_name = "Begin"
def _merge_runtime_inputs(self, runtime_inputs):
if runtime_inputs:
return runtime_inputs
fields = self.get_input_elements()
query = self._canvas.globals.get("sys.query")
if not fields or query is None or query == "":
return {}
if isinstance(query, dict):
return {
key: value if isinstance(value, dict) else {"value": value}
for key, value in query.items()
if key in fields
}
if len(fields) == 1:
return {next(iter(fields)): {"value": query}}
return {}
def _invoke(self, **kwargs):
if self.check_if_canceled("Begin processing"):
return

30
agent/component/fillup.py
View File

@@ -21,9 +21,6 @@ from agent.component.base import ComponentParamBase, ComponentBase
from api.db.services.file_service import FileService
_INITIAL_USER_INPUT_CONSUMED_KEY = "sys.__initial_user_input_consumed__"
class UserFillUpParam(ComponentParamBase):
def __init__(self):
super().__init__()
@@ -39,32 +36,7 @@ class UserFillUp(ComponentBase):
component_name = "UserFillUp"
def _merge_runtime_inputs(self, runtime_inputs):
if runtime_inputs:
return runtime_inputs
fields = self.get_input_elements()
if not fields:
return {}
if self._canvas.globals.get(_INITIAL_USER_INPUT_CONSUMED_KEY):
return {}
query = self._canvas.globals.get("sys.query")
if query is None or query == "":
return {}
if isinstance(query, dict):
matched = {key: value if isinstance(value, dict) else {"value": value} for key, value in query.items() if key in fields}
if matched:
self._canvas.globals[_INITIAL_USER_INPUT_CONSUMED_KEY] = True
return matched
if len(fields) == 1:
field_name = next(iter(fields))
self._canvas.globals[_INITIAL_USER_INPUT_CONSUMED_KEY] = True
return {field_name: {"value": query}}
return {}
return runtime_inputs or {}
def _resolve_input_value(self, value, layout_recognize):
if isinstance(value, dict) and value.get("type", "").lower().find("file") >= 0:

40
internal/agent/canvas/interrupt_resume.go
View File

@@ -118,17 +118,6 @@ func UserFillUpNodeBody(cpnID string, params map[string]any) func(ctx context.Co
return out, nil
}
// Initial-run fast path: match the legacy Python canvas behavior
// where Begin/UserFillUp consume the current run's inputs
// directly. We only auto-consume when the node declares form
// fields; plain wait-for-user nodes (no inputs schema) still
// interrupt on first execution.
if data, ok := initialUserFillUpData(ctx, inputSpec); ok {
out := buildUserFillUpResumeOutput(cpnID, inputSpec, data)
out["__cpn_id__"] = cpnID
return out, nil
}
// First-call branch: emit the interrupt signal. The returned
// error implements error; eino's runner catches it, persists a
// checkpoint, and bubbles it up.
@@ -172,35 +161,6 @@ func buildUserFillUpResumeOutput(cpnID string, inputSpec map[string]any, data an
return out
}
func initialUserFillUpData(ctx context.Context, inputSpec map[string]any) (any, bool) {
fields, _ := inputSpec["inputs"].(map[string]any)
if len(fields) == 0 {
return nil, false
}
state, _, err := GetStateFromContext[*CanvasState](ctx)
if err != nil || state == nil {
return nil, false
}
if consumed, _ := state.Sys["__initial_user_input_consumed__"].(bool); consumed {
return nil, false
}
raw, err := state.GetVar("sys.query")
if err != nil || raw == nil {
return nil, false
}
if values, ok := raw.(map[string]any); ok {
state.Sys["__initial_user_input_consumed__"] = true
return values, true
}
text, ok := raw.(string)
if !ok || text == "" {
return nil, false
}
state.Sys["__initial_user_input_consumed__"] = true
return text, true
}
// IsInterruptError reports whether err carries an eino interrupt signal.
//
// Used by the orchestrator Driver to distinguish wait-for-user from

64
internal/agent/canvas/interrupt_resume_test.go
View File

@@ -334,74 +334,18 @@ func TestBuildUserFillUpResumeOutput_ValueFieldMirrorsResumeData(t *testing.T) {
}
}
func TestInitialUserFillUpData_UsesSysQueryWhenSchemaPresent(t *testing.T) {
func TestUserFillUpNodeBody_DoesNotConsumeSysQuery(t *testing.T) {
state := NewCanvasState("run-1", "task-1")
state.Sys["query"] = "loop"
ctx := WithState(context.Background(), state)
got, ok := initialUserFillUpData(ctx, map[string]any{
body := UserFillUpNodeBody("ufu_1", map[string]any{
"inputs": map[string]any{
"demo": map[string]any{"type": "options"},
},
})
if !ok {
t.Fatal("expected initialUserFillUpData to consume sys.query")
}
if got != "loop" {
t.Fatalf("got %v, want loop", got)
}
}
func TestInitialUserFillUpData_UsesStructuredSysQueryWhenSchemaPresent(t *testing.T) {
state := NewCanvasState("run-1", "task-1")
state.Sys["query"] = map[string]any{"kb": "da1", "query": "合同"}
ctx := WithState(context.Background(), state)
got, ok := initialUserFillUpData(ctx, map[string]any{
"inputs": map[string]any{
"kb": map[string]any{"type": "line"},
"query": map[string]any{"type": "line"},
},
})
if !ok {
t.Fatal("expected initialUserFillUpData to consume structured sys.query")
}
values, ok := got.(map[string]any)
if !ok {
t.Fatalf("got type %T, want map[string]any", got)
}
if values["kb"] != "da1" || values["query"] != "合同" {
t.Fatalf("got %#v, want kb=da1 query=合同", values)
}
}
func TestInitialUserFillUpData_SkipsWhenNoSchema(t *testing.T) {
state := NewCanvasState("run-1", "task-1")
state.Sys["query"] = "loop"
ctx := WithState(context.Background(), state)
if got, ok := initialUserFillUpData(ctx, map[string]any{}); ok || got != nil {
t.Fatalf("expected no auto-consume without schema, got (%v, %v)", got, ok)
}
}
func TestInitialUserFillUpData_ConsumesOnlyOnce(t *testing.T) {
state := NewCanvasState("run-1", "task-1")
state.Sys["query"] = "loop"
ctx := WithState(context.Background(), state)
spec := map[string]any{
"inputs": map[string]any{
"demo": map[string]any{"type": "options"},
},
}
got, ok := initialUserFillUpData(ctx, spec)
if !ok || got != "loop" {
t.Fatalf("first auto-consume = (%v, %v), want (loop, true)", got, ok)
}
got2, ok2 := initialUserFillUpData(ctx, spec)
if ok2 || got2 != nil {
t.Fatalf("second auto-consume = (%v, %v), want (nil, false)", got2, ok2)
if _, err := body(ctx, nil); err == nil {
t.Fatal("UserFillUp should interrupt instead of consuming sys.query")
}
}

202
internal/service/agent_run_e2e_test.go
View File

@@ -189,7 +189,11 @@ func TestRunAgent_RealCanvas_BeginMessage(t *testing.T) {
}
makeCanvasWithDSL(t, "canvas-hello", "user-1", "tenant-1", "v-hello", dsl)
svc := NewAgentService()
tracker, mr := newRunTrackerForTest(t, 30*24*time.Hour)
cpClient := redis.NewClient(&redis.Options{Addr: mr.Addr()})
t.Cleanup(func() { _ = cpClient.Close() })
cp := canvas.NewRedisCheckPointStoreWithClient(cpClient, 30*24*time.Hour)
svc := NewAgentServiceWithOptions(cp, nil, tracker)
events, err := svc.RunAgent(
context.Background(),
"user-1",
@@ -590,10 +594,11 @@ func TestRunAgent_RealCanvas_GroupedParallelOuterFollower(t *testing.T) {
// TestRunAgent_AllFixture_LoopInterruptResume drives the real all.json
// fixture through the production RunAgent interrupt/resume path:
//
// 1. First input "loop" is consumed directly by UserFillUp:Menu,
// routes through Switch:Route into Loop:InputUntil1, and pauses at
// UserFillUp:LoopInput.
// 2. Second input "1" resumes LoopInput, satisfies Switch:LoopCheck's
// 1. The first input reaches UserFillUp:Menu and pauses for the menu
// selection.
// 2. The second input "loop" resumes the menu, routes through
// Switch:Route into Loop:InputUntil1, and pauses at UserFillUp:LoopInput.
// 3. The third input "1" resumes LoopInput, satisfies Switch:LoopCheck's
// exit condition, and the workflow finishes at Message:LoopDone.
func TestRunAgent_AllFixture_LoopInterruptResume(t *testing.T) {
testDB := setupServiceTestDB(t)
@@ -652,8 +657,8 @@ func TestRunAgent_AllFixture_LoopInterruptResume(t *testing.T) {
if waiting1[0].CpnID == "" {
t.Fatal("run 1: waiting cpn_id is empty")
}
if waiting1[0].Tips != "请输入任意内容，输入 `1` 则退出循环：" {
t.Fatalf("run 1: waiting tips = %q, want %q", waiting1[0].Tips, "请输入任意内容，输入 `1` 则退出循环：")
if waiting1[0].Tips != "请选择要演示的模块：" {
t.Fatalf("run 1: waiting tips = %q, want menu prompt", waiting1[0].Tips)
}
if len(waiting1[0].Inputs) == 0 {
t.Fatal("run 1: waiting inputs is empty")
@@ -675,7 +680,7 @@ func TestRunAgent_AllFixture_LoopInterruptResume(t *testing.T) {
"canvas-all",
"session-all-loop",
"",
"1",
"loop",
)
if err != nil {
t.Fatalf("RunAgent run 2: %v", err)
@@ -684,14 +689,31 @@ func TestRunAgent_AllFixture_LoopInterruptResume(t *testing.T) {
if len(errs2) > 0 {
t.Fatalf("run 2: unexpected error events: %+v", errs2)
}
if len(waiting2) > 0 {
t.Fatalf("run 2: did not expect another waiting_for_user event, got %+v", waiting2)
if len(waiting2) != 1 {
t.Fatalf("run 2: expected loop input prompt, got %+v", waiting2)
}
if waiting2[0].Tips != "请输入任意内容，输入 `1` 则退出循环：" {
t.Fatalf("run 2: waiting tips = %q, want loop prompt", waiting2[0].Tips)
}
if !done2 {
t.Error("run 2: missing done event")
t.Error("run 2: expected done terminator after waiting_for_user")
}
if len(messages2) != 0 {
t.Fatalf("run 2: expected no message before loop input, got %d", len(messages2))
}
events3, err := svc.RunAgent(
context.Background(), "user-1", "canvas-all", "session-all-loop", "", "1",
)
if err != nil {
t.Fatalf("RunAgent run 3: %v", err)
}
messages2, waiting3, errs3, done3 := drainAgentEvents(t, events3)
if len(errs3) > 0 || len(waiting3) > 0 || !done3 {
t.Fatalf("run 3: errs=%+v waiting=%+v done=%v", errs3, waiting3, done3)
}
if len(messages2) != 1 {
t.Fatalf("run 2: expected 1 message event after resume, got %d", len(messages2))
t.Fatalf("run 3: expected 1 message event after resume, got %d", len(messages2))
}
if !strings.Contains(messages2[0].Content, "循环结束") {
t.Errorf("run 2: Content = %q, want substring %q", messages2[0].Content, "循环结束")
@@ -735,7 +757,7 @@ func TestRunAgent_AllFixture_LoopInterruptResume_MultiTurn(t *testing.T) {
svc := NewAgentServiceWithOptions(cp, nil, tracker)
sessionID := "session-all-loop-multi"
inputs := []string{"loop", "aaa", "bbb", "1"}
inputs := []string{"loop", "loop", "aaa", "bbb", "1"}
var allMessages []canvas.MessageEvent
for i, input := range inputs {
@@ -756,8 +778,15 @@ func TestRunAgent_AllFixture_LoopInterruptResume_MultiTurn(t *testing.T) {
}
allMessages = append(allMessages, messages...)
switch input {
case "loop", "aaa", "bbb":
switch i {
case 0:
if len(waiting) != 1 || waiting[0].Tips != "请选择要演示的模块：" {
t.Fatalf("run %d (%q): expected menu prompt, got %+v", i+1, input, waiting)
}
if !done {
t.Errorf("run %d (%q): missing done terminator after waiting_for_user", i+1, input)
}
case 1, 2, 3:
if len(waiting) != 1 {
t.Fatalf("run %d (%q): expected 1 waiting_for_user event, got %+v", i+1, input, waiting)
}
@@ -767,7 +796,7 @@ func TestRunAgent_AllFixture_LoopInterruptResume_MultiTurn(t *testing.T) {
if !done {
t.Errorf("run %d (%q): missing done terminator after waiting_for_user", i+1, input)
}
case "1":
case 4:
if len(waiting) != 0 {
t.Fatalf("run %d (%q): did not expect another waiting_for_user event, got %+v", i+1, input, waiting)
}
@@ -847,6 +876,9 @@ func TestRunAgent_AllFixture_IterationFormatsItems(t *testing.T) {
if len(waiting1) != 1 {
t.Fatalf("run 1: expected 1 waiting_for_user event, got %+v", waiting1)
}
if waiting1[0].Tips != "请选择要演示的模块：" {
t.Fatalf("run 1: waiting tips = %q, want menu prompt", waiting1[0].Tips)
}
// SSE channel always closes with the `done` terminator,
// even when the run paused for user input.
if !done1 {
@@ -859,7 +891,7 @@ func TestRunAgent_AllFixture_IterationFormatsItems(t *testing.T) {
"canvas-all-iteration",
sessionID,
"",
"a,b,c,d,e",
"iteration",
)
if err != nil {
t.Fatalf("RunAgent run 2: %v", err)
@@ -868,14 +900,31 @@ func TestRunAgent_AllFixture_IterationFormatsItems(t *testing.T) {
if len(errs2) > 0 {
t.Fatalf("run 2: unexpected error events: %+v", errs2)
}
if len(waiting2) != 0 {
t.Fatalf("run 2: did not expect another waiting_for_user event, got %+v", waiting2)
if len(waiting2) != 1 {
t.Fatalf("run 2: expected iteration input prompt, got %+v", waiting2)
}
if waiting2[0].Tips == "请选择要演示的模块：" {
t.Fatalf("run 2: menu was not resumed: %+v", waiting2)
}
if !done2 {
t.Fatal("run 2: missing done event")
t.Fatal("run 2: expected done terminator after waiting_for_user")
}
if len(messages2) != 1 {
t.Fatalf("run 2: expected 1 message event, got %d", len(messages2))
if len(messages2) != 0 {
t.Fatalf("run 2: expected no message before iteration input, got %d", len(messages2))
}
events3, err := svc.RunAgent(
context.Background(), "user-1", "canvas-all-iteration", sessionID, "", "a,b,c,d,e",
)
if err != nil {
t.Fatalf("RunAgent run 3: %v", err)
}
messages2, waiting3, errs3, done3 := drainAgentEvents(t, events3)
if len(errs3) > 0 {
t.Fatalf("run 3: unexpected error events: %+v", errs3)
}
if len(waiting3) != 0 || !done3 {
t.Fatalf("run 3: waiting=%+v done=%v", waiting3, done3)
}
content := messages2[0].Content
// Go renders []any{"a","b","c","d","e"} via fmt.Sprintf("%v", ...)
@@ -916,7 +965,11 @@ func TestRunAgent_AllFixture_VarAssigner(t *testing.T) {
}
makeCanvasWithDSL(t, "canvas-all-var-assigner", "user-1", "tenant-1", "v-all-var-assigner", dsl)
svc := NewAgentService()
tracker, mr := newRunTrackerForTest(t, 30*24*time.Hour)
cpClient := redis.NewClient(&redis.Options{Addr: mr.Addr()})
t.Cleanup(func() { _ = cpClient.Close() })
cp := canvas.NewRedisCheckPointStoreWithClient(cpClient, 30*24*time.Hour)
svc := NewAgentServiceWithOptions(cp, nil, tracker)
events, err := svc.RunAgent(
context.Background(),
"user-1",
@@ -932,14 +985,32 @@ func TestRunAgent_AllFixture_VarAssigner(t *testing.T) {
if len(errs) > 0 {
t.Fatalf("unexpected error events: %+v", errs)
}
if len(waiting) > 0 {
t.Fatalf("did not expect waiting_for_user events: %+v", waiting)
if len(waiting) != 1 || waiting[0].Tips != "请选择要演示的模块：" {
t.Fatalf("expected menu waiting_for_user event: %+v", waiting)
}
if !done {
t.Fatal("missing done event")
t.Fatal("missing done event after menu prompt")
}
if len(messages) != 0 {
t.Fatalf("expected no message before menu resume, got %d", len(messages))
}
events, err = svc.RunAgent(
context.Background(),
"user-1",
"canvas-all-var-assigner",
"session-all-var-assigner",
"",
"var_assigner",
)
if err != nil {
t.Fatalf("RunAgent resume: %v", err)
}
messages, waiting, errs, done = drainAgentEvents(t, events)
if len(errs) > 0 || len(waiting) > 0 || !done {
t.Fatalf("resume: messages=%+v waiting=%+v errs=%+v done=%v", messages, waiting, errs, done)
}
if len(messages) != 1 {
t.Fatalf("expected 1 message event, got %d", len(messages))
t.Fatalf("expected 1 message event after menu resume, got %d", len(messages))
}
if messages[0].Content != "env.counter=1" {
t.Fatalf("Content = %q, want %q", messages[0].Content, "env.counter=1")
@@ -978,7 +1049,11 @@ func TestRunAgent_AllFixture_DataOps(t *testing.T) {
}
makeCanvasWithDSL(t, "canvas-all-data-ops", "user-1", "tenant-1", "v-all-data-ops", dsl)
svc := NewAgentService()
tracker, mr := newRunTrackerForTest(t, 30*24*time.Hour)
cpClient := redis.NewClient(&redis.Options{Addr: mr.Addr()})
t.Cleanup(func() { _ = cpClient.Close() })
cp := canvas.NewRedisCheckPointStoreWithClient(cpClient, 30*24*time.Hour)
svc := NewAgentServiceWithOptions(cp, nil, tracker)
events, err := svc.RunAgent(
context.Background(),
"user-1",
@@ -994,14 +1069,32 @@ func TestRunAgent_AllFixture_DataOps(t *testing.T) {
if len(errs) > 0 {
t.Fatalf("unexpected error events: %+v", errs)
}
if len(waiting) > 0 {
t.Fatalf("did not expect waiting_for_user events: %+v", waiting)
if len(waiting) != 1 || waiting[0].Tips != "请选择要演示的模块：" {
t.Fatalf("expected menu waiting_for_user event: %+v", waiting)
}
if !done {
t.Fatal("missing done event")
t.Fatal("missing done event after menu prompt")
}
if len(messages) != 0 {
t.Fatalf("expected no message before menu resume, got %d", len(messages))
}
events, err = svc.RunAgent(
context.Background(),
"user-1",
"canvas-all-data-ops",
"session-all-data-ops",
"",
"data_ops",
)
if err != nil {
t.Fatalf("RunAgent resume: %v", err)
}
messages, waiting, errs, done = drainAgentEvents(t, events)
if len(errs) > 0 || len(waiting) > 0 || !done {
t.Fatalf("resume: messages=%+v waiting=%+v errs=%+v done=%v", messages, waiting, errs, done)
}
if len(messages) != 1 {
t.Fatalf("expected 1 message event, got %d (%v)", len(messages), messages)
t.Fatalf("expected 1 message event after menu resume, got %d (%v)", len(messages), messages)
}
t.Logf("data_ops message content:\n%s", messages[0].Content)
if !strings.Contains(messages[0].Content, "ListOperations Sort desc") {
@@ -1095,10 +1188,11 @@ func TestRunAgent_RealCanvas_CompileFails(t *testing.T) {
// TestRunAgent_AllFixture_CategorizeResume drives the categorize branch
// of all.json through the real checkpoint-backed interrupt/resume path:
//
// 1. First input "categorize" is consumed by UserFillUp:Menu, which routes
// through Switch:Route into UserFillUp:CateInput and pauses there.
// 2. Second input "hello" must resume CateInput itself, not be re-consumed
// by the menu as a fresh branch selection.
// 1. First input "categorize" pauses at UserFillUp:Menu.
// 2. Second input "categorize" resumes the menu, routes through Switch:Route
// into UserFillUp:CateInput, and pauses there.
// 3. Third input "hello" must resume CateInput itself, not be re-consumed by
// the menu as a fresh branch selection.
//
// This specifically covers the buildRunFunc fix that clears sys.query /
// Invoke(query) on resume. Without that fix, the resumed payload can be
@@ -1161,8 +1255,8 @@ func TestRunAgent_AllFixture_CategorizeResume(t *testing.T) {
if len(waiting1) != 1 {
t.Fatalf("run 1: expected 1 waiting_for_user, got %d (errs=%+v)", len(waiting1), errs1)
}
if !strings.Contains(waiting1[0].Tips, "分类") {
t.Fatalf("run 1: waiting tips = %q, want categorize prompt", waiting1[0].Tips)
if waiting1[0].Tips != "请选择要演示的模块：" {
t.Fatalf("run 1: waiting tips = %q, want menu prompt", waiting1[0].Tips)
}
if !done1 {
t.Error("run 1: expected done terminator after waiting_for_user")
@@ -1174,7 +1268,7 @@ func TestRunAgent_AllFixture_CategorizeResume(t *testing.T) {
"canvas-all-categorize",
"session-all-categorize",
"",
"hello",
"categorize",
)
if err != nil {
t.Fatalf("RunAgent run 2: %v", err)
@@ -1187,17 +1281,39 @@ func TestRunAgent_AllFixture_CategorizeResume(t *testing.T) {
for i, m := range messages2 {
t.Logf(" run2 msg[%d]: %q", i, m.Content)
}
if len(waiting2) != 0 {
t.Fatalf("run 2: did not expect another waiting_for_user event, got %+v", waiting2)
if len(waiting2) != 1 {
t.Fatalf("run 2: expected categorize prompt, got %+v", waiting2)
}
if !strings.Contains(waiting2[0].Tips, "分类") {
t.Fatalf("run 2: waiting tips = %q, want categorize prompt", waiting2[0].Tips)
}
if len(errs2) != 0 {
t.Fatalf("run 2: expected no downstream errors, got %+v", errs2)
}
if !done2 {
t.Fatal("run 2: expected done event after successful categorize branch completion")
t.Fatal("run 2: expected done terminator after waiting_for_user")
}
if len(messages2) != 0 {
t.Fatalf("run 2: expected no message before CateInput resume, got %d", len(messages2))
}
events3, err := svc.RunAgent(
context.Background(),
"user-1",
"canvas-all-categorize",
"session-all-categorize",
"",
"hello",
)
if err != nil {
t.Fatalf("RunAgent run 3: %v", err)
}
messages2, waiting3, errs3, done3 := drainAgentEvents(t, events3)
if len(errs3) != 0 || len(waiting3) != 0 || !done3 {
t.Fatalf("run 3: messages=%+v waiting=%+v errs=%+v done=%v", messages2, waiting3, errs3, done3)
}
if len(messages2) != 1 {
t.Fatalf("run 2: expected 1 message event, got %d", len(messages2))
t.Fatalf("run 3: expected 1 message event, got %d", len(messages2))
}
if !strings.Contains(messages2[0].Content, "分类结果=Retrieval -> Retrieval") {
t.Fatalf("run 2: message = %q, want categorize retrieval branch output", messages2[0].Content)