AI Agent的规划能力：从ReAct到Tree-of-Planning

Thought: 用户要查北京天气，我需要调用天气API
Action: search_weather(city="北京")
Observation: 北京今天晴，25°C
Thought: 获取到天气信息，可以回答用户了
Answer: 北京今天天气晴朗，气温25°C，适合外出活动。

class ReActAgent:
    def __init__(self, llm, tools, max_steps=10):
        self.llm = llm
        self.tools = tools
        self.max_steps = max_steps
    
    def run(self, task):
        trajectory = []
        for step in range(self.max_steps):
            # 生成下一步思考和行为
            prompt = self._build_prompt(task, trajectory)
            response = self.llm.generate(prompt)
            
            thought, action = self._parse(response)
            trajectory.append({"thought": thought, "action": action})
            
            if action["type"] == "finish":
                return action["answer"]
            
            # 执行工具
            if action["type"] == "tool":
                result = self.tools[action["name"]](**action["params"])
                trajectory.append({"observation": result})
        
        return "达到最大步数限制"
    
    def _build_prompt(self, task, trajectory):
        prompt = f"""任务：{task}

可用工具：{list(self.tools.keys())}

历史：
"""
        for item in trajectory:
            if "thought" in item:
                prompt += f"Thought: {item['thought']}\n"
                prompt += f"Action: {item['action']}\n"
            elif "observation" in item:
                prompt += f"Observation: {item['observation']}\n"
        
        prompt += "Thought:"
        return prompt

class ReflexionAgent(ReActAgent):
    def __init__(self, llm, tools, max_steps=10, max_trials=3):
        super().__init__(llm, tools, max_steps)
        self.max_trials = max_trials
        self.reflections = []
    
    def run(self, task):
        for trial in range(self.max_trials):
            result = super().run(task)
            
            if self._is_success(result):
                return result
            
            # 生成反思
            reflection = self._reflect(task, self.trajectory, result)
            self.reflections.append(reflection)
            
            # 带着反思重新尝试
            task_context = f"""
            任务：{task}
            
            过去的尝试和反思：
            {self.reflections[-1]}
            
            避免重复之前的错误。
            """
            task = task_context
        
        return result
    
    def _reflect(self, task, trajectory, result):
        prompt = f"""分析以下失败的执行过程，总结经验教训：

任务：{task}
执行过程：{trajectory}
结果：{result}（失败）

请反思：
1. 哪一步出了问题？
2. 为什么会出错？
3. 下次应该怎么避免？
"""
        return self.llm.generate(prompt)

                    根节点（初始状态）
                   /        |        \
            子节点A      子节点B      子节点C
           /    \         |          |
        A1     A2        B1         C1
        |
        A11

class LATSAgent:
    def __init__(self, llm, tools, n_expansions=5, depth_limit=5):
        self.llm = llm
        self.tools = tools
        self.n_expansions = n_expansions
        self.depth_limit = depth_limit
    
    def run(self, task):
        root = StateNode(task=task, state=None, parent=None)
        
        for _ in range(self.depth_limit):
            # 1. 选择：UCB选择最有前景的节点
            node = self._select(root)
            
            # 2. 扩展：生成可能的行动
            children = self._expand(node)
            
            # 3. 评估：用LLM评估每个子节点的价值
            for child in children:
                child.value = self._evaluate(child)
            
            # 4. 模拟：快速估算该路径的最终收益
            for child in children:
                child.reward = self._simulate(child)
            
            # 5. 回溯：更新路径上所有节点的值
            self._backpropagate(children)
            
            # 检查是否找到解
            best = max(children, key=lambda c: c.value)
            if best.is_terminal and best.is_success:
                return self._extract_path(best)
        
        # 返回最优路径
        return self._extract_best_path(root)
    
    def _expand(self, node):
        """生成多个可能的下一步行动"""
        prompt = f"""当前状态：{node.state}
任务：{node.task}
历史：{node.get_path()}

生成{self.n_expansions}个不同的下一步行动方案：
"""
        responses = self.llm.generate_multiple(prompt, n=self.n_expansions)
        return [StateNode(
            task=node.task,
            state=self._execute(r),
            parent=node,
            action=r
        ) for r in responses]
    
    def _evaluate(self, node):
        """用LLM评估节点价值"""
        prompt = f"""评估当前状态距离完成任务的距离：
任务：{node.task}
当前状态：{node.state}
已完成步骤：{node.get_path()}

评分（0-10）："""
        return float(self.llm.generate(prompt))

class PlanExecuteAgent:
    def __init__(self, planner, executor):
        self.planner = planner
        self.executor = executor
    
    def run(self, task):
        # 1. 生成初始计划
        plan = self.planner.generate_plan(task)
        
        while not plan.is_complete():
            # 2. 执行下一步
            step = plan.current_step()
            result = self.executor.execute(step)
            
            # 3. 评估是否需要重新规划
            if self._needs_replan(task, plan, result):
                plan = self.planner.replan(task, plan, result)
            else:
                plan.advance()
        
        return plan.get_final_result()

def needs_replan(self, task, plan, result):
    # 检测执行结果是否与预期不符
    if not result.success:
        return True
    if result.side_effects:
        return True
    # 检测剩余步骤是否仍然可行
    remaining = plan.remaining_steps()
    feasibility = self.llm.evaluate(
        f"给定当前状态{result.state}，以下步骤是否可行：{remaining}"
    )
    return not feasibility

class PlanningEvaluator:
    def evaluate(self, task, plan, execution_result):
        return {
            "completeness": self._completeness(task, plan),
            # 计划是否覆盖了任务的所有方面
            "correctness": self._correctness(plan),
            # 计划的逻辑是否正确
            "efficiency": self._efficiency(plan, execution_result),
            # 是否存在冗余或可合并的步骤
            "adaptability": self._adaptability(plan, execution_result),
            # 遇到意外时的调整能力
            "granularity": self._granularity(task, plan),
            # 步骤粒度是否合适——太粗无法执行，太细浪费时间
        }

任务步骤 < 5？
  → ReAct（简单高效）

任务步骤 5-15？
  → Plan-and-Execute（先规划再执行）

任务步骤 > 15 或有多条可能路径？
  → LATS（树搜索）

任务需要多轮尝试？
  → Reflexion（从失败中学习）