当前位置：首页 > news >正文

大模型Agent工作流设计模式深度解析：从ReAct到ReWOO的实践演进

news 2025/11/16 6:51:39

摘要：本文深入探讨了大语言模型智能体（LLM Agent）的工作流设计模式，对比分析了ReAct、ReWOO等主流框架的核心思想、适用场景与工程实践。通过完整的代码实现，展示如何构建一个具备任务分解、工具调用和自我反思能力的生产级Agent系统，并分享在真实业务落地中的性能优化策略与踩坑经验。

引言：当大模型学会"思考"与"行动"

2023年以来，以ReAct（Reasoning+Acting）为代表的智能体架构彻底改变了大模型的应用范式。我们不再满足于简单的问答，而是期望模型能够像人类专家一样：理解复杂任务、制定执行计划、调用外部工具、反思中间结果、动态调整策略。然而，简单的ReAct模式在生产环境中常面临思考链过长、工具调用冗余、推理成本高等问题。本文将带你深入Agent工作流设计的内核，探索更高效、更可靠的架构模式。

一、Agent工作流的核心设计模式

1.1 ReAct模式：经典但非银弹

ReAct的核心是思考-行动-观察的循环：

# 伪代码示例
while not task_solved:thought = model.think(task_context)action = model.decide_tool(thought)observation = execute(action)task_context += f"\nThought:{thought}\nAction:{action}\nObs:{observation}"

痛点：每次推理都需重新生成历史上下文，导致token消耗指数级增长，在需要10+轮工具调用的任务中，GPT-4的单次任务成本可达$0.5+。

1.2 ReWOO模式：模块化推理的前向规划

ReWOO（Reasoning Without Observation） 将推理与执行解耦，采用计划先行、批量执行策略：

Planner：一次性生成完整任务计划和依赖图
Worker：并行执行所有工具调用
Solver：整合结果生成最终答案

这种模式的优势：

减少70%以上的LLM调用次数
支持工具调用并行化
计划可解释、可缓存

二、生产级Agent系统实现

下面实现一个支持ReAct与ReWOO双模式的通用Agent框架：

from typing import List, Dict, Any, Optional
from abc import ABC, abstractmethod
import asyncio
import jsonclass Tool(ABC):@abstractmethodasync def execute(self, params: Dict) -> str:passclass SearchTool(Tool):async def execute(self, params: Dict) -> str:query = params["query"]# 模拟搜索return f"搜索结果：关于'{query}'的10条相关信息..."class CalculatorTool(Tool):async def execute(self, params: Dict) -> str:expr = params["expression"]return str(eval(expr))class BaseAgent(ABC):def __init__(self, llm, tools: Dict[str, Tool]):self.llm = llmself.tools = tools@abstractmethodasync def run(self, task: str) -> str:pass# ReWOO实现
class ReWOOAgent(BaseAgent):def __init__(self, llm, tools: Dict[str, Tool]):super().__init__(llm, tools)self.planner_prompt = """将任务分解为可执行步骤。对每个步骤：1. 使用工具名称和参数2. 格式：[stepX] 工具名(参数) | 依赖: stepY,stepZ任务：{task}可用工具：{tool_names}示例：任务：计算2023年诺贝尔物理学奖得主的年龄[step1] Search(query="2023年诺贝尔物理学奖得主") | 依赖: [step2] Calculator(expression="2023-1964") | 依赖: step1"""async def run(self, task: str) -> str:# 1. 计划阶段tool_names = list(self.tools.keys())plan_prompt = self.planner_prompt.format(task=task, tool_names=tool_names)plan = await self.llm.generate(plan_prompt)# 解析计划为DAGsteps = self._parse_plan(plan)# 2. 执行阶段results = {}for step_id, step_info in steps.items():# 等待依赖完成deps = step_info["dependencies"]if deps:await asyncio.gather(*[self._wait_for_dependency(d, results) for d in deps])# 执行当前步骤tool = self.tools[step_info["tool"]]params = self._fill_params(step_info["params"], results)results[step_id] = await tool.execute(params)# 3. 总结阶段final_prompt = f"""任务：{task}执行结果：{results}请整合信息给出最终答案。"""return await self.llm.generate(final_prompt)def _parse_plan(self, plan: str) -> Dict:# 解析计划文本为结构化数据steps = {}for line in plan.split("\n"):if line.startswith("[step"):# 解析逻辑略passreturn steps# ReAct实现
class ReActAgent(BaseAgent):async def run(self, task: str, max_rounds: int = 10):context = f"任务：{task}\n"for i in range(max_rounds):# 思考thought = await self.llm.generate(context + "下一步思考：",stop=["Action:"])# 行动action = await self.llm.generate(context + thought + "\nAction:",stop=["Observation:"])if "Finish" in action:return action.split("Finish:")[1]# 执行tool_name, params = self._parse_action(action)observation = await self.tools[tool_name].execute(params)context += f"{thought}\nAction: {action}\nObservation: {observation}\n"return "任务未完成"

三、性能优化实战技巧

3.1 思考链压缩

使用总结模型对过长的上下文进行压缩：

async def compress_history(self, history: str) -> str:prompt = f"将以下执行历史压缩为关键信息摘要（<100字）：\n{history}"return await self.llm.generate(prompt, max_tokens=150)

3.2 工具调用缓存

对重复查询使用Redis缓存：

import hashlibclass CachedTool(Tool):def __init__(self, underlying_tool: Tool, redis_client):self.tool = underlying_toolself.redis = redis_clientasync def execute(self, params: Dict) -> str:key = hashlib.md5(json.dumps(params).encode()).hexdigest()cached = await self.redis.get(key)if cached:return cachedresult = await self.tool.execute(params)await self.redis.setex(key, 3600, result)  # 缓存1小时return result

3.3 动态温度调度

class AdaptiveLLM:def __init__(self, base_model):self.model = base_modelasync def generate(self, prompt: str, **kwargs):# 简单任务用低温度，创意任务用高温度complexity = self._estimate_complexity(prompt)temperature = 0.3 if complexity < 0.5 else 0.7return await self.model.generate(prompt, temperature=temperature, **kwargs)