TL;DR

• 系列方向： 从零搭建 AI Test Agent，逐步实现大模型接入、工具调用、任务循环、任务规划和状态管理，最终形成一个能够辅助软件测试工作的 Agent 项目。
• 本篇重点： 第四篇实现了 Agent Loop，让 Agent 能够连续调用测试工具。但面对复杂测试任务，仅仅依靠循环还不够。本篇增加任务规划、任务依赖和执行状态管理，让 Agent 按照明确的计划完成测试工作。

---

一、Agent 能连续调用工具，就算具备任务规划能力了吗？

先回顾一下前面的项目。

到第四篇为止，我们已经实现了：

• Agent 基础架构。
• 大模型接入。
• Tool Calling。
• Agent Loop 多轮执行。

现在可以让 Agent 执行这样的任务：

检查订单 ORDER1002 是否存在异常：

1. 查询订单详情。
2. 检查订单金额。
3. 检查订单状态。
4. 生成测试报告。

Agent 可以根据模型返回的工具调用请求，逐步执行这些操作。

但如果测试任务再复杂一点呢？

例如：

请检查用户 USER001 的订单情况：

1. 查询用户信息。
2. 确认用户是否有效。
3. 查询用户订单。
4. 检查订单金额。
5. 检查订单状态。
6. 汇总异常订单。
7. 生成测试报告。

这里就出现了几个问题：

• 用户信息查询失败，后续订单查询是否还要继续？
• 哪些任务必须等待前面的任务完成？
• 某一步执行失败，后续任务应该继续还是跳过？
• Agent 如何知道哪些任务已经完成？
• 如果程序中途异常，能否知道任务执行到了哪一步？

这些问题已经不只是 Tool Calling 或 Agent Loop 能够解决的了。

我们需要给 Agent 增加一个独立的任务管理层。

我的理解是：

Agent Loop 负责持续执行，任务规划负责决定做什么，状态管理负责记录实际执行到了哪里。

这三部分应该分开设计。

二、本篇准备实现什么？

这次仍然沿用订单检查案例，不重新搭建项目。

本篇增加以下能力：

功能	具体作用
任务拆解	将用户需求转换为多个独立任务
任务计划	记录每个任务需要执行什么
依赖管理	控制任务之间的执行顺序
状态管理	记录待执行、执行中、完成、失败和阻塞
失败处理	依赖任务失败时，阻止后续任务执行
结果汇总	根据真实工具结果生成测试报告

本篇先不引入复杂的多 Agent 协作，也不使用重量级工作流框架。

我们先用 Python 实现一个简单的任务调度器，看看任务规划真正落地时需要处理哪些问题。

三、先设计任务执行模型

在写代码之前，需要先确定任务的数据结构。

一个任务至少应该包含：

{
  "task_id": "task_001",
  "description": "查询订单详情",
  "tool_name": "query_order_detail",
  "arguments": {
    "order_id": "ORDER1002"
  },
  "dependencies": [],
  "status": "pending",
  "result": null,
  "error": null
}

几个字段需要特别说明：

• task_id：任务唯一标识。
• description：任务说明。
• tool_name：需要调用的工具。
• arguments：工具执行参数。
• dependencies：当前任务依赖哪些任务。
• status：任务执行状态。
• result：实际执行结果。
• error：执行失败原因。

3.1 定义任务状态

本篇使用五种状态：

状态	含义
pending	等待执行
running	正在执行
completed	工具执行完成
failed	工具执行失败
blocked	因依赖任务失败而无法执行

这里有一个容易混淆的地方：

任务执行完成，不代表业务检查通过。

例如：

{
  "success": true,
  "passed": false,
  "message": "订单金额异常：-20.0"
}

这表示金额检查工具成功运行了，但检查结果不通过。

因此任务状态应该是 completed，而不是 failed。

如果把工具执行失败和业务断言失败混在一起，后面统计任务状态就会出现问题。

3.2 任务之间的依赖关系

例如订单检查任务：

查询订单详情
      |
      +----------+
      |          |
   检查金额    检查状态
      |          |
      +----------+
           |
       汇总报告

金额检查和状态检查都依赖订单查询。

如果订单查询失败，那么金额检查和状态检查就不应该继续执行。

这就是任务依赖关系存在的意义。

四、项目结构调整

继续使用前面的 ai-test-agent 项目。

ai-test-agent/
│
├── main.py
├── agent.py
├── llm_client.py
├── tools.py
├── task_manager.py
├── requirements.txt
└── README.md

本篇新增：

task_manager.py

负责：

• 保存任务计划。
• 检查任务依赖。
• 更新任务状态。
• 判断任务是否可以执行。
• 处理失败任务和阻塞任务。

这样设计以后，Agent 不需要直接管理所有任务状态，任务管理器负责执行层面的约束。

五、第一步：让大模型生成任务计划

首先需要让模型把自然语言需求转换成结构化任务。

例如用户输入：

检查订单 ORDER1002 是否存在异常，
需要查询订单详情、检查金额、检查状态，
最后生成测试报告。

模型应该返回类似下面的 JSON：

{
  "tasks": [
    {
      "task_id": "task_001",
      "description": "查询订单详情",
      "tool_name": "query_order_detail",
      "arguments": {
        "order_id": "ORDER1002"
      },
      "dependencies": []
    },
    {
      "task_id": "task_002",
      "description": "检查订单金额",
      "tool_name": "check_order_amount",
      "arguments": {
        "order_id": "ORDER1002"
      },
      "dependencies": ["task_001"]
    },
    {
      "task_id": "task_003",
      "description": "检查订单状态",
      "tool_name": "check_order_status",
      "arguments": {
        "order_id": "ORDER1002"
      },
      "dependencies": ["task_001"]
    }
  ]
}

注意，这里没有让模型直接执行工具。

模型只负责提出任务计划，真正的执行仍然交给 Python。

这样做有一个很重要的好处：

模型可以负责灵活拆解任务，但任务是否允许执行，最终由程序决定。

5.1 编写任务规划代码

在 agent.py 中增加任务规划逻辑。

import json
import re

from llm_client import LLMClient


PLANNER_PROMPT = """
你是一个软件测试任务规划助手。

请将用户的测试需求拆分为可执行的测试任务。

目前允许使用的工具：

1. query_order_detail
   参数：order_id

2. check_order_amount
   参数：order_id

3. check_order_status
   参数：order_id

规划要求：

1. 每个任务必须有唯一 task_id。
2. 每个任务必须指定 description。
3. 每个任务必须指定 tool_name。
4. arguments 必须是 JSON 对象。
5. dependencies 必须是 task_id 数组。
6. 检查金额和状态之前，必须先完成订单查询。
7. 不要添加不存在的工具。
8. 只返回 JSON，不要添加 Markdown 代码块。

返回格式：

{
  "tasks": [
    {
      "task_id": "task_001",
      "description": "任务描述",
      "tool_name": "工具名称",
      "arguments": {
        "order_id": "订单编号"
      },
      "dependencies": []
    }
  ]
}
"""


def parse_plan(content):
    """解析模型生成的任务计划"""

    content = content.strip()

    # 兼容模型偶尔返回 Markdown JSON 代码块的情况
    content = re.sub(
        r"^```(?:json)?\s*|\s*```$",
        "",
        content
    )

    return json.loads(content)

这里使用正则表达式处理可能出现的 Markdown 代码块。

但需要注意：这并不意味着可以直接信任模型返回的数据。

下一步必须进行严格校验。

六、第二步：校验任务计划

模型生成的任务计划可能存在问题，例如：

• 使用了不存在的工具。
• 工具参数不正确。
• 任务 ID 重复。
• 依赖了不存在的任务。
• 任务之间形成循环依赖。

如果不做校验就直接执行，Agent 很容易出现不可控行为。

因此，我们增加一个计划校验模块。

6.1 task_manager.py

# task_manager.py

import json

from tools import execute_tool


ALLOWED_TOOLS = {
    "query_order_detail",
    "check_order_amount",
    "check_order_status"
}


VALID_STATUSES = {
    "pending",
    "running",
    "completed",
    "failed",
    "blocked"
}


class TaskManager:

    def __init__(self, tasks):

        if not isinstance(tasks, list):
            raise ValueError("tasks 必须是数组")

        if not tasks:
            raise ValueError("任务计划不能为空")

        if len(tasks) > 20:
            raise ValueError("单次任务数量不能超过 20")

        self.tasks = {}

        for item in tasks:

            task_id = item.get("task_id")

            if not isinstance(task_id, str) or not task_id:
                raise ValueError("任务 ID 不合法")

            if task_id in self.tasks:
                raise ValueError(f"任务 ID 重复：{task_id}")

            tool_name = item.get("tool_name")

            if tool_name not in ALLOWED_TOOLS:
                raise ValueError(
                    f"不允许执行的工具：{tool_name}"
                )

            arguments = item.get("arguments")

            if not isinstance(arguments, dict):
                raise ValueError(
                    f"任务参数必须是对象：{task_id}"
                )

            if set(arguments) != {"order_id"}:
                raise ValueError(
                    f"任务参数不正确：{task_id}"
                )

            if not isinstance(arguments["order_id"], str):
                raise ValueError(
                    f"订单编号必须是字符串：{task_id}"
                )

            dependencies = item.get("dependencies", [])

            if not isinstance(dependencies, list):
                raise ValueError(
                    f"任务依赖必须是数组：{task_id}"
                )

            if not all(
                isinstance(dep, str)
                for dep in dependencies
            ):
                raise ValueError(
                    f"任务依赖 ID 必须是字符串：{task_id}"
                )

            if task_id in dependencies:
                raise ValueError(
                    f"任务不能依赖自身：{task_id}"
                )

            self.tasks[task_id] = {
                "task_id": task_id,
                "description": item.get(
                    "description",
                    ""
                ),
                "tool_name": tool_name,
                "arguments": arguments,
                "dependencies": dependencies,
                "status": "pending",
                "result": None,
                "error": None
            }

        self.validate_dependencies()
        self.validate_cycles()

    def validate_dependencies(self):
        """检查依赖任务是否存在"""

        for task in self.tasks.values():

            for dep in task["dependencies"]:

                if dep not in self.tasks:
                    raise ValueError(
                        f"不存在的依赖任务：{dep}"
                    )

    def validate_cycles(self):
        """检查是否存在循环依赖"""

        visited = set()
        visiting = set()

        def dfs(task_id):

            if task_id in visiting:
                raise ValueError("任务计划存在循环依赖")

            if task_id in visited:
                return

            visiting.add(task_id)

            task = self.tasks[task_id]

            for dep in task["dependencies"]:
                dfs(dep)

            visiting.remove(task_id)
            visited.add(task_id)

        for task_id in self.tasks:
            dfs(task_id)

    def get_ready_tasks(self):
        """获取依赖已经完成的任务"""

        ready = []

        for task in self.tasks.values():

            if task["status"] != "pending":
                continue

            dependencies = [
                self.tasks[dep]["status"]
                for dep in task["dependencies"]
            ]

            if all(
                status == "completed"
                for status in dependencies
            ):
                ready.append(task)

        return ready

    def update_blocked_tasks(self):
        """将依赖失败的任务标记为阻塞"""

        changed = True

        while changed:

            changed = False

            for task in self.tasks.values():

                if task["status"] != "pending":
                    continue

                dependency_statuses = [
                    self.tasks[dep]["status"]
                    for dep in task["dependencies"]
                ]

                if any(
                    status in {"failed", "blocked"}
                    for status in dependency_statuses
                ):
                    task["status"] = "blocked"
                    task["error"] = "依赖任务执行失败或被阻塞"
                    changed = True

    def execute_task(self, task):
        """执行单个任务"""

        task["status"] = "running"

        try:

            result = execute_tool(
                task["tool_name"],
                task["arguments"]
            )

            task["result"] = result

            if result.get("success") is True:
                task["status"] = "completed"
            else:
                task["status"] = "failed"
                task["error"] = result.get(
                    "message",
                    "工具执行失败"
                )

        except Exception as exc:

            task["status"] = "failed"
            task["error"] = str(exc)

    def run(self):
        """按依赖关系调度任务"""

        while True:

            self.update_blocked_tasks()

            ready_tasks = self.get_ready_tasks()

            if not ready_tasks:
                break

            # 每次取一个任务执行，便于控制和观察状态
            task = ready_tasks[0]

            print(
                f"\n执行任务：{task['task_id']} "
                f"- {task['description']}"
            )

            self.execute_task(task)

            print(
                f"任务状态：{task['status']}"
            )

        self.update_blocked_tasks()

        pending_tasks = [
            task
            for task in self.tasks.values()
            if task["status"] == "pending"
        ]

        if pending_tasks:
            raise RuntimeError(
                "存在无法调度的任务，请检查任务依赖"
            )

        return list(self.tasks.values())

    def get_summary(self):
        """汇总任务状态"""

        summary = {
            "pending": 0,
            "running": 0,
            "completed": 0,
            "failed": 0,
            "blocked": 0
        }

        for task in self.tasks.values():
            summary[task["status"]] += 1

        return summary

这部分代码是本篇的核心。

它做了几件事情：

1. 检查任务计划是否合法。
2. 检查依赖是否存在。
3. 检查是否存在循环依赖。
4. 只执行依赖已经完成的任务。
5. 依赖任务失败时，将后续任务标记为阻塞。
6. 保存每个任务的执行结果和状态。

这里有一个重要的设计选择：本篇采用顺序调度，一次执行一个已经满足依赖条件的任务。

虽然金额检查和状态检查在订单查询完成后可以并行，但我们先不引入并发执行，避免把线程安全、并发状态更新等问题混进第一版实现。

七、第三步：整合 Agent 和任务管理器

现在把任务规划和任务执行连接起来。

修改 agent.py：

# agent.py

import json
import re

from llm_client import LLMClient
from task_manager import TaskManager


PLANNER_PROMPT = """
你是一个软件测试任务规划助手。

请将用户需求转换为可执行的测试任务计划。

允许使用的工具：

1. query_order_detail
2. check_order_amount
3. check_order_status

规则：

1. 每个任务必须有唯一 task_id。
2. 每个任务必须指定 description。
3. 每个任务必须指定 tool_name。
4. arguments 必须是 JSON 对象，只允许包含 order_id。
5. dependencies 必须是任务 ID 数组。
6. 金额检查和状态检查必须依赖订单查询任务。
7. 不允许使用其他工具。
8. 只输出 JSON，不要添加 Markdown。

返回结构：

{
  "tasks": [
    {
      "task_id": "task_001",
      "description": "查询订单详情",
      "tool_name": "query_order_detail",
      "arguments": {
        "order_id": "ORDER1002"
      },
      "dependencies": []
    }
  ]
}
"""


class AITestAgent:

    def __init__(self):

        self.llm = LLMClient()

    def create_plan(self, task):

        messages = [
            {
                "role": "system",
                "content": PLANNER_PROMPT
            },
            {
                "role": "user",
                "content": task
            }
        ]

        response = self.llm.chat(messages)

        content = response.get("content") or ""

        content = content.strip()

        content = re.sub(
            r"^```(?:json)?\s*|\s*```$",
            "",
            content
        )

        plan = json.loads(content)

        if not isinstance(plan, dict):
            raise ValueError("模型返回的计划格式不正确")

        return plan

    def generate_report(self, tasks, summary):

        lines = [
            "订单测试执行报告",
            "",
            "一、任务执行情况",
            f"任务总数：{len(tasks)}",
            f"已完成：{summary['completed']}",
            f"执行失败：{summary['failed']}",
            f"被阻塞：{summary['blocked']}",
            "",
            "二、任务明细"
        ]

        for task in tasks:

            lines.append(
                f"\n任务：{task['description']}"
            )

            lines.append(
                f"状态：{task['status']}"
            )

            if task["result"] is not None:
                lines.append(
                    "结果：" + json.dumps(
                        task["result"],
                        ensure_ascii=False
                    )
                )

            if task["error"]:
                lines.append(
                    f"失败原因：{task['error']}"
                )

        # 业务检查是否全部通过，单独判断
        business_failures = []

        for task in tasks:

            result = task["result"]

            if (
                isinstance(result, dict)
                and result.get("passed") is False
            ):
                business_failures.append(
                    task["description"]
                )

        lines.extend([
            "",
            "三、测试结论"
        ])

        if summary["failed"] or summary["blocked"]:
            lines.append(
                "任务执行不完整，请检查失败或阻塞任务。"
            )
        elif business_failures:
            lines.append(
                "任务均已执行，但发现业务检查不通过："
                + "、".join(business_failures)
            )
        else:
            lines.append(
                "任务执行完成，业务检查通过。"
            )

        return "\n".join(lines)

    def run(self, user_task):

        print("========== 生成任务计划 ==========")

        plan = self.create_plan(user_task)

        tasks = plan.get("tasks")

        manager = TaskManager(tasks)

        print("\n========== 任务计划 ==========")

        print(
            json.dumps(
                tasks,
                ensure_ascii=False,
                indent=2
            )
        )

        print("\n========== 开始执行任务 ==========")

        execution_results = manager.run()

        summary = manager.get_summary()

        report = self.generate_report(
            execution_results,
            summary
        )

        return {
            "tasks": execution_results,
            "summary": summary,
            "report": report
        }

这里有一个值得注意的地方：

最终报告由 Python 根据任务执行记录生成，而不是让大模型重新编造一份测试结果。

后续如果需要更自然的报告，可以再让大模型润色，但订单金额、检查状态、失败原因等事实信息应该以工具返回结果为准。

八、启动程序进行验证

8.1 main.py

# main.py

from agent import AITestAgent


if __name__ == "__main__":

    agent = AITestAgent()

    task = """
请检查订单 ORDER1002 是否存在异常。

需要完成：
1. 查询订单详情。
2. 检查订单金额是否合法。
3. 检查订单状态是否正常。
4. 汇总检查结果并生成测试报告。
"""

    result = agent.run(task)

    print("\n========== 任务状态汇总 ==========")

    print(result["summary"])

    print("\n========== 最终测试报告 ==========")

    print(result["report"])

8.2 启动命令

模型环境变量沿用第四篇的配置：

python main.py

8.3 预期任务计划

模型可能生成以下任务计划：

任务 ID	任务内容	依赖任务
task_001	查询订单详情	无
task_002	检查订单金额	task_001
task_003	检查订单状态	task_001

注意，任务计划是模型生成的，实际任务顺序由任务管理器根据依赖关系决定，不应该依赖模型返回数组的排列顺序。

8.4 预期执行结果

订单 ORDER1002 的模拟数据为：

{
  "order_id": "ORDER1002",
  "amount": -20.00,
  "status": "unknown",
  "user_id": "USER002"
}

预期任务状态：

任务	状态	说明
查询订单详情	completed	订单存在，查询成功
检查订单金额	completed	工具执行成功，但业务检查不通过
检查订单状态	completed	工具执行成功，但业务检查不通过

最终状态汇总：

{
  "pending": 0,
  "running": 0,
  "completed": 3,
  "failed": 0,
  "blocked": 0
}

这里三个任务都是 completed，并不代表订单没有问题。

实际的业务检查结果是：

• 金额检查：不通过。
• 状态检查：不通过。

这也是为什么我们需要把任务执行状态和业务检查结果分开记录。

九、验证异常场景：订单不存在怎么办？

正常执行只是最基本的验证。

对于测试 Agent，我更关心的是异常情况下，它能不能正确处理任务。

把 main.py 中的订单编号改为：

task = """
请检查订单 ORDER9999 是否存在异常。

需要完成：
1. 查询订单详情。
2. 检查订单金额是否合法。
3. 检查订单状态是否正常。
4. 汇总检查结果并生成测试报告。
"""

根据当前模拟数据，ORDER9999 不存在。

预期执行过程：

========== 开始执行任务 ==========

执行任务：task_001 - 查询订单详情

任务状态：failed

========== 任务状态汇总 ==========

{
    "pending": 0,
    "running": 0,
    "completed": 0,
    "failed": 1,
    "blocked": 2
}

为什么金额检查和状态检查没有执行？

因为它们都依赖 task_001。

订单查询失败后，任务管理器会把后续任务标记为 blocked。

这就避免了在订单不存在的情况下，继续执行没有意义的业务检查。

需要说明的是，这里展示的是基于当前模拟工具和任务依赖设计的预期结果。接入真实模型后，仍然需要检查模型生成的计划是否符合预期，并通过实际运行验证。

十、从工程角度看，还有哪些问题需要解决？

本篇已经实现了任务拆解、依赖管理和状态记录，但离实际测试平台还有一定距离。

10.1 模型生成的计划不能完全信任

模型可能出现以下情况：

• 生成不存在的工具。
• 参数字段错误。
• 任务依赖不合理。
• 忘记添加必要的检查任务。
• 将不相关的任务设置成依赖关系。

因此，任务计划必须经过程序校验。

如果涉及删除数据、支付、发布等高风险操作，还应该增加人工审批和更严格的权限控制。

10.2 当前失败处理比较简单

本篇采用的策略是：

依赖任务失败，后续依赖任务直接阻塞。

这种方式适合当前的订单检查案例，但不是所有测试任务都应该采用相同策略。

例如：

• 登录失败，依赖登录状态的测试应该阻塞。
• 某个独立接口检查失败，其他独立接口仍然可以继续。
• 报告生成失败，不应该让已经完成的测试重新执行。

后续可以根据任务类型和业务规则配置不同的失败处理策略。

10.3 当前状态只保存在内存中

目前的任务状态保存在 Python 对象中。

程序结束后，执行记录就会丢失。

真实项目中可以进一步使用 SQLite 或 MySQL 保存：

• 任务 ID。
• 所属测试任务。
• 任务依赖。
• 执行状态。
• 开始时间和结束时间。
• 工具执行结果。
• 异常信息。

这样就能够支持任务恢复、历史查询和执行过程追踪。

10.4 还没有真正实现失败恢复

现在的逻辑是失败后标记为 failed，依赖任务标记为 blocked。

但没有实现自动重试，也没有实现从中断位置恢复。

如果后续增加重试功能，还需要考虑：

• 哪些工具允许重试？
• 最大重试次数是多少？
• 重试间隔如何设置？
• 工具是否具有副作用？
• 重试是否会造成重复数据？

尤其是有副作用的工具，不能简单地失败后重复执行。

这些问题都需要结合具体业务设计。

---

十一、本篇小结

回顾一下，本篇在第四篇 Agent Loop 的基础上，增加了一个独立的任务管理层。

目前项目已经具备以下能力：

篇章	实现内容
01	搭建 AI Test Agent 基础架构
02	接入大模型，理解测试任务
03	Tool Calling，动态调用测试工具
04	Agent Loop，多轮执行测试任务
05	任务规划、任务依赖和状态管理

我认为，这一步是 Agent 从简单 Demo 走向工程化的重要变化。

之前我们主要关注模型能不能调用工具，现在开始关注任务能不能被正确拆解、执行和追踪。

这两者有明显区别。

一个能够调用工具的 Agent，不一定能够稳定完成复杂任务；但一个具备任务计划和状态管理能力的 Agent，已经开始具备自动化工作流的基础。

写在最后：

这一篇我更想强调的，不是让大模型生成一份漂亮的任务 JSON，而是把模型的计划真正变成一套可以执行、可以校验、可以追踪的任务流程。

模型可以提出计划，但不能直接决定程序的执行权限；模型可以分析结果，但不能改变工具实际返回的测试事实。

对于软件测试 Agent 来说，这种边界比单纯增加模型能力更加重要。