4. 实现一个规划器Agent

任务目标

构建一个具备任务分解与逐步执行能力的规划器Agent。该Agent接收一个高层自然语言任务（如“组织一场技术分享会”），将其分解为有序的子任务列表，并模拟逐步执行每个子任务，最终汇总执行结果。核心价值在于演示“规划-执行”这一Agent基础范式，而非实现具体业务逻辑。

核心实现

import json
from typing import List, Dict, Callable

class PlannerAgent:
    def __init__(self, executor: Callable[[str], str]):
        """
        executor: 实际执行单个子任务的函数。
                  输入子任务描述，返回执行结果字符串。
        """
        self.executor = executor
        self.plan: List[Dict] = []
        self.results: List[Dict] = []

    def decompose(self, task: str) -> List[str]:
        """
        将复杂任务分解为有序子任务列表。
        此处使用基于规则的关键词匹配，演示分解逻辑。
        实际场景可替换为LLM调用。
        """
        task_map = {
            "会议": ["确定会议主题与目标", "邀请参会人员", "准备会议材料", "召开会议并记录纪要"],
            "报告": ["收集数据与资料", "撰写报告大纲", "填充报告内容", "校对并输出终稿"],
            "活动": ["确定活动预算与规模", "场地预订与物料准备", "宣传与报名", "现场执行与收尾"]
        }
        for key, subtasks in task_map.items():
            if key in task:
                return subtasks
        # 默认分解策略
        return [f"分析任务：{task}", "制定执行方案", "分步实施", "汇总并检查结果"]

    def run(self, task: str) -> Dict:
        """执行完整规划-执行流程"""
        self.plan = []
        self.results = []

        # 1. 规划阶段
        subtasks = self.decompose(task)
        for i, sub in enumerate(subtasks, 1):
            self.plan.append({"step": i, "task": sub})

        # 2. 执行阶段
        for item in self.plan:
            result = self.executor(item["task"])
            self.results.append({
                "step": item["step"],
                "task": item["task"],
                "status": "success" if result else "failed",
                "output": result
            })

        # 3. 汇总
        return {
            "original_task": task,
            "plan": self.plan,
            "execution": self.results,
            "summary": {
                "total": len(self.results),
                "succeeded": sum(1 for r in self.results if r["status"] == "success")
            }
        }

# 模拟执行器：实际场景中可调用API、执行代码或操作工具
def mock_executor(subtask: str) -> str:
    # 模拟不同子任务的执行耗时与输出
    return f"[完成] {subtask}"

# 使用示例
if __name__ == "__main__":
    agent = PlannerAgent(executor=mock_executor)
    result = agent.run("组织一场产品发布会活动")
    print(json.dumps(result, ensure_ascii=False, indent=2))

运行与验证

运行方式：直接执行 python planner_agent.py（需Python 3.8+，无第三方依赖）。

验证方法：

1. 观察输出JSON中 plan 数组是否包含4个有序子任务（因输入含“活动”关键词）；
2. 确认 execution 中每个子任务均有 status: "success" 且 output 非空；
3. 检查 summary.succeeded 等于 summary.total，值为4。

预期输出（节选）：

{
  "original_task": "组织一场产品发布会活动",
  "plan": [
    {"step": 1, "task": "确定活动预算与规模"},
    {"step": 2, "task": "场地预订与物料准备"},
    {"step": 3, "task": "宣传与报名"},
    {"step": 4, "task": "现场执行与收尾"}
  ],
  "summary": {"total": 4, "succeeded": 4}
}

扩展说明

本实现将规划器与执行器解耦：decompose 方法负责任务分解（当前为规则驱动），executor 函数负责具体执行。若要接入真实LLM，只需将 decompose 内部替换为一次LLM调用（如让模型返回JSON子任务列表），并将 mock_executor 替换为实际工具调用函数即可，无需改动Agent主流程。这种设计使规划策略与执行细节可独立演进，是生产级Agent的常见架构。