循环上限控制：单轮对话限制最大迭代次数（默认5次），防止无限循环调用工具。
| 维度 | Plan-Solve（Day4静态规划） | ReAct（Day6动态反思） |
|---|---|---|
| 规划时机 | 对话开头一次性生成全部子任务清单 | 每执行完一步动态思考下一步，无预先完整计划 |
| 纠错能力 | 前期规划出错，后续任务全部失效，无法动态修正 | 每轮可根据工具返回结果调整思路，支持自我纠错 |
| 适用场景 | 固定流程、依赖明确、简单复合型问题 | 开放式复杂问题、结果不确定、需要多轮试错检索/计算 |
| Token消耗 | 一次性规划消耗大量Token | 每轮思考轻量化，分步消耗，灵活可控 |
| 实现难度 | 中等，调度器串行执行预设任务 | 偏高，需维护循环上下文、反思逻辑 |
普通ReAct仅执行「思考-行动」，增强版增加反思逻辑：
工具返回结果后，额外让模型判断3个问题：
前几日工具分散定义，新增工具需要修改多处代码，不利于维护，今日抽象统一网关层：
新增工具无需修改调度逻辑，仅需注册元数据：
ToolMeta(
name="xxx",
description="工具功能描述",
params_model=Pydantic参数模型,
run_func=异步执行函数,
timeout=10,
enable=True
)
用户提问
解决方案：
解决方案：
统一工具网关注册机制，所有工具集中注册；网关自动路由、统一拦截，新增工具仅新增元数据与执行函数，无侵入改动核心调度代码。
解决方案：
网关层统一设置工具独立超时；全局捕获所有异常，格式化错误信息作为Observation回填上下文，循环不中断；连续失败触发熔断降级。
解决方案：
aiohttp、pydantic、faiss-cpu、aioredis、fastapi、uvicorn、numpy
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:
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()),
timeout=meta.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(
name="calculator",
description="数学加减乘除计算器，仅处理数字运算",
params_model=CalcParams,
run_func=calculator
))
gateway.register_tool(ToolMeta(
name="rag_search",
description="检索私有知识库，查询RAG、Agent相关专业知识",
params_model=RagSearchParams,
run_func=rag_search
))
import 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_name="qwen-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）
输出仅标准JSON，key：enough_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_query=user_query, context=context_text)
raw = await self.llm.chat([{"role": "user", "content": prompt}], temperature=0.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, tools=self.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, temperature=0.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_k=3)
# 复合型问题，需要多轮反思
q = "先计算 (120+80)*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())
from fastapi import FastAPI, Query
import asyncio
from react_agent import ReActAgent
app = FastAPI(title="Day6 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", reload=True)
tool_gateway.py，新增自定义工具（如取模运算），验证注册机制无需修改调度代码max_loop循环上限，测试达到上限强制汇总逻辑ReAct三元循环
Thought：模型读取上下文，思考是否需要调用工具；
Action：输出标准化工具调用参数；
Observation：网关执行工具，结果回填对话；
循环直至信息充足或达到最大轮次，输出最终答案。
ReAct vs Plan-Solve
Plan-Solve：一次性生成全部任务，静态串行执行，适合流程固定、依赖清晰的简单复合问题；无法动态纠错。
ReAct：单步动态思考，每轮根据工具返回调整思路，支持自我反思纠错，适合开放式、信息不确定、需要多轮试错的复杂问题。
工具网关工程价值
统一管理所有工具，标准化注册、调用、拦截；新增工具无侵入；统一处理参数校验、限流、超时、熔断、日志；上层Agent无需关心工具底层实现，解耦业务工具与推理核心。
三层结构：工具元注册层、网关调度层、中间件拦截层。
最大循环轮次作用
防止模型无限循环重复调用工具，造成API资源浪费、Token爆炸、接口超时；到达上限强制停止工具调用，基于已有信息汇总回答。
熔断降级作用
同一工具连续多次失败后临时禁用，避免持续无效请求消耗额度；防止单个工具故障拖垮整个Agent对话流程，提升服务稳定性。
防止无限重复调用工具
① 设置全局最大循环次数；② 反思Prompt约束连续重复调用需停止；③ 上下文记录历史工具操作，模型可见过往调用记录；④ 网关增加单轮最大调用次数限制。
网关中间件能力
参数校验、并发限流、独立超时控制、异常捕获格式化、连续失败熔断、权限过滤、调用日志埋点、鉴权拦截。
ReAct自我反思实现
单独构造反思Prompt，传入全部对话与工具结果，强制模型输出结构化JSON判断是否信息充足、是否需要继续调用工具；temperature=0保证判断稳定，根据返回布尔值决定是否继续循环。
无侵入新增工具
仅需要定义Pydantic参数模型、异步执行函数、构造ToolMeta元数据，调用gateway.register_tool完成注册；网关自动生成OpenAI标准工具Schema，Agent调度逻辑完全无需改动。
Day6完成工业级动态ReAct智能体与统一工具网关两大核心模块，弥补Day4静态规划无法动态纠错的短板。
统一工具网关是工程落地必备分层设计，解决工具散乱、维护困难问题；ReAct动态反思是复杂开放式业务场景首选推理框架，面试高频考察两种规划框架对比。
目前全套体系：异步LLM、SSE流式、Function Calling、RAG、Plan-Solve静态规划、Redis分层持久记忆、统一工具网关、ReAct动态反思智能体，完整覆盖中小型Agent项目全部底层能力。
下一日学习内容：多模态Agent、图文Embedding、本地轻量向量库Chroma持久化。
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。