Plan-and-Execute Agent：从”知道”到”理解”#

1
"""Plan-and-Execute Agent — 先规划后执行。"""
2
from __future__ import annotations
3

4
from paicli.llm.client import LlmClient, Message
5
from paicli.llm import debug_logger
6
from paicli.plan.planner import Planner
7
from paicli.plan.task import Task, TaskStatus
8
from paicli.tool.registry import ToolRegistry
9

10
EXECUTION_PROMPT = """你是一个任务执行专家。请根据当前任务和上下文，选择合适的工具或生成回复。
11

12
当前任务类型：{type}
13
任务描述：{desc}
14

15
可用工具：
16
1. read_file - 读取文件内容，参数：{{"path": "文件路径"}}
17
2. write_file - 写入文件内容，参数：{{"path": "文件路径", "content": "内容"}}
18
3. execute_command - 执行命令，参数：{{"command": "命令"}}
19
4. create_project - 创建项目，参数：{{"name": "名称", "type": "java|python|node"}}
20

21
如果是ANALYSIS或VERIFICATION类型任务，请直接输出分析结果，不需要调用工具。
22

23
请用中文回复。"""
24

25
MAX_TASK_ITERATIONS = 5
26

27

28
class PlanExecuteAgent:
29
    """Plan-and-Execute Agent — 复杂任务分解后执行。"""
30

31
    def __init__(self, llm_client: LlmClient) -> None:
32
        self._llm = llm_client
33
        self._tools = ToolRegistry()
34
        self._planner = Planner(llm_client)
35

36
    def run(self, goal: str) -> str:
37
        """运行任务：规划 → 执行。"""
38
        try:
39
            return self._run_with_plan(goal)
40
        except Exception as e:
41
            debug_logger.generate_html_report()
42
            return f"❌ 执行失败: {e}"
43

44
    def _run_with_plan(self, goal: str) -> str:
45
        plan = self._planner.create_plan(goal)
46
        return self._execute_plan(goal, plan)
47

48
    def _execute_plan(self, goal: str, plan) -> str:
49
        print(plan.visualize())
50
        print("🚀 开始执行计划...\n")
51

52
        plan.mark_started()
53
        final_parts: list[str] = []
54

55
        for task_id in plan.execution_order:
56
            task = plan.get_task(task_id)
57
            if task is None:
58
                continue
59

60
            tasks_map = {t.id: t for t in plan.all_tasks}
61
            if not task.is_executable(tasks_map):
62
                print(f"⏭️ 跳过任务（依赖未完成）: {task_id}")
63
                task.mark_skipped()
64
                continue
65

66
            print(f"▶️ 执行任务: {task.description}")
67
            task.mark_started()
68

69
            try:
70
                result = self._execute_task(goal, plan, task)
71
                task.mark_completed(result)
72
                preview = result[:100] + ("..." if len(result) > 100 else "")
73
                print(f"✅ 完成: {preview}\n")
74

75
            except Exception as e:
76
                task.mark_failed(str(e))
77
                print(f"❌ 失败: {e}\n")
78

79
                if plan.progress < 0.5:
80
                    print("🔄 尝试重新规划...\n")
81
                    replanned = self._planner.replan(plan, str(e))
82
                    return self._execute_plan(goal, replanned)
83
                final_parts.append(f"任务 {task_id} 失败: {e}")
84

85
        if not final_parts:
86
            final_parts.append(self._build_final_result(plan))
87

88
        if plan.has_failed:
89
            plan.mark_failed()
90
            debug_logger.generate_html_report()
91
            return "⚠️ 计划部分完成，有任务失败。\n" + "\n".join(final_parts)
92

93
        plan.mark_completed()
94
        debug_logger.generate_html_report()
95
        return "✅ 计划执行完成！\n" + "\n".join(final_parts)
96

97
    def _execute_task(self, goal: str, plan, task: Task) -> str:
98
        """执行单个任务：多轮工具调用循环（类似 ReAct）。"""
99
        prompt = EXECUTION_PROMPT.format(type=task.type.value, desc=task.description)
100
        messages: list[Message] = [
101
            Message.system(prompt),
102
            Message.user(self._build_task_context(goal, plan, task)),
103
        ]
104

105
        all_results: list[str] = []
106

107
        for iteration in range(MAX_TASK_ITERATIONS):
108
            response = self._llm.chat(messages, tools=self._tools.get_tool_definitions())
109

110
            if not response.has_tool_calls():
111
                if all_results and not (response.content or "").strip():
112
                    return "\n".join(all_results).strip()
113
                return response.content or ""
114

115
            messages.append(Message.assistant(content=response.content, tool_calls=response.tool_calls))
116

117
            for tc in response.tool_calls:
118
                tool_name = tc.function.name
119
                print(f"   🔧 调用工具: {tool_name}")
120
                result = self._tools.execute_tool(tc.function.name, tc.function.arguments)
121
                debug_logger.log_tool_result(self._llm._call_id, tc.function.name, tc.function.arguments, result)
122
                all_results.append(result)
123
                messages.append(Message.tool_result(tc.id, result))
124

125
        return "\n".join(all_results).strip()
126

127
    @staticmethod
128
    def _build_task_context(goal: str, plan, task: Task) -> str:
129
        parts = [f"总目标：{goal}", f"当前任务：{task.description}"]
130
        if task.dependencies:
131
            parts.append("依赖任务结果：")
132
            for dep_id in task.dependencies:
133
                dep = plan.get_task(dep_id)
134
                if dep is None:
135
                    continue
136
                parts.append(f"- {dep.id} / {dep.description} / 状态={dep.status.value}")
137
                if dep.result:
138
                    parts.append(dep.result)
139
        else:
140
            parts.append("依赖任务：无")
141
        parts.append("请执行此任务。如果是ANALYSIS或VERIFICATION类型，请基于以上上下文直接给出结果。")
142
        return "\n".join(parts)
143

144
    @staticmethod
145
    def _build_final_result(plan) -> str:
146
        leaf_results = [
147
            t.result for t in plan.all_tasks
148
            if not t.dependents and t.result
149
        ]
150
        if leaf_results:
151
            return "\n".join(leaf_results)
152
        last = next((t.result for t in reversed(plan.all_tasks) if t.result), None)
153
        return last or ""

1
"""规划器 — 使用 LLM 将复杂任务分解为执行计划。"""
2
from __future__ import annotations
3

4
import json
5
import re
6
import time
7

8
from paicli.llm.client import LlmClient, Message
9
from paicli.plan.task import Task, TaskType
10
from paicli.plan.execution_plan import ExecutionPlan
11

12
PLANNING_PROMPT = """你是一个任务规划专家。请将用户的复杂任务分解为一系列可执行的子任务。
13

14
可用任务类型：
15
- FILE_READ: 读取文件内容
16
- FILE_WRITE: 写入文件内容
17
- COMMAND: 执行Shell命令
18
- ANALYSIS: 分析结果并做出决策
19
- VERIFICATION: 验证结果是否正确
20

21
请按以下JSON格式输出执行计划：
22
{
23
    "summary": "任务摘要",
24
    "tasks": [
25
        {
26
            "id": "task_1",
27
            "description": "任务描述",
28
            "type": "FILE_READ",
29
            "dependencies": []
30
        },
31
        {
32
            "id": "task_2",
33
            "description": "任务描述",
34
            "type": "FILE_WRITE",
35
            "dependencies": ["task_1"]
36
        }
37
    ]
38
}
39

40
规则：
41
1. 每个任务必须有唯一的id（如 task_1, task_2）
42
2. dependencies列出依赖的任务id
43
3. 任务应该按执行顺序排列
44
4. 任务描述要具体明确
45
5. 复杂任务拆分为5-10个子任务
46

47
只输出JSON，不要有其他内容。"""
48

49

50
class Planner:
51
    """使用 LLM 将复杂任务分解为 ExecutionPlan。"""
52

53
    def __init__(self, llm_client: LlmClient) -> None:
54
        self._llm = llm_client
55

56
    def create_plan(self, goal: str) -> ExecutionPlan:
57
        print(f"📋 正在规划任务: {goal}\n")
58
        messages = [
59
            Message.system(PLANNING_PROMPT),
60
            Message.user(f"请为以下任务制定执行计划：\n{goal}"),
61
        ]
62
        response = self._llm.chat(messages)
63
        return self._parse_plan(goal, response.content or "")
64

65
    def replan(self, failed_plan: ExecutionPlan, failure_reason: str) -> ExecutionPlan:
66
        print(f"🔄 重新规划，原因: {failure_reason}\n")
67
        context_parts = [
68
            f"原任务: {failed_plan.goal}",
69
            f"失败原因: {failure_reason}",
70
            "已完成的任务:",
71
        ]
72
        for task in failed_plan.all_tasks:
73
            if task.status == TaskStatus.COMPLETED:
74
                context_parts.append(f"- {task.id}: {task.description}")
75
        context_parts.append("\n请制定新的执行计划，避开之前的问题。")
76
        return self.create_plan("\n".join(context_parts))
77

78
    def _parse_plan(self, goal: str, plan_json: str) -> ExecutionPlan:
79
        cleaned = re.sub(r"```json\s*", "", plan_json)
80
        cleaned = re.sub(r"```\s*", "", cleaned).strip()
81

82
        data = json.loads(cleaned)
83
        summary = data.get("summary", "")
84
        tasks_data = data.get("tasks", [])
85

86
        plan = ExecutionPlan(f"plan_{int(time.time() * 1000)}", goal)
87
        plan.summary = summary
88

89
        # 第一遍：创建任务（不处理依赖，因为可能有前向引用）
90
        id_mapping: dict[str, str] = {}
91
        for i, task_node in enumerate(tasks_data, 1):
92
            original_id = task_node.get("id", f"task_{i}")
93
            new_id = f"task_{i}"
94
            id_mapping[original_id] = new_id
95

96
            description = task_node.get("description", "")
97
            type_str = task_node.get("type", "ANALYSIS")
98
            task_type = self._parse_task_type(type_str)
99

100
            plan.add_task(Task(id=new_id, description=description, type=task_type))
101

102
        # 第二遍：处理依赖关系
103
        for i, task_node in enumerate(tasks_data, 1):
104
            new_id = f"task_{i}"
105
            task = plan.get_task(new_id)
106
            if task is None:
107
                continue
108

109
            for dep_id in task_node.get("dependencies", []):
110
                mapped = id_mapping.get(dep_id, dep_id)
111
                if plan.get_task(mapped) is not None:
112
                    task.dependencies.append(mapped)
113
                    dep_task = plan.get_task(mapped)
114
                    if dep_task is not None and new_id not in dep_task.dependents:
115
                        dep_task.dependents.append(new_id)
116

117
        if not plan.compute_execution_order():
118
            raise ValueError("计划中存在循环依赖")
119

120
        return plan
121

122
    @staticmethod
123
    def _parse_task_type(type_str: str) -> TaskType:
124
        try:
125
            return TaskType(type_str.upper())
126
        except ValueError:
127
            return TaskType.ANALYSIS

1
"""任务节点 — 表示一个可执行的任务单元。"""
2
from __future__ import annotations
3

4
import time
5
from dataclasses import dataclass, field
6
from enum import Enum
7

8

9
class TaskType(Enum):
10
    PLANNING = "PLANNING"
11
    FILE_READ = "FILE_READ"
12
    FILE_WRITE = "FILE_WRITE"
13
    COMMAND = "COMMAND"
14
    ANALYSIS = "ANALYSIS"
15
    VERIFICATION = "VERIFICATION"
16

17

18
class TaskStatus(Enum):
19
    PENDING = "PENDING"
20
    RUNNING = "RUNNING"
21
    COMPLETED = "COMPLETED"
22
    FAILED = "FAILED"
23
    SKIPPED = "SKIPPED"
24

25

26
@dataclass
27
class Task:
28
    """一个可执行的任务单元，带依赖关系和状态追踪。"""
29
    id: str
30
    description: str
31
    type: TaskType
32
    status: TaskStatus = TaskStatus.PENDING
33
    result: str | None = None
34
    error: str | None = None
35
    dependencies: list[str] = field(default_factory=list)
36
    dependents: list[str] = field(default_factory=list)
37
    start_time: float = 0.0
38
    end_time: float = 0.0
39

40
    def mark_started(self) -> None:
41
        self.status = TaskStatus.RUNNING
42
        self.start_time = time.time()
43

44
    def mark_completed(self, result: str) -> None:
45
        self.status = TaskStatus.COMPLETED
46
        self.result = result
47
        self.end_time = time.time()
48

49
    def mark_failed(self, error: str) -> None:
50
        self.status = TaskStatus.FAILED
51
        self.error = error
52
        self.end_time = time.time()
53

54
    def mark_skipped(self) -> None:
55
        self.status = TaskStatus.SKIPPED
56
        self.end_time = time.time()
57

58
    @property
59
    def duration_ms(self) -> float:
60
        if self.start_time == 0:
61
            return 0.0
62
        end = self.end_time or time.time()
63
        return (end - self.start_time) * 1000
64

65
    def is_executable(self, all_tasks: dict[str, Task]) -> bool:
66
        if self.status != TaskStatus.PENDING:
67
            return False
68
        for dep_id in self.dependencies:
69
            dep = all_tasks.get(dep_id)
70
            if dep is None or dep.status != TaskStatus.COMPLETED:
71
                return False
72
        return True
73

74
    def __str__(self) -> str:
75
        return f"Task[{self.id}: {self.description}] ({self.status.value})"

1
"""执行计划 — 包含一组有依赖关系的任务，支持拓扑排序与可视化。"""
2
from __future__ import annotations
3

4
import time
5
from enum import Enum
6

7
from paicli.plan.task import Task, TaskStatus
8

9

10
class PlanStatus(Enum):
11
    CREATED = "CREATED"
12
    RUNNING = "RUNNING"
13
    COMPLETED = "COMPLETED"
14
    FAILED = "FAILED"
15
    CANCELLED = "CANCELLED"
16

17

18
class ExecutionPlan:
19
    """执行计划：管理一组带依赖关系的任务，拓扑排序决定执行顺序。"""
20

21
    def __init__(self, plan_id: str, goal: str) -> None:
22
        self.id = plan_id
23
        self.goal = goal
24
        self.summary: str = ""
25
        self.status = PlanStatus.CREATED
26
        self._tasks: dict[str, Task] = {}
27
        self._execution_order: list[str] = []
28
        self.start_time: float = 0.0
29
        self.end_time: float = 0.0
30

31
    def add_task(self, task: Task) -> None:
32
        self._tasks[task.id] = task
33
        for dep_id in task.dependencies:
34
            dep = self._tasks.get(dep_id)
35
            if dep is not None and task.id not in dep.dependents:
36
                dep.dependents.append(task.id)
37

38
    def get_task(self, task_id: str) -> Task | None:
39
        return self._tasks.get(task_id)
40

41
    @property
42
    def all_tasks(self) -> list[Task]:
43
        return list(self._tasks.values())
44

45
    @property
46
    def root_tasks(self) -> list[Task]:
47
        return [t for t in self._tasks.values() if not t.dependencies]
48

49
    @property
50
    def executable_tasks(self) -> list[Task]:
51
        tasks_map = {t.id: t for t in self._tasks.values()}
52
        return [t for t in self._tasks.values() if t.is_executable(tasks_map)]
53

54
    def compute_execution_order(self) -> bool:
55
        self._execution_order.clear()
56
        visited: set[str] = set()
57
        visiting: set[str] = set()
58

59
        for task in self._tasks.values():
60
            if task.id not in visited:
61
                if not self._topological_sort(task, visited, visiting):
62
                    return False
63
        return True
64

65
    def _topological_sort(self, task: Task, visited: set[str], visiting: set[str]) -> bool:
66
        if task.id in visiting:
67
            return False
68
        if task.id in visited:
69
            return True
70

71
        visiting.add(task.id)
72
        for dep_id in task.dependencies:
73
            dep = self._tasks.get(dep_id)
74
            if dep is not None:
75
                if not self._topological_sort(dep, visited, visiting):
76
                    return False
77

78
        visiting.discard(task.id)
79
        visited.add(task.id)
80
        self._execution_order.append(task.id)
81
        return True
82

83
    @property
84
    def execution_order(self) -> list[str]:
85
        if not self._execution_order:
86
            self.compute_execution_order()
87
        return list(self._execution_order)
88

89
    @property
90
    def progress(self) -> float:
91
        if not self._tasks:
92
            return 1.0
93
        completed = sum(1 for t in self._tasks.values() if t.status == TaskStatus.COMPLETED)
94
        return completed / len(self._tasks)
95

96
    @property
97
    def is_all_completed(self) -> bool:
98
        return all(t.status == TaskStatus.COMPLETED for t in self._tasks.values())
99

100
    @property
101
    def has_failed(self) -> bool:
102
        return any(t.status == TaskStatus.FAILED for t in self._tasks.values())
103

104
    def mark_started(self) -> None:
105
        self.status = PlanStatus.RUNNING
106
        self.start_time = time.time()
107

108
    def mark_completed(self) -> None:
109
        self.status = PlanStatus.COMPLETED
110
        self.end_time = time.time()
111

112
    def mark_failed(self) -> None:
113
        self.status = PlanStatus.FAILED
114
        self.end_time = time.time()
115

116
    _STATUS_ICONS = {
117
        TaskStatus.PENDING: "⏳",
118
        TaskStatus.RUNNING: "▶️",
119
        TaskStatus.COMPLETED: "✅",
120
        TaskStatus.FAILED: "❌",
121
        TaskStatus.SKIPPED: "⏭️",
122
    }
123

124
    def visualize(self) -> str:
125
        lines: list[str] = []
126
        lines.append("╔══════════════════════════════════════════════════════════╗")
127
        goal_display = self.goal if len(self.goal) <= 46 else self.goal[:43] + "..."
128
        lines.append(f"║  执行计划: {goal_display:<46}║")
129
        lines.append("╠══════════════════════════════════════════════════════════╣")
130

131
        for i, task_id in enumerate(self.execution_order, 1):
132
            task = self._tasks[task_id]
133
            icon = self._STATUS_ICONS.get(task.status, "?")
134
            deps = "无" if not task.dependencies else ",".join(task.dependencies)
135
            desc = task.description if len(task.description) <= 50 else task.description[:47] + "..."
136
            lines.append(f"║  {i}. {icon} {task.id:<20} [{task.type.value:<10}] 依赖: {deps:<10}║")
137
            lines.append(f"║     {desc}")
138

139
        lines.append("╚══════════════════════════════════════════════════════════╝")
140
        lines.append(f"   进度: {self.progress * 100:.0f}% | 状态: {self.status.value}")
141
        return "\n".join(lines)
142

143
    def __str__(self) -> str:
144
        return f"ExecutionPlan[{self.id}: {self.goal}] ({len(self._tasks)} tasks, {self.status.value})"