Plan-and-Execute Agent：先规划后执行#

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
        """所有依赖都已完成才能执行。"""
67
        if self.status != TaskStatus.PENDING:
68
            return False
69
        for dep_id in self.dependencies:
70
            dep = all_tasks.get(dep_id)
71
            if dep is None or dep.status != TaskStatus.COMPLETED:
72
                return False
73
        return True
74

75
    def __str__(self) -> str:
76
        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
        """计算拓扑排序。返回 False 表示有环。"""
56
        self._execution_order.clear()
57
        visited: set[str] = set()
58
        visiting: set[str] = set()
59

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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
        """为复杂任务创建执行计划。"""
58
        print(f"📋 正在规划任务: {goal}\n")
59

60
        messages = [
61
            Message.system(PLANNING_PROMPT),
62
            Message.user(f"请为以下任务制定执行计划：\n{goal}"),
63
        ]
64

65
        response = self._llm.chat(messages)
66
        return self._parse_plan(goal, response.content or "")
67

68
    def replan(self, failed_plan: ExecutionPlan, failure_reason: str) -> ExecutionPlan:
69
        """根据执行结果重新规划。"""
70
        print(f"🔄 重新规划，原因: {failure_reason}\n")
71

72
        context_parts = [
73
            f"原任务: {failed_plan.goal}",
74
            f"失败原因: {failure_reason}",
75
            "已完成的任务:",
76
        ]
77
        for task in failed_plan.all_tasks:
78
            if task.status == TaskStatus.COMPLETED:
79
                context_parts.append(f"- {task.id}: {task.description}")
80
        context_parts.append("\n请制定新的执行计划，避开之前的问题。")
81

82
        return self.create_plan("\n".join(context_parts))
83

84
    def _parse_plan(self, goal: str, plan_json: str) -> ExecutionPlan:
85
        """解析 LLM 生成的计划 JSON。"""
86
        cleaned = re.sub(r"```json\s*", "", plan_json)
87
        cleaned = re.sub(r"```\s*", "", cleaned).strip()
88

89
        data = json.loads(cleaned)
90
        summary = data.get("summary", "")
91
        tasks_data = data.get("tasks", [])
92

93
        plan = ExecutionPlan(f"plan_{int(time.time() * 1000)}", goal)
94
        plan.summary = summary
95

96
        # 第一遍：创建任务（不处理依赖，因为可能有前向引用）
97
        id_mapping: dict[str, str] = {}
98
        for i, task_node in enumerate(tasks_data, 1):
99
            original_id = task_node.get("id", f"task_{i}")
100
            new_id = f"task_{i}"
101
            id_mapping[original_id] = new_id
102

103
            description = task_node.get("description", "")
104
            type_str = task_node.get("type", "ANALYSIS")
105
            task_type = self._parse_task_type(type_str)
106

107
            plan.add_task(Task(id=new_id, description=description, type=task_type))
108

109
        # 第二遍：处理依赖关系
110
        for i, task_node in enumerate(tasks_data, 1):
111
            new_id = f"task_{i}"
112
            task = plan.get_task(new_id)
113
            if task is None:
114
                continue
115

116
            for dep_id in task_node.get("dependencies", []):
117
                mapped = id_mapping.get(dep_id, dep_id)
118
                if plan.get_task(mapped) is not None:
119
                    task.dependencies.append(mapped)
120
                    dep_task = plan.get_task(mapped)
121
                    if dep_task is not None and new_id not in dep_task.dependents:
122
                        dep_task.dependents.append(new_id)
123

124
        if not plan.compute_execution_order():
125
            raise ValueError("计划中存在循环依赖")
126

127
        return plan
128

129
    @staticmethod
130
    def _parse_task_type(type_str: str) -> TaskType:
131
        try:
132
            return TaskType(type_str.upper())
133
        except ValueError:
134
            return TaskType.ANALYSIS

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
_ACTION_KEYWORDS = ("创建", "写", "读", "执行", "编译", "运行", "修改", "删除", "然后", "接着", "再", "最后")
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, user_input: str) -> str:
37
        """运行任务，自动判断走规划还是直接执行。"""
38
        try:
39
            if self._should_plan(user_input):
40
                return self._run_with_plan(user_input)
41
            return self._run_simple(user_input)
42
        except Exception as e:
43
            debug_logger.generate_html_report()
44
            return f"❌ 执行失败: {e}"
45

46
    @staticmethod
47
    def _should_plan(input_text: str) -> bool:
48
        action_count = sum(1 for kw in _ACTION_KEYWORDS if kw in input_text)
49
        return action_count >= 3 or len(input_text) > 50
50

51
    def _run_with_plan(self, goal: str) -> str:
52
        plan = self._planner.create_plan(goal)
53
        return self._execute_plan(goal, plan)
54

55
    def _execute_plan(self, goal: str, plan) -> str:
56
        print(plan.visualize())
57
        print("🚀 开始执行计划...\n")
58

59
        plan.mark_started()
60
        final_parts: list[str] = []
61

62
        for task_id in plan.execution_order:
63
            task = plan.get_task(task_id)
64
            if task is None:
65
                continue
66

67
            tasks_map = {t.id: t for t in plan.all_tasks}
68
            if not task.is_executable(tasks_map):
69
                print(f"⏭️ 跳过任务（依赖未完成）: {task_id}")
70
                task.mark_skipped()
71
                continue
72

73
            print(f"▶️ 执行任务: {task.description}")
74
            task.mark_started()
75

76
            try:
77
                result = self._execute_task(goal, plan, task)
78
                task.mark_completed(result)
79
                preview = result[:100] + ("..." if len(result) > 100 else "")
80
                print(f"✅ 完成: {preview}\n")
81

82
            except Exception as e:
83
                task.mark_failed(str(e))
84
                print(f"❌ 失败: {e}\n")
85

86
                if plan.progress < 0.5:
87
                    print("🔄 尝试重新规划...\n")
88
                    replanned = self._planner.replan(plan, str(e))
89
                    return self._execute_plan(goal, replanned)
90
                final_parts.append(f"任务 {task_id} 失败: {e}")
91

92
        if not final_parts:
93
            final_parts.append(self._build_final_result(plan))
94

95
        if plan.has_failed:
96
            plan.mark_failed()
97
            debug_logger.generate_html_report()
98
            return "⚠️ 计划部分完成，有任务失败。\n" + "\n".join(final_parts)
99

100
        plan.mark_completed()
101
        debug_logger.generate_html_report()
102
        return "✅ 计划执行完成！\n" + "\n".join(final_parts)
103

104
    def _execute_task(self, goal: str, plan, task: Task) -> str:
105
        """执行单个任务：调 LLM → 如有工具调用则执行。"""
106
        prompt = EXECUTION_PROMPT.format(type=task.type.value, desc=task.description)
107
        messages = [
108
            Message.system(prompt),
109
            Message.user(self._build_task_context(goal, plan, task)),
110
        ]
111

112
        response = self._llm.chat(messages, tools=self._tools.get_tool_definitions())
113

114
        if response.has_tool_calls():
115
            results: list[str] = []
116
            for tc in response.tool_calls:
117
                print(f"   🔧 调用工具: {tc.function.name}")
118
                result = self._tools.execute_tool(tc.function.name, tc.function.arguments)
119
                debug_logger.log_tool_result(self._llm._call_id, tc.function.name, tc.function.arguments, result)
120
                results.append(result)
121
            return "\n".join(results)
122

123
        return response.content or ""
124

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

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

153
    def _run_simple(self, user_input: str) -> str:
154
        print("💡 简单任务，直接执行...\n")
155

156
        messages = [
157
            Message.system("你是一个智能编程助手，可以调用工具完成任务。"),
158
            Message.user(user_input),
159
        ]
160
        response = self._llm.chat(messages, tools=self._tools.get_tool_definitions())
161

162
        if response.has_tool_calls():
163
            results: list[str] = []
164
            for tc in response.tool_calls:
165
                result = self._tools.execute_tool(tc.function.name, tc.function.arguments)
166
                debug_logger.log_tool_result(self._llm._call_id, tc.function.name, tc.function.arguments, result)
167
                results.append(result)
168
            debug_logger.generate_html_report()
169
            return "\n".join(results)
170

171
        debug_logger.generate_html_report()
172
        return response.content or ""