17｜Runner 与事件流：看见 Agent 的执行过程 - eino教程

17Runner 与事件流看见 Agent 的执行过程《Eino 实战用 Go 构建 AI 应用与智能体》· 第 17 篇关键词Runner、Query、Run、AsyncIterator、AgentEvent、流式事件、判空前置条件已完成 16从流程到 Agent。第 16 篇的事件流里三行输出就把 ReAct 循环讲清楚了。这就是事件流的价值它让 Agent 的内部过程变得可见。对用户而言可见意味着“知道它在干什么”对开发者而言可见意味着“出错时知道哪一步坏了”。这一篇系统讲怎么消费事件流。一、Runner 是执行入口runner:adk.NewRunner(ctx,adk.RunnerConfig{Agent:agent,EnableStreaming:true,// CheckPointStore: store, // 需要中断恢复时提供第 20 篇})iter:runner.Query(ctx,工单 T-1001 什么状态)RunnerConfig的三个字段字段作用Agent要运行的 AgentEnableStreaming是否以流式方式产生模型输出事件CheckPointStore检查点存储支持中断后恢复两个入口方法方法输入用途Query(ctx, 问题)单个字符串便捷方法包装成用户消息Run(ctx, messages)[]*schema.Message传入完整消息列表适合多轮返回的都是*AsyncIterator[*AgentEvent]用Next()逐个取事件for{event,ok:iter.Next()if!ok{break// 迭代结束}// 处理 event}ok false表示事件流结束不是错误。错误通过event.Err传递。二、AgentEvent 的结构typeAgentEventstruct{AgentNamestring// 产生事件的 AgentRunPath[]RunStep// 从根 Agent 到当前事件的路径Output*AgentOutput// 消息输出Action*AgentAction// 动作退出、中断、转交Errerror// 错误}访问前必须判空Output和Action都可能为nil——不同事件类型携带的字段不同。中断事件里Output就是空的。这一点在第 20 篇的实测中直接踩到了不判空会 panic。// 安全的取用方式funcmessageOutput(event*adk.AgentEvent)*adk.MessageVariant{ifeventnil||event.Outputnil{returnnil}returnevent.Output.MessageOutput}三层指针逐层判空Output→MessageOutput→Message或MessageStream。先用 error 分支ifevent.Err!nil{// 必须先处理错误continue}忽略event.Err会让失败伪装成“正常结束但没输出”。建议把错误分支放在最前面。三、MessageVariant消息输出typeMessageVariantstruct{IsStreamingboolMessage*schema.Message MessageStream*schema.StreamReader[*schema.Message]Role schema.RoleType// assistant 或 toolToolNamestring// 工具结果时非空}关键字段字段用途Role区分模型输出assistant和工具结果toolToolName工具结果时给出工具名流式下这是识别来源的唯一方式IsStreaming内容在MessageStream还是Message四、流式与非流式的区别EnableStreaming会改变事件的内容形态。下面这段程序对比两种模式已实机运行 非流式EnableStreamingfalse [非流] roleassistant tool content [非流] roletool toolget_ticket content工单 T-1001 状态已分派 [非流] roleassistant tool contentT-1001 已分派给网络支持组。 流式EnableStreamingtrue [流式] roleassistant tool content [非流] roletool toolget_ticket content工单 T-1001 状态已分派 [流式] roleassistant tool contentT-1001 已分派给网络支持组。注意第二行无论哪种模式工具结果都是非流的。原因不难理解工具执行是应用侧的行为没有“逐字生成”的过程。因此消费事件时必须两种形态都处理不能假设“开了流式就全是流”。处理流式内容的代码ifmv.IsStreamingmv.MessageStream!nil{for{chunk,err:mv.MessageStream.Recv()iferrors.Is(err,io.EOF){break}iferr!nil{returnerr}// 拿到分片可以立即推送给前端}}读到io.EOF就结束这是正常终止第 09 篇讲过同样的规则。关于关闭ADK 发出的流式事件默认已启用自动关闭正常读到EOF即可。但如果你提前break跳出循环应显式Close()读取端避免资源泄漏。五、Action事件之外的信号有些信息不在消息里而在Action上typeAgentActionstruct{Exitbool// 是否结束Interrupted*InterruptInfo// 是否被中断TransferToAgent*TransferToAgentAction BreakLoop*BreakLoopAction CustomizedAction any}动作含义处理ExitAgent 决定结束正常收尾Interrupted被中断等待外部输入保存现场并等待恢复第 20 篇TransferToAgent转交给其他 Agent多智能体场景第 22–26 篇判断“是否被中断”必须看Action.Interrupted而不是靠有没有输出。中断时可能一个字的输出都没有。六、为交互界面组织事件事件流的典型消费场景是前端展示。一个实用的映射方式事件特征前端呈现roleassistant且流式逐字追加到回答气泡roleassistant且请求工具显示“正在查询…”roletool显示工具名与结果摘要可折叠Action.Interrupted弹出确认框Err ! nil显示错误并提供重试几个实践建议不要把工具的原始结果直接展示给用户。可能含内部信息且格式冗长。展示摘要必要时提供折叠详情。工具事件是提升体验的关键。用户看到“正在查询工单”比盯着转圈更有耐心。流式分片只做追加不要重排。前端应保证顺序渲染。区分“过程”与“结论”。中间的工具调用属于过程信息最终回答才是结论。七、完整示例下面这段程序不访问网络用同一份代码分别以流式和非流式运行观察两种事件形态packagemainimport(contexterrorsfmtiologstringsgithub.com/cloudwego/eino/adkgithub.com/cloudwego/eino/components/modelgithub.com/cloudwego/eino/components/toolgithub.com/cloudwego/eino/components/tool/utilsgithub.com/cloudwego/eino/composegithub.com/cloudwego/eino/schema)typeTicketReqstruct{TicketIDstringjson:ticket_id jsonschema:description工单编号}typemockModelstruct{}func(m*mockModel)Generate(_context.Context,in[]*schema.Message,_...model.Option)(*schema.Message,error){for_,msg:rangein{ifmsg.Roleschema.Tool{returnschema.AssistantMessage(T-1001 已分派给网络支持组。,nil),nil}}returnschema.AssistantMessage(,[]schema.ToolCall{{ID:call_1,Type:function,Function:schema.FunctionCall{Name:get_ticket,Arguments:{ticket_id:T-1001},},}}),nil}func(m*mockModel)Stream(ctx context.Context,in[]*schema.Message,opts...model.Option)(*schema.StreamReader[*schema.Message],error){msg,err:m.Generate(ctx,in,opts...)iferr!nil{returnnil,err}sr,sw:schema.Pipe[*schema.Message](1)gofunc(){defersw.Close()sw.Send(msg,nil)}()returnsr,nil}func(m*mockModel)WithTools([]*schema.ToolInfo)(model.ToolCallingChatModel,error){returnm,nil}funcmain(){iferr:run(false);err!nil{log.Fatal(err)}fmt.Println()iferr:run(true);err!nil{log.Fatal(err)}}funcrun(streamingbool)error{ctx:context.Background()label:非流式ifstreaming{label流式}fmt.Printf( %sEnableStreaming%v\n,label,streaming)ticketTool,err:utils.InferTool[*TicketReq,string](get_ticket,查询工单状态,func(_context.Context,req*TicketReq)(string,error){return工单 req.TicketID 状态已分派,nil})iferr!nil{returnerr}agent,err:adk.NewChatModelAgent(ctx,adk.ChatModelAgentConfig{Name:assistant,Description:工单助手,Instruction:需要时调用工具查询。,Model:mockModel{},ToolsConfig:adk.ToolsConfig{ToolsNodeConfig:compose.ToolsNodeConfig{Tools:[]tool.BaseTool{ticketTool},}},})iferr!nil{returnerr}runner:adk.NewRunner(ctx,adk.RunnerConfig{Agent:agent,EnableStreaming:streaming})iter:runner.Query(ctx,T-1001 什么状态)for{event,ok:iter.Next()if!ok{break}ifevent.Err!nil{returnevent.Err}mv:messageOutput(event)ifmvnil{continue}// 流式与非流式的差异内容所在的字段不同。ifmv.IsStreamingmv.MessageStream!nil{// 流式需要自己消费读取器。ADK 默认已设置自动关闭这里正常读到 EOF 即可。varsb strings.Builderfor{chunk,err:mv.MessageStream.Recv()iferrors.Is(err,io.EOF){break}iferr!nil{returnerr}sb.WriteString(chunk.Content)}fmt.Printf( [流式] role%-9s tool%-10s content%q\n,mv.Role,mv.ToolName,sb.String())continue}ifmv.Message!nil{fmt.Printf( [非流] role%-9s tool%-10s content%q\n,mv.Role,mv.ToolName,mv.Message.Content)}}returnnil}// messageOutput 安全地取出消息输出避免访问空指针。funcmessageOutput(event*adk.AgentEvent)*adk.MessageVariant{ifeventnil||event.Outputnil{returnnil}returnevent.Output.MessageOutput}八、多轮对话用 Run 传历史Query只接受一句话。要多轮对话需要用Run传入完整消息列表history:[]*schema.Message{schema.SystemMessage(你是工单助手。),schema.UserMessage(T-1001 什么状态),}iter:runner.Run(ctx,history)// ... 消费事件Agent 不会自动记住上一次运行。每次Run都是独立的执行历史要由你自己维护并重新传入——和普通 ChatModel 调用的规则一致第 03 篇。第 18 篇会讨论会话与记忆的处理方式。九、常见问题panic空指针event.Output或MessageOutput为nil。逐层判空尤其是中断和工具事件。流式事件里Message是空的流式内容在MessageStream里。检查IsStreaming再决定读哪个字段。工具结果拿不到工具名用mv.ToolName不要在流式模式下尝试从内容里推断。程序卡住不结束流式读取器没有读到EOF比如提前break却没有关闭。检查读取循环与关闭逻辑。错误被“吞掉”了没有处理event.Err。把错误分支放在事件处理的最前面。想拿模型的推理过程看消息的ReasoningContent字段前提是模型支持返回。它和Content是分开的事件流本身不展示内部思维链。十、小结Runner 是执行入口Query收单条字符串Run收完整消息列表。事件用Next()迭代ok false表示结束错误在event.Err里。Output与Action可能为nil访问前必须逐层判空。Role区分模型输出与工具结果ToolName标识工具来源。开流式后模型输出是流工具结果仍是非流——两种形态都要处理。判断中断要看Action.Interrupted不能靠有无输出。多轮对话不会自动记忆历史需自行维护。下一篇《上下文与记忆Agent 应该记住哪些信息》我们讨论上下文归属、会话共享与长期记忆。参考资料与版本说明资料核对日期为2026 年 10 月 9 日五分钟上手 Eino ADKEino ADK 概述Eino 主仓库本文示例按Eino v0.9.21编写使用模拟模型不访问网络已实机运行正文输出为真实结果。