智能体工作流的人工输入¶
Supported in ADKPython v2.0.0Go v2.0.0
能够在数据输入、决策验证或操作授权等环节请求人工输入，是许多智能体驱动工作流的重要组成部分。ADK 中基于图的工作流可以包含专门为获取人工输入而设计的人机交互（HITL）节点。这些节点不需要运行人工智能（AI）模型，从而使输入过程更具可预测性和可靠性。
开始使用¶
你可以使用 RequestInput 类和一个文本提示在图中实现人工输入节点。以下代码示例展示了如何在 Workflow 图中添加人工输入节点：
from google.adk.events import RequestInput
from google.adk import Workflow
def step1(): # 人工输入步骤
yield RequestInput(message="Enter a number:")
def step2(node_input):
return node_input * 2
root_agent = Workflow(
name="root_agent",
edges=[('START', step1, step2)],
)
在此代码示例中，step1 会暂停智能体的执行，直到系统收到用户的输入。一旦系统收到用户的输入，该输入就会被传递到下一个节点。
在 ADK Go v2.0.0 中，HITL 图节点通过 workflow.NewEmittingFunctionNode 和 workflow.ResumeOrRequestInput 构建。这是 Python 中 RequestInput 节点的直接等价物：
- 在首次执行时，
workflow.ResumeOrRequestInput发出一个session.RequestInput事件（以Event.RequestedInput的形式呈现）并返回ErrNodeInterrupted，从而暂停工作流。 - 在人工回复后，节点会从顶部重新调用（
RerunOnResume: &true），ResumeOrRequestInput返回回复内容，该内容通过event.Output作为类型化输入流向下一个节点。
// newGraphHITLWorkflow demonstrates a graph HITL node using
// workflow.NewEmittingFunctionNode and workflow.ResumeOrRequestInput.
//
// This is the Go equivalent of the Python RequestInput node:
//
// def step1(): # Human input step
// yield RequestInput(message="Enter a number:")
//
// def step2(node_input):
// return node_input * 2
//
// root_agent = Workflow(
// name="root_agent",
// edges=[('START', step1, step2)],
// )
//
// On the first pass, step1Node emits a RequestInput event and pauses the
// workflow (ErrNodeInterrupted). After the human replies, the node is re-run
// and ResumeOrRequestInput returns the reply, which flows as typed input to
// step2Node via event.Output.
func newGraphHITLWorkflow() (agent.Agent, error) {
rerun := true
// step1Node: pauses for human input on the first pass, returns the
// human's reply on resume. workflow.ResumeOrRequestInput handles both
// phases — no manual re-entry bookkeeping needed.
step1Node := workflow.NewEmittingFunctionNode[any, string]("step1",
func(ctx agent.Context, _ any, emit func(*session.Event) error) (string, error) {
reply, err := workflow.ResumeOrRequestInput(ctx, emit, session.RequestInput{
InterruptID: "enter_number",
Message: "Enter a number:",
})
if err != nil {
// ErrNodeInterrupted on first pass — workflow pauses here.
return "", err
}
// On resume, reply is the human's text response.
number, _ := reply.(string)
return number, nil
},
workflow.NodeConfig{RerunOnResume: &rerun},
)
// step2Node: receives the human's input as its typed string input via
// event.Output and doubles the number.
step2Node := workflow.NewFunctionNode("step2",
func(_ agent.Context, input string) (string, error) {
return fmt.Sprintf("You entered: %s (doubled: %s%s)", input, input, input), nil
},
workflow.NodeConfig{},
)
return workflowagent.New(workflowagent.Config{
Name: "root_agent",
Description: "Pauses for a number from the user, then doubles it.",
Edges: workflow.Chain(workflow.Start, step1Node, step2Node),
})
}
配置选项¶
人工输入节点可以使用 RequestInput 类，支持以下配置选项：
message： 向用户提供的说明人工输入请求的文本。payload： 作为人工输入请求一部分的结构化数据。response_schema： 人工响应必须遵循的数据结构。
注意：响应 schema 输入限制
对于 response_schema 设置，RequestInput 类不会自动重新格式化人工响应以匹配指定的数据结构。人工响应必须以指定格式提供。为了获得更好的用户体验，可以考虑提供用户界面来收集结构化数据，或使用智能体节点将非结构化数据转换为所需的格式。
session.RequestInput 携带以下字段，它们与 Python 的 RequestInput 参数直接对应：
InterruptID（string）：此暂停点的唯一标识符。使用稳定的前缀加 UUID 来避免跨工作流运行时的冲突。等同于 Python 中的隐式中断 ID。Message（string）：显示给用户的人类可读提示。等同于 Python 的message参数。Payload（any）：可选的结构化数据，随提示一起发送，以便客户端渲染额外的上下文。等同于 Python 的payload参数。
workflow.NodeConfig.RerunOnResume 控制恢复时的行为：
&true：节点主体从顶部重新执行；ResumeOrRequestInput在第二次执行时返回人工回复。使用ResumeOrRequestInput的节点必须设置此项。&false或nil（叶子节点默认值）：回复被路由到节点的后继节点作为输入，跳过被中断的节点。
注意：来自客户端的结构化响应
ADK Go 不会自动解析或验证人工回复负载的结构。如果你的工作流需要结构化反馈，请在前端界面或下游智能体节点中对响应进行验证，然后再执行后续操作。
人工输入示例¶
以下代码示例展示了更详细的人工输入请求。
请求带消息和负载的输入¶
以下代码示例展示了如何在工作流节点中构建 RequestInput 对象，包括 负载 和 响应 schema。在此示例中，ActivitiesList 预期由一个组成活动列表的智能体节点完成，而 get_user_feedback() 节点向用户请求反馈。
class ActivitiesList(BaseModel):
"""行程应为每个活动的字典列表。每个活动包含名称和描述"""
itinerary: List[Dict[str, str]]
class UserFeedback(BaseModel):
"""用户预期的响应结构。"""
user_response: str
async def get_user_feedback(node_input: ActivitiesList):
"""
获取用户对智能体初始行程的意见，以便扩展、更改列表或退出循环
"""
message = (
f"""
这是你推荐的基础行程：\n{node_input}\n\n
这些项目中哪些吸引了你（如果有）？
"""
)
yield RequestInput(
message=message,
payload=node_input,
response_schema=UserFeedback,
)
以下代码示例展示了一个三节点图：一个构建器节点生成结构化行程，一个 HITL 节点将其作为 Payload 与提示一起发送，最后一个节点根据用户的反馈执行操作。Payload 字段允许客户端在用户回复之前渲染完整的行程：
// ItineraryItem represents a single activity in a travel plan.
type ItineraryItem struct {
Name string `json:"name"`
Description string `json:"description"`
}
// newItineraryReviewWorkflow demonstrates a graph HITL node that sends a
// structured payload alongside the input prompt so the client can render
// additional context for the user. This mirrors Python's:
//
// async def get_user_feedback(node_input: ActivitiesList):
// yield RequestInput(
// message="Which items appeal to you?",
// payload=node_input,
// response_schema=UserFeedback,
// )
func newItineraryReviewWorkflow() (agent.Agent, error) {
rerun := true
// buildItineraryNode: generates an itinerary and passes it to the HITL
// node as its typed output via event.Output.
buildItineraryNode := workflow.NewFunctionNode("build_itinerary",
func(_ agent.Context, _ any) ([]ItineraryItem, error) {
return []ItineraryItem{
{Name: "Eiffel Tower", Description: "Iconic iron lattice tower."},
{Name: "Louvre Museum", Description: "World's largest art museum."},
{Name: "Seine River Cruise", Description: "Scenic boat tour of Paris."},
}, nil
},
workflow.NodeConfig{},
)
// reviewNode: sends the itinerary as payload alongside the prompt so the
// client can display it. On resume, the human's selection is returned.
reviewNode := workflow.NewEmittingFunctionNode[[]ItineraryItem, string]("get_user_feedback",
func(ctx agent.Context, itinerary []ItineraryItem, emit func(*session.Event) error) (string, error) {
reply, err := workflow.ResumeOrRequestInput(ctx, emit, session.RequestInput{
InterruptID: "itinerary_review",
Message: fmt.Sprintf("Here is your recommended itinerary (%d activities). Which items appeal to you?", len(itinerary)),
Payload: itinerary, // structured payload rendered by the client
})
if err != nil {
// ErrNodeInterrupted on first pass — workflow pauses here.
return "", err
}
feedback, _ := reply.(string)
return feedback, nil
},
workflow.NodeConfig{RerunOnResume: &rerun},
)
// finalNode: receives the user's feedback and produces a confirmation.
finalNode := workflow.NewFunctionNode("finalize",
func(_ agent.Context, feedback string) (string, error) {
return fmt.Sprintf("Itinerary finalised with your feedback: %q", feedback), nil
},
workflow.NodeConfig{},
)
return workflowagent.New(workflowagent.Config{
Name: "concierge_workflow",
Description: "Builds an itinerary, asks the user for feedback, then finalises.",
Edges: workflow.Chain(workflow.Start, buildItineraryNode, reviewNode, finalNode),
})
}
工具确认：LLM 智能体中的审批提示¶
工具确认是一种独立的、LLM 智能体级别的机制，用于是/否审批提示。与图 HITL 节点不同，工具确认在 llmagent 工具函数内部工作，而不是作为独立的图节点。当你希望 LLM 智能体在执行特定工具调用之前暂停并请求审批时，这个机制非常有用。
以下代码示例展示了如何在工作流节点中构建 RequestInput 对象，包括 响应 schema：
在 functiontool.Config 中设置 RequireConfirmation: true 可在工具执行前进行静态的是/否审批，或者从工具内部调用 ctx.RequestConfirmation 来设置自定义提示消息：
// DoubleNumberArgs holds the input for the doubleNumber tool.
type DoubleNumberArgs struct {
Number int `json:"number" jsonschema:"The number to double."`
}
// DoubleNumberResults holds the output of the doubleNumber tool.
type DoubleNumberResults struct {
Result int `json:"result"`
}
// doubleNumber is a tool that doubles the given number.
// Because RequireConfirmation is true, the framework automatically pauses
// execution and emits an "adk_request_confirmation" event to the client before
// running the tool. The client must reply with a FunctionResponse confirming
// or denying the action.
func doubleNumber(_ agent.Context, args DoubleNumberArgs) (DoubleNumberResults, error) {
return DoubleNumberResults{Result: args.Number * 2}, nil
}
// newSimpleHITLAgent creates an LLM agent with a tool that always requires
// user confirmation before it executes (tool-confirmation pattern).
func newSimpleHITLAgent(ctx context.Context) (agent.Agent, error) {
model, err := gemini.NewModel(ctx, modelName, &genai.ClientConfig{})
if err != nil {
return nil, fmt.Errorf("failed to create model: %w", err)
}
doubleNumberTool, err := functiontool.New(
functiontool.Config{
Name: "double_number",
Description: "Doubles the given number. Requires user approval before running.",
RequireConfirmation: true,
},
doubleNumber,
)
if err != nil {
return nil, fmt.Errorf("failed to create tool: %w", err)
}
return llmagent.New(llmagent.Config{
Name: "double_number_agent",
Model: model,
Instruction: "You are a helpful assistant. When asked to double a number, use the double_number tool.",
Tools: []tool.Tool{doubleNumberTool},
})
}
使用自定义提示和手动重入处理：
// BookFlightArgs holds the input for the bookFlight tool.
type BookFlightArgs struct {
Origin string `json:"origin" jsonschema:"Departure airport code."`
Destination string `json:"destination" jsonschema:"Arrival airport code."`
Date string `json:"date" jsonschema:"Travel date in YYYY-MM-DD format."`
}
// BookFlightResults holds the outcome of the bookFlight tool.
type BookFlightResults struct {
Status string `json:"status"`
ConfirmNumber string `json:"confirm_number,omitempty"`
}
// bookFlight is a tool that pauses for human approval before completing a
// booking (tool-confirmation pattern with a custom hint message).
func bookFlight(ctx agent.Context, args BookFlightArgs) (BookFlightResults, error) {
if confirmation := ctx.ToolConfirmation(); confirmation != nil {
if !confirmation.Confirmed {
return BookFlightResults{Status: "Booking cancelled by user."}, nil
}
return BookFlightResults{
Status: "Booking confirmed.",
ConfirmNumber: "FLT-20251031",
}, nil
}
hint := fmt.Sprintf(
"The agent wants to book a flight from %s to %s on %s. Do you approve?",
args.Origin, args.Destination, args.Date,
)
if err := ctx.RequestConfirmation(hint, nil); err != nil {
return BookFlightResults{}, fmt.Errorf("failed to request confirmation: %w", err)
}
return BookFlightResults{Status: "Awaiting user approval."}, nil
}
// newHITLWithHintAgent creates an LLM agent whose bookFlight tool manually
// requests confirmation with a descriptive hint (tool-confirmation pattern).
func newHITLWithHintAgent(ctx context.Context) (agent.Agent, error) {
model, err := gemini.NewModel(ctx, modelName, &genai.ClientConfig{})
if err != nil {
return nil, fmt.Errorf("failed to create model: %w", err)
}
bookFlightTool, err := functiontool.New(
functiontool.Config{
Name: "book_flight",
Description: "Books a flight between two airports on a given date.",
},
bookFlight,
)
if err != nil {
return nil, fmt.Errorf("failed to create tool: %w", err)
}
return llmagent.New(llmagent.Config{
Name: "flight_booking_agent",
Model: model,
Instruction: "You are a flight booking assistant. Help the user book flights.",
Tools: []tool.Tool{bookFlightTool},
})
}