Skip to content
🔗 分享本题
查看我的学习进度 →

Agent 将目标拆成带依赖的可执行 DAG,并根据进展检查结果动态重规划

🧠 记忆锚点:先拆可执行步骤,标依赖;结果偏离就重规划。

💡 答案要点

规划 = 把大任务分解成可执行的小步骤

规划方法对比

方法原理优点缺点适用
Chain of Thought逐步推理简单直观不可回退简单任务
Tree of Thoughts树状搜索可探索多路径计算量大需要试错
Plan-and-Execute先整体规划再执行结构清晰计划可能过时明确任务
ReWOO预规划+批量执行高效并行灵活性差工具调用多

方案1: Plan-and-Execute (推荐⭐)

流程:

Step 1: Planning - 制定完整计划
Step 2: Execution - 逐步执行
Step 3: Replanning - 根据结果调整计划

实现:

展开 Python 代码示例(110 行)
python
class PlanAndExecuteAgent:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools

    def run(self, goal):
        # Step 1: 制定计划
        plan = self.make_plan(goal)
        print(f"计划: {plan}")

        # Step 2: 执行每个步骤
        results = []
        for step in plan:
            result = self.execute_step(step, results)
            results.append(result)

            # Step 3: 检查是否需要重新规划
            if self.should_replan(step, result):
                plan = self.replan(goal, results)
                print(f"重新规划: {plan}")

        # Step 4: 综合结果
        final_answer = self.synthesize(goal, results)
        return final_answer

    def make_plan(self, goal):
        """制定计划"""
        prompt = f"""
        任务: {goal}

        请制定详细的执行计划,每个步骤要具体可执行。

        输出格式(JSON):
        [
          {{"step": 1, "action": "搜索最新的AI新闻", "tool": "search"}},
          {{"step": 2, "action": "总结新闻要点", "tool": "llm"}},
          {{"step": 3, "action": "生成周报", "tool": "llm"}}
        ]
        """

        plan_json = self.llm.generate(prompt)
        return json.loads(plan_json)

    def execute_step(self, step, previous_results):
        """执行单个步骤"""
        tool_name = step["tool"]
        action = step["action"]

        # 构造上下文(之前步骤的结果)
        context = "\n".join([
            f"步骤{i+1}结果: {r}"
            for i, r in enumerate(previous_results)
        ])

        # 执行工具
        if tool_name == "search":
            result = self.tools["search"].run(action)
        elif tool_name == "llm":
            result = self.llm.generate(f"{context}\n\n{action}")
        else:
            result = self.tools[tool_name].run(action)

        print(f"步骤{step['step']}: {action}{result[:100]}...")
        return result

    def should_replan(self, step, result):
        """判断是否需要重新规划"""
        # 检查执行失败
        if "错误" in result or "失败" in result:
            return True

        # 让LLM判断
        prompt = f"""
        步骤: {step['action']}
        结果: {result}

        这个结果是否符合预期? 是否需要调整后续计划?
        回答: 是/否
        """
        decision = self.llm.generate(prompt).strip()
        return decision == "是"

    def replan(self, goal, results):
        """重新规划"""
        context = "\n".join([f"已完成{i+1}: {r[:50]}..." for i, r in enumerate(results)])

        prompt = f"""
        原始任务: {goal}
        已完成步骤:
        {context}

        请根据当前进展,重新规划剩余步骤。
        """
        new_plan = self.llm.generate(prompt)
        return json.loads(new_plan)

# 使用示例
agent = PlanAndExecuteAgent(llm, tools)

result = agent.run("写一份本周AI行业的周报")

# 输出:
# 计划: [
#   {"step": 1, "action": "搜索本周AI新闻", "tool": "search"},
#   {"step": 2, "action": "总结新闻", "tool": "llm"},
#   {"step": 3, "action": "撰写周报", "tool": "llm"}
# ]
# 步骤1: 搜索本周AI新闻 → 找到10篇新闻...
# 步骤2: 总结新闻 → OpenAI发布GPT-5, Google推出Gemini Ultra...
# 步骤3: 撰写周报 → 本周AI行业动态:...

方案2: Tree of Thoughts (思维树)

适用: 需要探索多种可能性的任务(如写作、创意)

展开 Python 代码示例(78 行)
python
class TreeOfThoughts:
    def __init__(self, llm, depth=3, breadth=3):
        self.llm = llm
        self.depth = depth  # 思考深度
        self.breadth = breadth  # 每层生成几个候选

    def solve(self, problem):
        # 根节点
        root = TreeNode(problem, score=0)

        # 逐层扩展
        for level in range(self.depth):
            # 对当前层每个节点
            for node in self.get_layer_nodes(root, level):
                # 生成多个候选下一步
                candidates = self.generate_candidates(node, self.breadth)

                # 评估每个候选
                for candidate in candidates:
                    score = self.evaluate(candidate)
                    child = TreeNode(candidate, score=score)
                    node.add_child(child)

        # 找到最优路径
        best_path = self.find_best_path(root)
        return best_path

    def generate_candidates(self, node, k):
        """生成k个候选思路"""
        prompt = f"""
        当前思路: {node.content}

        请生成{k}个不同的后续思路。
        """
        responses = []
        for _ in range(k):
            response = self.llm.generate(prompt, temperature=0.9)
            responses.append(response)

        return responses

    def evaluate(self, thought):
        """评估思路质量"""
        prompt = f"""
        评估以下思路的质量(0-10分):
        {thought}

        评分:
        """
        score = float(self.llm.generate(prompt).strip())
        return score

    def find_best_path(self, root):
        """找到最高分路径"""
        def dfs(node, path, score):
            if not node.children:
                return path, score

            best = (path, score)
            for child in node.children:
                candidate_path, candidate_score = dfs(
                    child,
                    path + [child.content],
                    score + child.score
                )
                if candidate_score > best[1]:
                    best = (candidate_path, candidate_score)

            return best

        path, score = dfs(root, [root.content], root.score)
        return path

# 使用
tot = TreeOfThoughts(llm, depth=3, breadth=3)
best_solution = tot.solve("写一篇关于AI伦理的文章")

# 会探索 3^3=27 种可能路径,选最优

方案3: ReWOO (预规划+批量执行)

优势: 一次性规划所有工具调用,批量并行执行

展开 Python 代码示例(70 行)
python
class ReWOO:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools

    def run(self, task):
        # Step 1: 一次性规划所有步骤
        plan = self.plan_all_steps(task)

        # Step 2: 识别可并行的步骤
        parallel_groups = self.identify_parallel_groups(plan)

        # Step 3: 批量执行
        results = {}
        for group in parallel_groups:
            # 并行执行同组步骤
            group_results = self.execute_parallel(group)
            results.update(group_results)

        # Step 4: 综合结果
        return self.synthesize(task, results)

    def plan_all_steps(self, task):
        prompt = f"""
        任务: {task}

        请规划完整步骤,标注依赖关系:

        格式:
        #E1 = Search[最新AI新闻]
        #E2 = LLM[总结 #E1]
        #E3 = Search[AI政策]
        #E4 = LLM[综合 #E2 和 #E3]
        """
        plan = self.llm.generate(prompt)
        return self.parse_plan(plan)

    def identify_parallel_groups(self, plan):
        """识别可并行步骤"""
        # #E1 和 #E3 无依赖,可并行
        # #E2 依赖 #E1
        # #E4 依赖 #E2 和 #E3

        groups = [
            [plan["E1"], plan["E3"]],  # 第1组:并行
            [plan["E2"]],               # 第2组:等E1完成
            [plan["E4"]]                # 第3组:等E2,E3完成
        ]
        return groups

    def execute_parallel(self, steps):
        """并行执行步骤"""
        import concurrent.futures

        with concurrent.futures.ThreadPoolExecutor() as executor:
            futures = {
                executor.submit(self.execute_step, step): step
                for step in steps
            }

            results = {}
            for future in concurrent.futures.as_completed(futures):
                step = futures[future]
                results[step["id"]] = future.result()

        return results

# 效果:
# 传统ReAct: 4个步骤串行,耗时20秒
# ReWOO: 步骤1,3并行,耗时12秒 (节省40%)

面试话术:

"Agent规划我用Plan-and-Execute,先用LLM制定完整计划,再逐步执行并根据结果调整。复杂任务用Tree of Thoughts探索多路径,每层生成3个候选思路,评分选最优。工具调用多时用ReWOO预规划+批量并行,比ReAct快40%。关键是要能动态调整计划,而不是死板执行。"

📚 参考:CoALA:Cognitive Architectures for Language Agents(规划综述)