今日总学习目标吃透 ReAct 动态反思智能体完整逻辑对比 Day4 Plan-Solve 静态规划差异实现带自我反思、循环迭代的 ReAct Agent支持中途修正工具调用逻辑统一封装全局工具网关标准化管理所有工具计算器、RAG检索、接口调用完善工具鉴权、参数校验、超时熔断、失败降级形成生产级工具调度层每日时长分配全天8h理论笔记阅读理解2.5h代码编写调试4h复盘面试背诵1.5h一、核心理论教学笔记1. ReAct 推理行动框架深度解析1.1 核心循环三元组Thought → Action → ObservationThought思考模型读取当前全部上下文、已有工具返回结果自主判断当前信息是否充足、下一步需要执行什么工具若信息足够直接输出最终答案终止循环。Action行动输出标准化工具调用指令包含工具名称、入参不做一次性全任务规划只决策单步操作。Observation观察程序执行工具捕获返回结果、异常信息把结果回填对话上下文进入下一轮思考。循环上限控制单轮对话限制最大迭代次数默认5次防止无限循环调用工具。1.2 ReAct VS Plan-Solve 核心对比面试高频维度Plan-SolveDay4静态规划ReActDay6动态反思规划时机对话开头一次性生成全部子任务清单每执行完一步动态思考下一步无预先完整计划纠错能力前期规划出错后续任务全部失效无法动态修正每轮可根据工具返回结果调整思路支持自我纠错适用场景固定流程、依赖明确、简单复合型问题开放式复杂问题、结果不确定、需要多轮试错检索/计算Token消耗一次性规划消耗大量Token每轮思考轻量化分步消耗灵活可控实现难度中等调度器串行执行预设任务偏高需维护循环上下文、反思逻辑1.3 ReAct 自我反思机制普通ReAct仅执行「思考-行动」增强版增加反思逻辑工具返回结果后额外让模型判断3个问题当前信息是否足够回答用户原始问题当前工具返回是否存在错误、缺失关键数据是否需要更换工具、补充检索、重新计算反思结论决定是否继续循环大幅减少无效工具调用。2. 统一工具网关设计生产工程必备前几日工具分散定义新增工具需要修改多处代码不利于维护今日抽象统一网关层2.1 三层工具架构工具元定义层统一注册工具名称、描述、入参Schema、异步执行函数、权限标签网关调度层统一接收模型工具调用请求路由分发到对应工具拦截中间件层统一处理参数校验、超时、限流、异常捕获、日志、熔断降级2.2 网关内置拦截能力参数校验Pydantic统一校验非法参数直接返回错误观察值并发限流全局信号量控制工具并行调用数量超时熔断单工具执行超时阈值超时直接返回失败信息异常捕获统一捕获代码/接口异常格式化错误回填上下文权限过滤支持给不同会话开放/禁用指定工具日志埋点记录工具调用耗时、入参、结果、成功失败状态2.3 标准化工具注册规范新增工具无需修改调度逻辑仅需注册元数据ToolMeta( namexxx, description工具功能描述, params_modelPydantic参数模型, run_func异步执行函数, timeout10, enableTrue )3. ReAct 工具网关 分层记忆 完整串联链路用户提问Redis读取会话记忆自动滑动窗口摘要压缩构建上下文消息进入ReAct循环Thought模型判断下一步动作工具调用/直接回答Action输出工具调用参数交给统一工具网关网关中间件校验、执行工具返回Observation观察结果结果回填上下文执行自我反思判断是否继续循环循环终止LLM整合全部观察输出最终答案保存本轮问答至Redis持久化记忆4. 工具熔断降级策略连续失败阈值同一工具连续调用2次失败临时禁用该工具返回降级提示超时降级工具超时后不重试直接告知用户当前工具暂时不可用权限降级无权限工具拦截返回提示不执行调用二、今日学习重点掌握ReAct循环推理流程、自我反思实现逻辑区分ReAct动态推理与Plan-Solve静态规划的适用场景封装通用可扩展工具网关统一管理所有工具计算器、RAG检索实现带循环上限、反思纠错的完整ReAct智能体整合Day5分层记忆实现持久化会话动态反思Agent一体化三、今日难点 解决方案难点1ReAct无限循环重复调用同一工具解决方案设置全局最大迭代轮次到达上限强制终止循环直接汇总现有结果反思Prompt约束连续两次调用相同工具且无新信息停止调用上下文记录每一轮工具执行记录模型可看到历史操作避免重复动作难点2工具零散、新增工具需要大量修改代码维护成本高解决方案统一工具网关注册机制所有工具集中注册网关自动路由、统一拦截新增工具仅新增元数据与执行函数无侵入改动核心调度代码。难点3工具执行超时、接口报错导致整个ReAct流程卡死解决方案网关层统一设置工具独立超时全局捕获所有异常格式化错误信息作为Observation回填上下文循环不中断连续失败触发熔断降级。难点4模型反思判断失效明明信息充足仍反复调用工具解决方案反思环节单独提供标准化判断模板强制输出布尔判断结果temperature0保证反思结论稳定减少误判Few-shot示例展示信息充足时直接终止循环的案例四、完整练习代码基于Day1~Day5全部代码扩展依赖不变aiohttp、pydantic、faiss-cpu、aioredis、fastapi、uvicorn、numpy1. 统一工具网关 tool_gateway.pyimport asyncio from pydantic import BaseModel, Field, ValidationError from typing import Dict, List, Optional, Callable, Any # 复用已有工具能力 from rag_store import RAGService # 工具元数据模型 class ToolMeta(BaseModel): name: str description: str params_model: type[BaseModel] run_func: Callable timeout: int 10 enable: bool True # 计算器参数模型 class CalcParams(BaseModel): num1: float Field(description数字1) num2: float Field(description数字2) op: str Field(description运算符 -*/) # RAG检索参数模型 class RagSearchParams(BaseModel): query: str Field(description知识库检索关键词) top_k: int Field(default2, description返回片段数量) # 异步工具函数 async def calculator(num1: float, num2: float, op: str) - str: try: match op: case : res num1 num2 case -: res num1 - num2 case *: res num1 * num2 case /: if num2 0: return 工具错误除数不能为0 res num1 / num2 case _: return f不支持运算符 {op} return f计算结果{num1} {op} {num2} {res} except Exception as e: return f计算异常{str(e)} # 全局RAG实例 global_rag RAGService() async def rag_search(query: str, top_k: int): res await global_rag.retrieve(query, top_k) if not res: return 知识库未查询到相关内容 text \n.join([item[text] for item in res]) return f知识库检索结果\n{text} # 统一工具网关 class ToolGateway: def __init__(self): self.tool_registry: Dict[str, ToolMeta] {} self.semaphore asyncio.Semaphore(5) self.fail_counter: Dict[str, int] {} # 工具连续失败计数 self.max_continuous_fail 2 # 注册工具 def register_tool(self, meta: ToolMeta): self.tool_registry[meta.name] meta self.fail_counter[meta.name] 0 # 获取OpenAI标准tools定义直接传给LLM def get_openai_tools_schema(self) - List[Dict]: tools [] for meta in self.tool_registry.values(): if not meta.enable: continue tools.append({ type: function, function: { name: meta.name, description: meta.description, parameters: meta.params_model.model_json_schema() } }) return tools # 执行单一工具 async def execute_tool(self, tool_name: str, args_raw: dict) - str: # 熔断判断 if self.fail_counter.get(tool_name, 0) self.max_continuous_fail: return f工具{tool_name}已触发熔断暂时无法使用 if tool_name not in self.tool_registry: return f不存在工具{tool_name} meta self.tool_registry[tool_name] # 参数校验 try: