企业官网建设流程全解析

【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 回顾

ReAct（Reasoning + Acting）模式在第 4 篇已介绍过——Agent 交替进行推理（Thought）和行动（Action），观察结果（Observation）后继续推理。

1.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):query=ev.query ctx.data["iteration"]=0else:query=ctx.data.get("query","")ctx.data["iteration"]=ev.iteration+1print(f" [观察]{ev.tool_name}→{ev.result[:60]}")ctx.data["query"]=query iteration=ctx.data["iteration"]# 模拟 LLM 推理（实际应调用 LLM）ifiteration>=2:# 信息足够，准备回答returnThoughtEvent(reasoning=f"经过{iteration}轮，信息已充足",next_action="answer",tool_input="")returnThoughtEvent(reasoning=f"第{iteration+1}轮：需要搜索更多信息",next_action="search",tool_input=query)@stepasyncdefact(self,ctx:Context,ev:ThoughtEvent)->ActionEvent|StopEvent:"""行动步骤：执行工具或给出最终答案"""ifev.next_action=="answer":returnStopEvent(result=f"最终答案：基于推理「{ev.reasoning}」")returnActionEvent(tool_name=ev.next_action,tool_input=ev.tool_input)@stepasyncdefobserve(self,ctx:Context,ev:ActionEvent)->ObservationEvent:"""观察步骤：执行工具并返回结果"""tool_fn=self.tools.get(ev.tool_name,lambdaq:"未知工具")result=tool_fn(ev.tool_input)returnObservationEvent(tool_name=ev.tool_name,result=result,iteration=ctx.data["iteration"])

1.3 事件流图

二、Function Calling Agent

2.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°C"def_get_time(self)->str:fromdatetimeimportdatetimereturndatetime.now().strftime("%H:%M:%S")@stepasyncdefplan(self,ctx:Context,ev:StartEvent)->ToolCallEvent|StopEvent:"""规划 Step：LLM 决定调用哪些工具"""query=ev.query.lower()if"天气"inqueryor"天气"inquery:returnToolCallEvent(tool_name="get_weather",arguments={"city":"北京"})elif"时间"inqueryor"几点"inquery:returnToolCallEvent(tool_name="get_time",arguments={})else:returnStopEvent(result=f"无需工具，直接回答：{ev.query}")@stepasyncdefexecute_tool(self,ctx:Context,ev:ToolCallEvent)->ToolResultEvent:"""执行工具"""tool_fn=self.tool_registry[ev.tool_name]result=tool_fn(**ev.arguments)returnToolResultEvent(tool_name=ev.tool_name,result=result)@stepasyncdefsynthesize(self,ctx:Context,ev:ToolResultEvent)->StopEvent:"""综合结果"""returnStopEvent(result=f"根据{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:"""生成草稿，请求人工审核"""draft=f"关于「{ev.topic}」的草稿内容..."ctx.data["draft"]=draftreturnHumanInputEvent(question="这份草稿是否可以发布？",context=draft)@stepasyncdefprocess_feedback(self,ctx:Context,ev:HumanResponseEvent)->StopEvent:"""处理人工反馈"""ifev.approved:returnStopEvent(result=f"已发布：{ctx.data['draft']}")else:returnStopEvent(result=f"已拒绝，反馈：{ev.answer}")

3.2 HITL 调用时序

四、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())# 使用registry=ToolRegistry()registry.register(name="search_web",fn=lambdaquery:f"搜索结果：{query}",description="搜索网页信息",parameters={"query":"搜索关键词"})registry.register(name="calculate",fn=lambdaexpression:str(eval(expression)),description="数学计算",parameters={"expression":"数学表达式"})

五、设计模式对比

本篇小结

下篇预告

第 39 篇：LlamaIndex 实战 — 构建 Corrective RAG 检索增强系统

模块六收官实战：构建一个自适应检索、质量评估、Web 搜索回退的完整 Corrective RAG 系统。

如果本篇内容对你有帮助，欢迎点赞收藏！有任何疑问，欢迎在评论区交流。