一、核心理论教学笔记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: params meta.params_model(**args_raw) except ValidationError as e: return f参数校验失败{str(e)} # 带超时执行 try: async with self.semaphore: result await asyncio.wait_for( meta.run_func(**params.model_dump()), timeoutmeta.timeout ) self.fail_counter[tool_name] 0 return result except asyncio.TimeoutError: self.fail_counter[tool_name] 1 return f工具{tool_name}执行超时 except Exception as e: self.fail_counter[tool_name] 1 return f工具执行异常{str(e)} # 网关初始化并注册工具 gateway ToolGateway() gateway.register_tool(ToolMeta( namecalculator, description数学加减乘除计算器仅处理数字运算, params_modelCalcParams, run_funccalculator )) gateway.register_tool(ToolMeta( namerag_search, description检索私有知识库查询RAG、Agent相关专业知识, params_modelRagSearchParams, run_funcrag_search ))2. ReAct 动态反思智能体 react_agent.pyimport asyncio import json import re from pydantic import BaseModel from typing import List, Dict # 复用模块 from llm_client_v2 import AsyncLLMClientV2 from memory_store import RedisChatMemory, MemoryCompressor from tool_gateway import gateway # 反思判断结果模型 class ReflectResult(BaseModel): enough_info: bool need_new_tool: bool reason: str class ReActAgent: def __init__(self, model_nameqwen-turbo): self.llm AsyncLLMClientV2(model_name) self.memory RedisChatMemory() self.compressor MemoryCompressor(self.llm) self.max_loop 5 # ReAct最大循环轮次 self.tools_schema gateway.get_openai_tools_schema() self.system_prompt 你是ReAct智能体遵循 Thought - Action - Observation 循环。 规则 1. Thought思考当前信息是否足够回答用户问题是否需要调用工具 2. Action需要工具则输出tool_calls信息充足直接输出最终回答 3. 可用工具calculator数学计算、rag_search知识库检索 每轮执行完工具后会进行自我反思判断是否继续循环。 禁止编造知识库不存在的信息数学计算必须调用计算器。 self.reflect_prompt 根据用户原始问题、历史对话、所有工具返回结果判断两点 1. enough_info现有信息是否足够完整回答用户问题true/false 2. need_new_tool是否需要继续调用新工具补充信息true/false 输出仅标准JSONkeyenough_info, need_new_tool, reason 用户问题{user_query} 当前全部上下文{context} async def init_memory(self): await self.memory.connect() async def close_memory(self): await self.memory.close() # 单轮反思判断是否终止循环 async def reflect(self, user_query: str, messages: List[Dict]) - ReflectResult: context_text \n.join([f{m[role]}:{m[content]} for m in messages]) prompt self.reflect_prompt.format(user_queryuser_query, contextcontext_text) raw await self.llm.chat([{role: user, content: prompt}], temperature0.0) match re.search(r\{.*\}, raw, re.S) json_str match.group() return ReflectResult.model_validate_json(json_str) # ReAct主循环 async def react_loop(self, user_query: str, messages: List[Dict]) - str: loop_count 0 while loop_count self.max_loop: loop_count 1 # 1. Thought阶段请求LLM判断是否调用工具 resp await self.llm._request(messages, toolsself.tools_schema) msg resp[choices][0][message] # 无工具调用直接返回答案终止循环 if tool_calls not in msg or not msg[tool_calls]: return msg[content] # 2. Action阶段执行工具 tool_call msg[tool_calls][0] call_id tool_call[id] func_name tool_call[function][name] func_args json.loads(tool_call[function][arguments]) obs await gateway.execute_tool(func_name, func_args) # 3. Observation回填上下文 messages.append(msg) messages.append({ role: tool, tool_call_id: call_id, content: obs }) # 4. 自我反思判断是否继续 reflect_res await self.reflect(user_query, messages) if reflect_res.enough_info and not reflect_res.need_new_tool: break # 循环达到上限汇总所有信息输出最终回答 final_resp await self.llm.chat(messages, temperature0.1) return final_resp # 完整对话入口带记忆 async def chat(self, session_id: str, user_query: str): # 读取历史记忆 history await self.memory.load_history(session_id) messages [{role: system, content: self.system_prompt}] history messages.append({role: user, content: user_query}) # 自动压缩超长上下文 compressed_msg await self.compressor.auto_compress(messages) # ReAct循环推理 final_ans await self.react_loop(user_query, compressed_msg) # 持久化本轮对话 await self.memory.append_message(session_id, user, user_query) await self.memory.append_message(session_id, assistant, final_ans) # 压缩覆盖历史防止膨胀 clean_history compressed_msg[1:] await self.memory.override_history(session_id, clean_history) return { session_id: session_id, query: user_query, answer: final_ans, max_loop_limit: self.max_loop } # 测试入口 async def test_react(): agent ReActAgent() await agent.init_memory() # 预加载知识库 await gateway.rag_search(RAG是什么, top_k3) # 复合型问题需要多轮反思 q 先计算 (12080)*5再介绍RAG的作用最后把计算结果和介绍整合 res await agent.chat(react_test_001, q) print(最终回答\n, res[answer]) await agent.close_memory() if __name__ __main__: asyncio.run(test_react())3. FastAPI 接口 main_react.pyfrom fastapi import FastAPI, Query import asyncio from react_agent import ReActAgent app FastAPI(titleDay6 ReAct动态反思Agent 统一工具网关) agent ReActAgent(qwen-turbo) app.on_event(startup) async def startup(): await agent.init_memory() app.on_event(shutdown) async def shutdown(): await agent.close_memory() app.get(/agent/react_chat) async def react_chat( session_id: str Query(..., description会话唯一ID), prompt: str Query(..., description用户提问) ): result await agent.chat(session_id, prompt) return result if __name__ __main__: import uvicorn uvicorn.run(main_react:app, reloadTrue)五、今日必做练习任务运行tool_gateway.py新增自定义工具如取模运算验证注册机制无需修改调度代码执行ReAct测试代码观察「思考→工具调用→反思→循环」完整流程修改max_loop循环上限测试达到上限强制汇总逻辑制造工具连续报错验证网关熔断机制工具临时禁用多轮连续对话验证Redis记忆持久化、自动压缩功能对比Plan-Solve与ReAct处理同一复杂问题观察两者执行流程差异六、今日配套面试题高阶Agent核心基础问答简述ReAct三元循环 Thought-Action-Observation 完整流程ReAct和Plan-Solve静态规划的核心区别各自适用场景统一工具网关有什么工程价值分层结构是怎样的ReAct为什么需要设置最大循环轮次不限制会出现什么问题工具网关熔断降级机制的作用是什么工程实操题ReAct出现无限重复调用同一工具如何从代码层面多重限制统一工具网关需要实现哪些通用拦截中间件能力如何实现ReAct自我反思让模型自主判断是否停止调用工具新增业务工具如何做到不修改Agent核心调度代码拓展思考题大厂Agent面试如何实现并行工具调用ReAct一次性执行多个无依赖工具工具网关如何增加权限控制不同用户分组开放不同工具ReAct循环Token持续上涨除了记忆压缩还有什么优化方案如何给ReAct增加记忆反思让模型自动遗忘无用历史对话面试题标准答案基础问答ReAct三元循环Thought模型读取上下文思考是否需要调用工具Action输出标准化工具调用参数Observation网关执行工具结果回填对话循环直至信息充足或达到最大轮次输出最终答案。ReAct vs Plan-SolvePlan-Solve一次性生成全部任务静态串行执行适合流程固定、依赖清晰的简单复合问题无法动态纠错。ReAct单步动态思考每轮根据工具返回调整思路支持自我反思纠错适合开放式、信息不确定、需要多轮试错的复杂问题。工具网关工程价值统一管理所有工具标准化注册、调用、拦截新增工具无侵入统一处理参数校验、限流、超时、熔断、日志上层Agent无需关心工具底层实现解耦业务工具与推理核心。三层结构工具元注册层、网关调度层、中间件拦截层。最大循环轮次作用防止模型无限循环重复调用工具造成API资源浪费、Token爆炸、接口超时到达上限强制停止工具调用基于已有信息汇总回答。熔断降级作用同一工具连续多次失败后临时禁用避免持续无效请求消耗额度防止单个工具故障拖垮整个Agent对话流程提升服务稳定性。工程实操题防止无限重复调用工具① 设置全局最大循环次数② 反思Prompt约束连续重复调用需停止③ 上下文记录历史工具操作模型可见过往调用记录④ 网关增加单轮最大调用次数限制。网关中间件能力参数校验、并发限流、独立超时控制、异常捕获格式化、连续失败熔断、权限过滤、调用日志埋点、鉴权拦截。ReAct自我反思实现单独构造反思Prompt传入全部对话与工具结果强制模型输出结构化JSON判断是否信息充足、是否需要继续调用工具temperature0保证判断稳定根据返回布尔值决定是否继续循环。无侵入新增工具仅需要定义Pydantic参数模型、异步执行函数、构造ToolMeta元数据调用gateway.register_tool完成注册网关自动生成OpenAI标准工具SchemaAgent调度逻辑完全无需改动。拓展思考题并行ReAct一轮Thought输出多个无依赖tool_calls使用asyncio.gather并发执行所有工具全部Observation统一回填上下文大幅提升多独立查询场景效率。工具权限控制工具元数据增加权限标签网关接收会话权限参数过滤当前用户无权限的工具LLM仅能获取用户可见工具列表无法调用受限工具。ReAct循环Token优化每次循环精简工具返回冗余文本分层记忆滑动窗口裁剪旧对话单次循环只保留关键Observation丢弃无关闲聊长工具返回结果摘要压缩。记忆反思遗忘机制每次反思时让模型评估历史对话权重低价值闲聊、过期信息标记记忆压缩阶段优先丢弃低权重内容只保留高价值计算、知识库、业务参数记录。
