今日总学习目标

1. 吃透 ReAct 动态反思智能体完整逻辑，对比 Day4 Plan-Solve 静态规划差异
2. 实现带自我反思、循环迭代的 ReAct Agent，支持中途修正工具调用逻辑
3. 统一封装全局工具网关，标准化管理所有工具（计算器、RAG检索、接口调用）
4. 完善工具鉴权、参数校验、超时熔断、失败降级，形成生产级工具调度层
   每日时长分配（全天8h）

• 理论笔记阅读+理解：2.5h
• 代码编写调试：4h
• 复盘+面试背诵：1.5h

一、核心理论教学笔记

1. ReAct 推理行动框架深度解析

1.1 核心循环三元组（Thought → Action → Observation）

1. Thought（思考）：模型读取当前全部上下文、已有工具返回结果，自主判断当前信息是否充足、下一步需要执行什么工具；若信息足够直接输出最终答案，终止循环。
2. Action（行动）：输出标准化工具调用指令，包含工具名称、入参；不做一次性全任务规划，只决策单步操作。
3. Observation（观察）：程序执行工具，捕获返回结果、异常信息，把结果回填对话上下文，进入下一轮思考。

循环上限控制：单轮对话限制最大迭代次数（默认5次），防止无限循环调用工具。

1.2 ReAct VS Plan-Solve 核心对比（面试高频）

维度	Plan-Solve（Day4静态规划）	ReAct（Day6动态反思）
规划时机	对话开头一次性生成全部子任务清单	每执行完一步动态思考下一步，无预先完整计划
纠错能力	前期规划出错，后续任务全部失效，无法动态修正	每轮可根据工具返回结果调整思路，支持自我纠错
适用场景	固定流程、依赖明确、简单复合型问题	开放式复杂问题、结果不确定、需要多轮试错检索/计算
Token消耗	一次性规划消耗大量Token	每轮思考轻量化，分步消耗，灵活可控
实现难度	中等，调度器串行执行预设任务	偏高，需维护循环上下文、反思逻辑

1.3 ReAct 自我反思机制

普通ReAct仅执行「思考-行动」，增强版增加反思逻辑：
工具返回结果后，额外让模型判断3个问题：

1. 当前信息是否足够回答用户原始问题？
2. 当前工具返回是否存在错误、缺失关键数据？
3. 是否需要更换工具、补充检索、重新计算？
   反思结论决定是否继续循环，大幅减少无效工具调用。

2. 统一工具网关设计（生产工程必备）

前几日工具分散定义，新增工具需要修改多处代码，不利于维护，今日抽象统一网关层：

2.1 三层工具架构

1. 工具元定义层：统一注册工具名称、描述、入参Schema、异步执行函数、权限标签
2. 网关调度层：统一接收模型工具调用请求，路由分发到对应工具
3. 拦截中间件层：统一处理参数校验、超时、限流、异常捕获、日志、熔断降级

2.2 网关内置拦截能力

1. 参数校验：Pydantic统一校验，非法参数直接返回错误观察值
2. 并发限流：全局信号量控制工具并行调用数量
3. 超时熔断：单工具执行超时阈值，超时直接返回失败信息
4. 异常捕获：统一捕获代码/接口异常，格式化错误回填上下文
5. 权限过滤：支持给不同会话开放/禁用指定工具
6. 日志埋点：记录工具调用耗时、入参、结果、成功失败状态

2.3 标准化工具注册规范

新增工具无需修改调度逻辑，仅需注册元数据：

ToolMeta(
    name="xxx",
    description="工具功能描述",
    params_model=Pydantic参数模型,
    run_func=异步执行函数,
    timeout=10,
    enable=True
)

3. ReAct + 工具网关 + 分层记忆 完整串联链路

用户提问

1. Redis读取会话记忆，自动滑动窗口+摘要压缩，构建上下文消息
2. 进入ReAct循环：
   • Thought：模型判断下一步动作（工具调用/直接回答）
   • Action：输出工具调用参数，交给统一工具网关
   • 网关中间件校验、执行工具，返回Observation观察结果
   • 结果回填上下文，执行自我反思，判断是否继续循环
3. 循环终止，LLM整合全部观察输出最终答案
4. 保存本轮问答至Redis持久化记忆

4. 工具熔断降级策略

• 连续失败阈值：同一工具连续调用2次失败，临时禁用该工具，返回降级提示
• 超时降级：工具超时后不重试，直接告知用户当前工具暂时不可用
• 权限降级：无权限工具拦截，返回提示不执行调用

二、今日学习重点

1. 掌握ReAct循环推理流程、自我反思实现逻辑
2. 区分ReAct动态推理与Plan-Solve静态规划的适用场景
3. 封装通用可扩展工具网关，统一管理所有工具（计算器、RAG检索）
4. 实现带循环上限、反思纠错的完整ReAct智能体
5. 整合Day5分层记忆，实现持久化会话+动态反思Agent一体化

三、今日难点 & 解决方案

难点1：ReAct无限循环重复调用同一工具

解决方案：

1. 设置全局最大迭代轮次，到达上限强制终止循环，直接汇总现有结果
2. 反思Prompt约束：连续两次调用相同工具且无新信息，停止调用
3. 上下文记录每一轮工具执行记录，模型可看到历史操作，避免重复动作

难点2：工具零散、新增工具需要大量修改代码，维护成本高

解决方案：
统一工具网关注册机制，所有工具集中注册；网关自动路由、统一拦截，新增工具仅新增元数据与执行函数，无侵入改动核心调度代码。

难点3：工具执行超时、接口报错导致整个ReAct流程卡死

解决方案：
网关层统一设置工具独立超时；全局捕获所有异常，格式化错误信息作为Observation回填上下文，循环不中断；连续失败触发熔断降级。

难点4：模型反思判断失效，明明信息充足仍反复调用工具

解决方案：

1. 反思环节单独提供标准化判断模板，强制输出布尔判断结果
2. temperature=0，保证反思结论稳定，减少误判
3. Few-shot示例展示信息充足时直接终止循环的案例

四、完整练习代码（基于Day1~Day5全部代码扩展）

依赖不变

aiohttp、pydantic、faiss-cpu、aioredis、fastapi、uvicorn、numpy

1. 统一工具网关 tool_gateway.py

import 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(default=2, 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: