【AI Agent 编排全栈实战】第 38 篇Agent 工作流 — ReAct、Function Calling 与 Human-in-the-Loop本系列定位面向 AI 应用开发者的 Agent 编排框架系统化教程。以 Python 为主技术栈系统讲解 5 大主流编排框架的编排架构范式、核心抽象和实战选型。本篇你将学到用 LlamaIndex Workflows 实现 ReAct Agent 工作流Function Calling Agent 的事件驱动实现Human-in-the-Loop 交互工作流Agent 工具集成与编排学完本篇你将能够在 LlamaIndex Workflows 中实现所有主流的 Agent 设计模式。一、ReAct Agent 工作流1.1 ReAct 回顾ReActReasoning Acting模式在第 4 篇已介绍过——Agent 交替进行推理Thought和行动Action观察结果Observation后继续推理。用户问题决定使用什么工具获取工具结果基于结果继续推理信息充足ThoughtActionObservationAnswerReAct 循环Thought → Action → Observation1.2 事件驱动实现importasynciofromllama_index.core.workflowimport(Workflow,step,StartEvent,StopEvent,Event,Context)classThoughtEvent(Event):reasoning:strnext_action:strtool_input:strclassActionEvent(Event):tool_name:strtool_input:strclassObservationEvent(Event):tool_name:strresult:striteration:intclassReActWorkflow(Workflow):ReAct Agent 事件驱动工作流tools{search:lambdaq:f搜索结果关于「{q}」的信息,calculator:lambdaq:f计算结果{eval(q)ifq.replace(.,).isdigit()else无法计算},}stepasyncdefreason(self,ctx:Context,ev:StartEvent|ObservationEvent)-ThoughtEvent:推理步骤决定下一步做什么ifisinstance(ev,StartEvent):queryev.query ctx.data[iteration]0else:queryctx.data.get(query,)ctx.data[iteration]ev.iteration1print(f [观察]{ev.tool_name}→{ev.result[:60]})ctx.data[query]query iterationctx.data[iteration]# 模拟 LLM 推理实际应调用 LLMifiteration2:# 信息足够准备回答returnThoughtEvent(reasoningf经过{iteration}轮信息已充足,next_actionanswer,tool_input)returnThoughtEvent(reasoningf第{iteration1}轮需要搜索更多信息,next_actionsearch,tool_inputquery)stepasyncdefact(self,ctx:Context,ev:ThoughtEvent)-ActionEvent|StopEvent:行动步骤执行工具或给出最终答案ifev.next_actionanswer:returnStopEvent(resultf最终答案基于推理「{ev.reasoning}」)returnActionEvent(tool_nameev.next_action,tool_inputev.tool_input)stepasyncdefobserve(self,ctx:Context,ev:ActionEvent)-ObservationEvent:观察步骤执行工具并返回结果tool_fnself.tools.get(ev.tool_name,lambdaq:未知工具)resulttool_fn(ev.tool_input)returnObservationEvent(tool_nameev.tool_name,resultresult,iterationctx.data[iteration])1.3 事件流图next_action ! answernext_action answerStartEventreasonThoughtEventObservationEventactActionEventStopEventobserveObservationEvent二、Function Calling Agent2.1 工作流实现frompydanticimportBaseModelclassToolCallEvent(Event):tool_name:strarguments:dictclassToolResultEvent(Event):tool_name:strresult:strclassFunctionCallingWorkflow(Workflow):Function Calling Agent 工作流def__init__(self,*args,**kwargs):super().__init__(*args,**kwargs)self.tool_registry{get_weather:self._get_weather,get_time:self._get_time,}def_get_weather(self,city:str)-str:returnf{city}晴25°Cdef_get_time(self)-str:fromdatetimeimportdatetimereturndatetime.now().strftime(%H:%M:%S)stepasyncdefplan(self,ctx:Context,ev:StartEvent)-ToolCallEvent|StopEvent:规划 StepLLM 决定调用哪些工具queryev.query.lower()if天气inqueryor天气inquery:returnToolCallEvent(tool_nameget_weather,arguments{city:北京})elif时间inqueryor几点inquery:returnToolCallEvent(tool_nameget_time,arguments{})else:returnStopEvent(resultf无需工具直接回答{ev.query})stepasyncdefexecute_tool(self,ctx:Context,ev:ToolCallEvent)-ToolResultEvent:执行工具tool_fnself.tool_registry[ev.tool_name]resulttool_fn(**ev.arguments)returnToolResultEvent(tool_nameev.tool_name,resultresult)stepasyncdefsynthesize(self,ctx:Context,ev:ToolResultEvent)-StopEvent:综合结果returnStopEvent(resultf根据{ev.tool_name}的结果{ev.result})三、Human-in-the-Loop 工作流3.1 HITL 事件设计classHumanInputEvent(Event):请求人工输入的事件question:strcontext:strclassHumanResponseEvent(Event):人工响应事件answer:strapproved:boolclassHITLWorkflow(Workflow):Human-in-the-Loop 工作流stepasyncdefgenerate_draft(self,ctx:Context,ev:StartEvent)-HumanInputEvent:生成草稿请求人工审核draftf关于「{ev.topic}」的草稿内容...ctx.data[draft]draftreturnHumanInputEvent(question这份草稿是否可以发布,contextdraft)stepasyncdefprocess_feedback(self,ctx:Context,ev:HumanResponseEvent)-StopEvent:处理人工反馈ifev.approved:returnStopEvent(resultf已发布{ctx.data[draft]})else:returnStopEvent(resultf已拒绝反馈{ev.answer})3.2 HITL 调用时序process_feedbackgenerate_draftWorkflow用户process_feedbackgenerate_draftWorkflow用户人工审核中...run(topic...)StartEventHumanInputEvent(可以发布吗)请求人工输入HumanResponseEvent(approvedTrue)HumanResponseEventStopEvent返回结果四、Agent 工具集成4.1 工具注册模式fromtypingimportCallable,AnyclassToolRegistry:工具注册表def__init__(self):self._tools:dict[str,Callable]{}self._schemas:dict[str,dict]{}defregister(self,name:str,fn:Callable,description:str,parameters:dict[str,str]):注册工具self._tools[name]fn self._schemas[name]{description:description,parameters:parameters,}defget(self,name:str)-Callable|None:returnself._tools.get(name)defget_schema(self,name:str)-dict:returnself._schemas.get(name,{})deflist_tools(self)-list[str]:returnlist(self._tools.keys())# 使用registryToolRegistry()registry.register(namesearch_web,fnlambdaquery:f搜索结果{query},description搜索网页信息,parameters{query:搜索关键词})registry.register(namecalculate,fnlambdaexpression:str(eval(expression)),description数学计算,parameters{expression:数学表达式})五、设计模式对比模式事件流适用场景ReActreason→act→observe 循环需要 Tool 使用 推理Function Callingplan→execute→synthesize明确的工具调用链HITLgenerate→wait→process高风险操作需人工审核Reflectiongenerate→evaluate→revise 循环质量优化迭代本篇小结知识点核心内容ReAct 工作流reason/act/observe 三步循环Function Callingplan/execute/synthesize 三步链HITLHumanInputEvent HumanResponseEvent工具注册ToolRegistry 集中管理工具事件驱动 AgentEvent 类型定义了 Agent 行为流程下篇预告第 39 篇LlamaIndex 实战 — 构建 Corrective RAG 检索增强系统模块六收官实战构建一个自适应检索、质量评估、Web 搜索回退的完整 Corrective RAG 系统。如果本篇内容对你有帮助欢迎点赞收藏有任何疑问欢迎在评论区交流。